Revert "Revert "Fix cross-worker revalidate API" (#49138)" (#49141)

Revert #49138 and split the server-ipc file into smaller modules.

packages/next/src/server/api-utils/node.ts

@@ -27,7 +27,6 @@ import {
   SYMBOL_PREVIEW_DATA,
   RESPONSE_LIMIT_DEFAULT,
 } from './index'
-import { createRequestResponseMocks } from '../lib/mock-request'
 import { getTracer } from '../lib/trace/tracer'
 import { NodeSpan } from '../lib/trace/constants'
 import { RequestCookies } from '../web/spec-extension/cookies'
@@ -36,6 +35,7 @@ import {
   PRERENDER_REVALIDATE_HEADER,
   PRERENDER_REVALIDATE_ONLY_GENERATED_HEADER,
 } from '../../lib/constants'
+import { invokeRequest } from '../lib/server-ipc/invoke-request'
 
 export function tryGetPreviewData(
   req: IncomingMessage | BaseNextRequest | Request,
@@ -194,7 +194,14 @@ export async function parseBody(
 type ApiContext = __ApiPreviewProps & {
   trustHostHeader?: boolean
   allowedRevalidateHeaderKeys?: string[]
-  revalidate?: (_req: IncomingMessage, _res: ServerResponse) => Promise<any>
+  hostname?: string
+  revalidate?: (config: {
+    urlPath: string
+    revalidateHeaders: { [key: string]: string | string[] }
+    opts: { unstable_onlyGenerated?: boolean }
+  }) => Promise<any>
+
+  // (_req: IncomingMessage, _res: ServerResponse) => Promise<any>
 }
 
 function getMaxContentLength(responseLimit?: ResponseLimit) {
@@ -453,20 +460,44 @@ async function revalidate(
         throw new Error(`Invalid response ${res.status}`)
       }
     } else if (context.revalidate) {
-      const mocked = createRequestResponseMocks({
-        url: urlPath,
-        headers: revalidateHeaders,
-      })
+      // We prefer to use the IPC call if running under the workers mode.
+      const ipcPort = process.env.__NEXT_PRIVATE_ROUTER_IPC_PORT
+      if (ipcPort) {
+        const ipcKey = process.env.__NEXT_PRIVATE_ROUTER_IPC_KEY
+        const res = await invokeRequest(
+          `http://${
+            context.hostname
+          }:${ipcPort}?key=${ipcKey}&method=revalidate&args=${encodeURIComponent(
+            JSON.stringify([{ urlPath, revalidateHeaders }])
+          )}`,
+          {
+            method: 'GET',
+            headers: {},
+          }
+        )
 
-      await context.revalidate(mocked.req, mocked.res)
-      await mocked.res.hasStreamed
+        const chunks = []
 
-      if (
-        mocked.res.getHeader('x-nextjs-cache') !== 'REVALIDATED' &&
-        !(mocked.res.statusCode === 404 && opts.unstable_onlyGenerated)
-      ) {
-        throw new Error(`Invalid response ${mocked.res.statusCode}`)
+        for await (const chunk of res) {
+          if (chunk) {
+            chunks.push(chunk)
+          }
+        }
+        const body = Buffer.concat(chunks).toString()
+        const result = JSON.parse(body)
+
+        if (result.err) {
+          throw new Error(result.err.message)
+        }
+
+        return
       }
+
+      await context.revalidate({
+        urlPath,
+        revalidateHeaders,
+        opts,
+      })
     } else {
       throw new Error(
         `Invariant: required internal revalidate method not passed to api-utils`

packages/next/src/server/dev/next-dev-server.ts

@@ -105,7 +105,7 @@ import { IncrementalCache } from '../lib/incremental-cache'
 import LRUCache from 'next/dist/compiled/lru-cache'
 import { NextUrlWithParsedQuery } from '../request-meta'
 import { deserializeErr, errorToJSON } from '../render'
-import { invokeRequest } from '../lib/server-ipc'
+import { invokeRequest } from '../lib/server-ipc/invoke-request'
 import { generateInterceptionRoutesRewrites } from '../../lib/generate-interception-routes-rewrites'
 
 // Load ReactDevOverlay only when needed

packages/next/src/server/lib/server-ipc.ts → packages/next/src/server/lib/server-ipc/index.ts

@@ -1,9 +1,9 @@
-import type NextServer from '../next-server'
-import { genExecArgv, getNodeOptionsWithoutInspect } from './utils'
-import { deserializeErr, errorToJSON } from '../render'
-import { IncomingMessage } from 'http'
+import type NextServer from '../../next-server'
+
+import { genExecArgv, getNodeOptionsWithoutInspect } from '../utils'
+import { deserializeErr, errorToJSON } from '../../render'
 import crypto from 'crypto'
-import isError from '../../lib/is-error'
+import isError from '../../../lib/is-error'
 
 // we can't use process.send as jest-worker relies on
 // it already and can cause unexpected message errors
@@ -88,7 +88,7 @@ export const createWorker = (
 ) => {
   const { initialEnv } = require('@next/env') as typeof import('@next/env')
   const { Worker } = require('next/dist/compiled/jest-worker')
-  const worker = new Worker(require.resolve('./render-server'), {
+  const worker = new Worker(require.resolve('../render-server'), {
     numWorkers: 1,
     // TODO: do we want to allow more than 10 OOM restarts?
     maxRetries: 10,
@@ -125,94 +125,13 @@ export const createWorker = (
       'clearModuleContext',
     ],
   }) as any as InstanceType<typeof Worker> & {
-    initialize: typeof import('./render-server').initialize
-    deleteCache: typeof import('./render-server').deleteCache
-    deleteAppClientCache: typeof import('./render-server').deleteAppClientCache
+    initialize: typeof import('../render-server').initialize
+    deleteCache: typeof import('../render-server').deleteCache
+    deleteAppClientCache: typeof import('../render-server').deleteAppClientCache
   }
 
   worker.getStderr().pipe(process.stderr)
   worker.getStdout().pipe(process.stdout)
 
   return worker
 }
-
-const forbiddenHeaders = [
-  'accept-encoding',
-  'content-length',
-  'keepalive',
-  'content-encoding',
-  'transfer-encoding',
-  // https://github.com/nodejs/undici/issues/1470
-  'connection',
-]
-
-export const filterReqHeaders = (
-  headers: Record<string, undefined | string | string[]>
-) => {
-  for (const [key, value] of Object.entries(headers)) {
-    if (
-      forbiddenHeaders.includes(key) ||
-      !(Array.isArray(value) || typeof value === 'string')
-    ) {
-      delete headers[key]
-    }
-  }
-  return headers
-}
-
-export const invokeRequest = async (
-  targetUrl: string,
-  requestInit: {
-    headers: IncomingMessage['headers']
-    method: IncomingMessage['method']
-  },
-  readableBody?: import('stream').Readable
-) => {
-  const parsedUrl = new URL(targetUrl)
-
-  // force localhost to IPv4 as some DNS may
-  // resolve to IPv6 instead
-  if (parsedUrl.hostname === 'localhost') {
-    parsedUrl.hostname = '127.0.0.1'
-  }
-  const invokeHeaders = filterReqHeaders({
-    ...requestInit.headers,
-  }) as IncomingMessage['headers']
-
-  const invokeRes = await new Promise<IncomingMessage>(
-    (resolveInvoke, rejectInvoke) => {
-      const http = require('http') as typeof import('http')
-
-      try {
-        const invokeReq = http.request(
-          targetUrl,
-          {
-            headers: invokeHeaders,
-            method: requestInit.method,
-          },
-          (res) => {
-            resolveInvoke(res)
-          }
-        )
-        invokeReq.on('error', (err) => {
-          rejectInvoke(err)
-        })
-
-        if (requestInit.method !== 'GET' && requestInit.method !== 'HEAD') {
-          if (readableBody) {
-            readableBody.pipe(invokeReq)
-            readableBody.on('close', () => {
-              invokeReq.end()
-            })
-          }
-        } else {
-          invokeReq.end()
-        }
-      } catch (err) {
-        rejectInvoke(err)
-      }
-    }
-  )
-
-  return invokeRes
-}

packages/next/src/server/lib/server-ipc/invoke-request.ts

@@ -0,0 +1,59 @@
+import { IncomingMessage } from 'http'
+import { filterReqHeaders } from './utils'
+
+export const invokeRequest = async (
+  targetUrl: string,
+  requestInit: {
+    headers: IncomingMessage['headers']
+    method: IncomingMessage['method']
+  },
+  readableBody?: import('stream').Readable
+) => {
+  const parsedUrl = new URL(targetUrl)
+
+  // force localhost to IPv4 as some DNS may
+  // resolve to IPv6 instead
+  if (parsedUrl.hostname === 'localhost') {
+    parsedUrl.hostname = '127.0.0.1'
+  }
+  const invokeHeaders = filterReqHeaders({
+    ...requestInit.headers,
+  }) as IncomingMessage['headers']
+
+  const invokeRes = await new Promise<IncomingMessage>(
+    (resolveInvoke, rejectInvoke) => {
+      const http = require('http') as typeof import('http')
+
+      try {
+        const invokeReq = http.request(
+          targetUrl,
+          {
+            headers: invokeHeaders,
+            method: requestInit.method,
+          },
+          (res) => {
+            resolveInvoke(res)
+          }
+        )
+        invokeReq.on('error', (err) => {
+          rejectInvoke(err)
+        })
+
+        if (requestInit.method !== 'GET' && requestInit.method !== 'HEAD') {
+          if (readableBody) {
+            readableBody.pipe(invokeReq)
+            readableBody.on('close', () => {
+              invokeReq.end()
+            })
+          }
+        } else {
+          invokeReq.end()
+        }
+      } catch (err) {
+        rejectInvoke(err)
+      }
+    }
+  )
+
+  return invokeRes
+}

packages/next/src/server/lib/server-ipc/utils.ts

@@ -0,0 +1,23 @@
+const forbiddenHeaders = [
+  'accept-encoding',
+  'content-length',
+  'keepalive',
+  'content-encoding',
+  'transfer-encoding',
+  // https://github.com/nodejs/undici/issues/1470
+  'connection',
+]
+
+export const filterReqHeaders = (
+  headers: Record<string, undefined | string | string[]>
+) => {
+  for (const [key, value] of Object.entries(headers)) {
+    if (
+      forbiddenHeaders.includes(key) ||
+      !(Array.isArray(value) || typeof value === 'string')
+    ) {
+      delete headers[key]
+    }
+  }
+  return headers
+}

packages/next/src/server/next-server.ts

@@ -106,7 +106,9 @@ import { getRouteRegex } from '../shared/lib/router/utils/route-regex'
 import { removePathPrefix } from '../shared/lib/router/utils/remove-path-prefix'
 import { addPathPrefix } from '../shared/lib/router/utils/add-path-prefix'
 import { pathHasPrefix } from '../shared/lib/router/utils/path-has-prefix'
-import { filterReqHeaders, invokeRequest } from './lib/server-ipc'
+import { invokeRequest } from './lib/server-ipc/invoke-request'
+import { filterReqHeaders } from './lib/server-ipc/utils'
+import { createRequestResponseMocks } from './lib/mock-request'
 
 export * from './base-server'
 
@@ -906,16 +908,13 @@ export default class NextNodeServer extends BaseServer {
       pageModule,
       {
         ...this.renderOpts.previewProps,
-        revalidate: (newReq: IncomingMessage, newRes: ServerResponse) =>
-          this.getRequestHandler()(
-            new NodeNextRequest(newReq),
-            new NodeNextResponse(newRes)
-          ),
+        revalidate: this.revalidate.bind(this),
         // internal config so is not typed
         trustHostHeader: (this.nextConfig.experimental as Record<string, any>)
           .trustHostHeader,
         allowedRevalidateHeaderKeys:
           this.nextConfig.experimental.allowedRevalidateHeaderKeys,
+        hostname: this.hostname,
       },
       this.minimalMode,
       this.renderOpts.dev,
@@ -1672,6 +1671,36 @@ export default class NextNodeServer extends BaseServer {
     }
   }
 
+  public async revalidate({
+    urlPath,
+    revalidateHeaders,
+    opts,
+  }: {
+    urlPath: string
+    revalidateHeaders: { [key: string]: string | string[] }
+    opts: { unstable_onlyGenerated?: boolean }
+  }) {
+    const mocked = createRequestResponseMocks({
+      url: urlPath,
+      headers: revalidateHeaders,
+    })
+
+    const handler = this.getRequestHandler()
+    await handler(
+      new NodeNextRequest(mocked.req),
+      new NodeNextResponse(mocked.res)
+    )
+    await mocked.res.hasStreamed
+
+    if (
+      mocked.res.getHeader('x-nextjs-cache') !== 'REVALIDATED' &&
+      !(mocked.res.statusCode === 404 && opts.unstable_onlyGenerated)
+    ) {
+      throw new Error(`Invalid response ${mocked.res.statusCode}`)
+    }
+    return {}
+  }
+
   public async render(
     req: BaseNextRequest | IncomingMessage,
     res: BaseNextResponse | ServerResponse,

test/production/app-dir/revalidate/app/layout.js

@@ -0,0 +1,7 @@
+export default function Root({ children }) {
+  return (
+    <html>
+      <body>{children}</body>
+    </html>
+  )
+}

test/production/app-dir/revalidate/app/page.js

@@ -0,0 +1,4 @@
+export default async function Page() {
+  const data = Math.random()
+  return <h1>{data}</h1>
+}

test/production/app-dir/revalidate/next.config.js

@@ -0,0 +1,5 @@
+module.exports = {
+  experimental: {
+    appDir: true,
+  },
+}

test/production/app-dir/revalidate/pages/api/revalidate.js

@@ -0,0 +1,8 @@
+export default async function (_req, res) {
+  try {
+    await res.revalidate('/')
+    return res.json({ revalidated: true })
+  } catch (err) {
+    return res.status(500).send('Error')
+  }
+}