fix: run Cache Components staged renders correctly on workerd

.changeset/bright-caches-stream.md

@@ -0,0 +1,11 @@
+---
+"@opennextjs/cloudflare": patch
+---
+
+fix: support Cache Components rendering on Workers
+
+Next.js renders Cache Components as a pipeline of event loop tasks and, between two of them, drains the immediates React queued so each stage flushes before the next unblocks more content. Its Node.js implementation builds that boundary out of `_idleStart` timer alignment and `process.nextTick`, neither of which behaves the same on workerd, so the render lands a stage late: runtime prefetches drop everything that arrives after their final task aborts the render, and document renders report cached data as uncached and fail. Replace the staged runner with a workerd implementation that waits for the request's outstanding immediates to settle before entering the next stage, and let Next.js resume partially prerendered routes instead of returning their cached shell as a complete response. The wait is scoped to the request rather than to the render, because Next.js awaits the RSC payload before it starts staging, so React resumes much of the work from promises created outside the staged run.
+
+Also keep module loading `CacheSignal` instances and subscriptions request scoped. A shared promise registry forwards current and future imports through request-owned notifications, so overlapping renders wait without sharing timer handles.
+
+Preserve the custom promise API on the wrapped `setImmediate`. Workerd does not provide this hook on the global callback API, but Next.js requires it when it installs `node:timers/promises.setImmediate`.

examples/e2e/experimental/e2e/concurrent-rsc.test.ts

@@ -0,0 +1,210 @@
+import { expect, test, type APIRequestContext } from "@playwright/test";
+
+/**
+ * Cache Components state Next keeps per process is shared by every request in a Worker isolate. When
+ * it holds request-bound I/O handles, an overlapping request clears a handle it does not own and
+ * workerd rejects it mid render, poisoning the isolate. Overlapping RSC prefetches are what a browser
+ * issues while hovering links, so they force that overlap cheaply.
+ */
+
+const ROUTE_PREFETCH = { rsc: "1", "next-router-prefetch": "1" };
+const SEGMENT_PREFETCH = { ...ROUTE_PREFETCH, "next-router-segment-prefetch": "/_tree" };
+// What the router sends on click: a dynamic RSC refetch, not a prefetch.
+const NAVIGATION = { rsc: "1", "next-url": "/" };
+
+/**
+ * Every response is checked against content it must contain, because a poisoned isolate answers 200
+ * with a body that is empty or cut short. Documents carry the prerendered `shell`; `resolvedHtml` and
+ * `resolvedRsc` are flushed last, so only a fully rendered response can contain them. Prefetch depth
+ * varies by route, but every valid prefetch contains a root model instead of only a close marker.
+ *
+ * The tracked-import routes await a dynamic import inside the render, which routes through
+ * `trackPendingChunkLoad` and the module loading `CacheSignal` — the state this suite guards.
+ */
+const ROUTES = [
+	{
+		path: "/ppr",
+		shell: "static component that does not change",
+		resolvedHtml: "This component should be SSR",
+		resolvedRsc: "This component should be SSR",
+	},
+	{
+		path: "/ppr/first",
+		shell: "Static shell",
+		resolvedHtml: "Dynamic slug: first",
+		resolvedRsc: '"data-testid":"dynamic-slug"',
+	},
+	{
+		path: "/ppr/second",
+		shell: "Static shell",
+		resolvedHtml: "Dynamic slug: second",
+		resolvedRsc: '"data-testid":"dynamic-slug"',
+	},
+	{
+		path: "/use-cache/ssr",
+		shell: "Cache",
+		resolvedHtml: 'data-testid="fully-cached"',
+		resolvedRsc: '"data-testid":"fully-cached"',
+	},
+	{
+		path: "/use-cache/isr",
+		shell: "Cache",
+		resolvedHtml: 'data-testid="fully-cached"',
+		resolvedRsc: '"data-testid":"fully-cached"',
+	},
+	{
+		path: "/tracked-import/first",
+		shell: "Tracked import shell",
+		resolvedHtml: "Imported module for first",
+		resolvedRsc: "Imported module for first",
+	},
+	{
+		path: "/tracked-import/second",
+		shell: "Tracked import shell",
+		resolvedHtml: "Imported module for second",
+		resolvedRsc: "Imported module for second",
+	},
+	{
+		path: "/large-shell/concurrent",
+		shell: "Large shell",
+		resolvedHtml: "Large dynamic: ",
+		resolvedRsc: '"data-testid":"large-dynamic"',
+	},
+] as const;
+
+type Route = (typeof ROUTES)[number];
+
+type Fetched = {
+	route: Route;
+	kind: keyof typeof VARIANTS;
+	status: number;
+	contentType: string;
+	body: Buffer;
+};
+
+async function fetchPath(
+	request: APIRequestContext,
+	route: Route,
+	kind: keyof typeof VARIANTS
+): Promise<Fetched> {
+	let response = await request.get(route.path, {
+		headers: VARIANTS[kind],
+		maxRedirects: 0,
+	});
+
+	// Next 16.3 redirects synthetic RSC requests for a fully static route to the same route with the
+	// cache-busting `_rsc` key that a browser normally supplies. Follow only that exact redirect shape,
+	// while keeping every other redirect visible as a failure below.
+	if (kind !== "document" && (response.status() === 307 || response.status() === 308)) {
+		const location = response.headers().location;
+		const redirected = location ? new URL(location, "http://opennext.invalid") : undefined;
+		if (
+			!location?.startsWith("/") ||
+			location.startsWith("//") ||
+			redirected?.pathname !== route.path ||
+			!redirected.searchParams.has("_rsc")
+		) {
+			throw new Error(`${kind} ${route.path} returned an unexpected redirect to ${location ?? "nowhere"}`);
+		}
+
+		await response.dispose();
+		response = await request.get(`${redirected.pathname}${redirected.search}`, {
+			headers: VARIANTS[kind],
+			maxRedirects: 0,
+		});
+	}
+
+	return {
+		route,
+		kind,
+		status: response.status(),
+		contentType: response.headers()["content-type"] ?? "",
+		body: await response.body(),
+	};
+}
+
+const VARIANTS = {
+	document: {} as Record<string, string>,
+	route: ROUTE_PREFETCH,
+	segment: SEGMENT_PREFETCH,
+	navigation: NAVIGATION,
+};
+
+function assertComplete(result: Fetched) {
+	const where = `${result.kind} ${result.route.path}`;
+	const isPrefetch = result.kind === "route" || result.kind === "segment";
+
+	// A poisoned isolate answers 200 with an empty or truncated body, so status alone proves nothing.
+	expect(result.status, `${where} should complete`).toEqual(200);
+
+	const body = result.body.toString("utf8");
+	if (isPrefetch) {
+		// Prefetch depth varies by route: some include the static shell and some only router metadata. A
+		// usable response always has a root model; the reported failure had only a one-byte close marker.
+		expect(result.contentType, `${where} should be an RSC payload`).toContain("text/x-component");
+		expect(body, `${where} should contain a root model`).toContain("0:{");
+		expect(body, `${where} should not contain a Flight error record`).not.toMatch(/^\w+:E\{/m);
+
+		if (result.kind === "segment") {
+			expect(body, `${where} should contain its router tree`).toContain('"tree"');
+			expect(body, `${where} should identify its build`).toContain('"buildId"');
+		}
+		return;
+	}
+
+	expect(body, `${where} should contain its prerendered shell`).toContain(result.route.shell);
+
+	if (result.contentType.includes("text/html")) {
+		// Truncated streams lose the closing tag that Next.js flushes last.
+		expect(body, `${where} should not be truncated`).toContain("</html>");
+		expect(body.replaceAll("<!-- -->", ""), `${where} should have resolved its dynamic hole`).toContain(
+			result.route.resolvedHtml
+		);
+		return;
+	}
+
+	expect(result.contentType, `${where} should be an RSC payload`).toContain("text/x-component");
+
+	// A dynamic RSC response has to carry the resolved model, which a truncated Flight stream loses.
+	expect(body, `${where} should have resolved its dynamic hole`).toContain(result.route.resolvedRsc);
+}
+
+test.describe("concurrent Cache Components requests", () => {
+	test("overlapping RSC prefetches all complete without poisoning the isolate", async ({ request }) => {
+		for (let round = 0; round < 3; round++) {
+			const results = await Promise.all(
+				ROUTES.flatMap((route) => [
+					fetchPath(request, route, "document"),
+					fetchPath(request, route, "route"),
+					fetchPath(request, route, "segment"),
+				])
+			);
+
+			for (const result of results) {
+				assertComplete(result);
+			}
+		}
+
+		// The failure outlives the requests that caused it, so check the isolate still serves traffic.
+		for (const route of ROUTES) {
+			assertComplete(await fetchPath(request, route, "document"));
+		}
+	});
+
+	test("a navigation refetch overlapping its partial prefetch completes", async ({ request }) => {
+		// Hover starts a partial prefetch; clicking before it settles fires the dynamic refetch while
+		// the prefetch request is finishing. The dynamic response must still stream to completion.
+		for (const route of ROUTES) {
+			const prefetch = fetchPath(request, route, "segment");
+			const refetch = fetchPath(request, route, "navigation");
+
+			assertComplete(await refetch);
+			assertComplete(await prefetch);
+		}
+
+		// A hang shows up on later traffic too, so prove the isolate is still healthy.
+		for (const route of ROUTES) {
+			assertComplete(await fetchPath(request, route, "navigation"));
+		}
+	});
+});

examples/e2e/experimental/e2e/hostile-cache-components.test.ts

@@ -0,0 +1,110 @@
+import { expect, test, type APIRequestContext } from "@playwright/test";
+
+const ROUTE_PREFETCH = { rsc: "1", "next-router-prefetch": "1" };
+const SEGMENT_PREFETCH = { ...ROUTE_PREFETCH, "next-router-segment-prefetch": "/_tree" };
+const RUNTIME_PREFETCH = { rsc: "1", "next-router-prefetch": "2" };
+const NAVIGATION = { rsc: "1", "next-url": "/" };
+
+const PAYLOAD_CASES = [
+	{ kind: "document", headers: {} },
+	{ kind: "document", headers: {} },
+	{ kind: "route", headers: ROUTE_PREFETCH },
+	{ kind: "segment", headers: SEGMENT_PREFETCH },
+	{ kind: "runtime", headers: RUNTIME_PREFETCH },
+	{ kind: "navigation", headers: NAVIGATION },
+] as const;
+
+async function getBody(request: APIRequestContext, path: string, headers: Record<string, string> = {}) {
+	const response = await request.get(path, { headers });
+	return {
+		status: response.status(),
+		contentType: response.headers()["content-type"] ?? "",
+		body: await response.body(),
+	};
+}
+
+function expectCompleteShell(body: Buffer, where: string) {
+	expect(body.byteLength, `${where} should include the wide shell`).toBeGreaterThan(100 * 1024);
+	const text = body.toString("utf8");
+	for (const index of [0, 47, 95]) {
+		expect(text, `${where} lost block ${index}`).toMatch(
+			new RegExp(`(?:"data-hostile-block":${index}|data-hostile-block="${index}")`)
+		);
+	}
+	expect(text, `${where} returned a Flight error`).not.toMatch(/^\w+:E\{/m);
+}
+
+function expectCompleteSegment(body: Buffer, where: string) {
+	const text = body.toString("utf8");
+	expect(body.byteLength, `${where} carried only a close marker`).toBeGreaterThan(1);
+	expect(text, `${where} lost its root model`).toContain("0:{");
+	expect(text, `${where} lost its router tree`).toContain('"tree"');
+	expect(text, `${where} lost its build id`).toContain('"buildId"');
+	expect(text, `${where} returned a Flight error`).not.toMatch(/^\w+:E\{/m);
+}
+
+test.describe("hostile Cache Components graph", () => {
+	test("cold and warm documents, prefetches, and navigation payloads complete", async ({ request }) => {
+		const path = `/hostile-shell/cold-${Date.now()}`;
+
+		for (const [index, { kind, headers }] of PAYLOAD_CASES.entries()) {
+			const session = `hostile-${Date.now()}-${index}`;
+			const result = await getBody(request, path, { ...headers, "x-session": session });
+			const where = `${kind} ${path}`;
+
+			expect(result.status, where).toEqual(200);
+			if (kind === "segment") {
+				expectCompleteSegment(result.body, where);
+			} else {
+				expectCompleteShell(result.body, where);
+			}
+			if (kind === "document") {
+				const html = result.body.toString("utf8");
+				expect(result.contentType).toContain("text/html");
+				expect(html).toContain("</html>");
+				expect(html).toContain("Hostile dynamic: ");
+				expect(html).toContain(path.split("/").at(-1));
+				expect(html).toContain(session);
+			} else {
+				expect(result.contentType).toContain("text/x-component");
+			}
+		}
+	});
+
+	test("same-route and cold-route runtime prefetches remain complete under overlap", async ({ request }) => {
+		const repeated = Array.from({ length: 24 }, (_, index) =>
+			getBody(request, "/hostile-shell/repeated", {
+				...RUNTIME_PREFETCH,
+				"x-session": `repeated-${index}`,
+			})
+		);
+		const cold = Array.from({ length: 24 }, (_, index) =>
+			getBody(request, `/hostile-shell/cold-${Date.now()}-${index}`, RUNTIME_PREFETCH)
+		);
+
+		const results = await Promise.all([...repeated, ...cold]);
+		for (const [index, result] of results.entries()) {
+			expect(result.status, `overlapping runtime prefetch ${index}`).toEqual(200);
+			expect(result.contentType).toContain("text/x-component");
+			expectCompleteShell(result.body, `overlapping runtime prefetch ${index}`);
+		}
+	});
+
+	test("client navigation resolves the request hole without a document reload", async ({ page }) => {
+		const session = `hostile-navigation-${Date.now()}`;
+		await page.setExtraHTTPHeaders({ "x-session": session });
+		await page.goto("/hostile-shell/navigation-first");
+
+		await expect(page.getByTestId("hostile-dynamic")).toContainText(
+			`Hostile dynamic: navigation-first:${session}:`
+		);
+		await expect(page.locator('[data-hostile-block="95"]:visible')).toBeVisible({ timeout: 15_000 });
+		await page.getByRole("link", { name: "Hostile shell second item" }).click();
+		await page.waitForURL("/hostile-shell/navigation-second");
+		await expect(page.getByTestId("hostile-dynamic")).toContainText(
+			`Hostile dynamic: navigation-second:${session}:`
+		);
+		await expect(page.locator('[data-hostile-block="95"]:visible')).toBeVisible({ timeout: 15_000 });
+		expect(await page.evaluate(() => performance.getEntriesByType("navigation").length)).toEqual(1);
+	});
+});