Ensure that the initial request.signal is passed to the wrapper (#848)

Author: sommeeeer

.changeset/four-walls-read.md

@@ -0,0 +1,22 @@
+---
+"@opennextjs/cloudflare": minor
+---
+
+Ensure that the initial request.signal is passed to the wrapper
+
+`request.signal.onabort` is now supported in route handlers. It requires that the signal from the original worker's request is passed to the handler. It will then pass along that `AbortSignal` through the `streamCreator` in the wrapper. This signal will destroy the response sent to NextServer when a client aborts, thus triggering the signal in the route handler.
+
+See the changelog in Cloudflare [here](https://developers.cloudflare.com/changelog/2025-05-22-handle-request-cancellation/).
+
+**Note:**  
+If you have a custom worker, you must update your code to pass the original `request.signal` to the handler. You also need to enable the compatibility flag `enable_request_signal` to use this feature.
+
+For example:
+
+```js
+// Before:
+return handler(reqOrResp, env, ctx);
+
+// After:
+return handler(reqOrResp, env, ctx, request.signal);
+```

examples/playground15/app/api/signal/abort/route.ts

@@ -0,0 +1,42 @@
+import { NextRequest, NextResponse } from "next/server";
+
+export async function GET(request: NextRequest) {
+	const stream = new ReadableStream({
+		async start(controller) {
+			request.signal.addEventListener("abort", async () => {
+				/**
+				 * I was not allowed to `revalidatePath` or `revalidateTag` here. I would run into this error from Next:
+				 * Error: Invariant: static generation store missing in revalidatePath
+				 *
+				 * Affected line:
+				 * https://github.com/vercel/next.js/blob/ea08bf27/packages/next/src/server/web/spec-extension/revalidate.ts#L89-L92
+				 *
+				 */
+				const host = new URL(request.url).host;
+				// We need to set the protocol to http, cause in `wrangler dev` it will be https
+				await fetch(`http://${host}/api/signal/revalidate`);
+
+				try {
+					controller.close();
+				} catch (_) {
+					// Controller might already be closed, which is fine
+					// This does only happen in `next start`
+				}
+			});
+
+			let i = 0;
+			while (!request.signal.aborted) {
+				controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify({ number: i++ })}\n\n`));
+				await new Promise((resolve) => setTimeout(resolve, 2_000));
+			}
+		},
+	});
+
+	return new NextResponse(stream, {
+		headers: {
+			"Content-Type": "text/event-stream",
+			"Cache-Control": "no-cache",
+			Connection: "keep-alive",
+		},
+	});
+}

examples/playground15/app/api/signal/revalidate/route.ts

@@ -0,0 +1,9 @@
+import { revalidatePath } from "next/cache";
+
+export const dynamic = "force-dynamic";
+
+export async function GET() {
+	revalidatePath("/signal");
+
+	return new Response("ok");
+}

examples/playground15/app/signal/_components/sse.tsx

@@ -0,0 +1,63 @@
+"use client";
+
+import { useEffect, useRef, useState } from "react";
+
+export default function SSE() {
+	const [events, setEvents] = useState<any[]>([]);
+	const [start, setStart] = useState(false);
+	const eventSourceRef = useRef<EventSource | null>(null);
+
+	useEffect(() => {
+		if (start) {
+			const e = new EventSource("/api/signal/abort");
+			eventSourceRef.current = e;
+
+			e.onmessage = (msg) => {
+				try {
+					const data = JSON.parse(msg.data);
+					setEvents((prev) => prev.concat(data));
+				} catch (err) {
+					console.log("failed to parse: ", err, msg);
+				}
+			};
+		}
+
+		return () => {
+			if (eventSourceRef.current) {
+				eventSourceRef.current.close();
+				eventSourceRef.current = null;
+			}
+		};
+	}, [start]);
+
+	const handleStart = () => {
+		setEvents([]);
+		setStart(true);
+	};
+
+	const handleClose = () => {
+		if (eventSourceRef.current) {
+			eventSourceRef.current.close();
+			eventSourceRef.current = null;
+		}
+		setStart(false);
+	};
+
+	return (
+		<section>
+			<div>
+				<button onClick={handleStart} data-testid="start-button" disabled={start}>
+					Start
+				</button>
+				<button onClick={handleClose} data-testid="close-button" disabled={!start}>
+					Close
+				</button>
+			</div>
+			{events.map((e, i) => (
+				<div key={i}>
+					Message {i}: {JSON.stringify(e)}
+				</div>
+			))}
+		</section>
+	);
+}

examples/playground15/app/signal/page.tsx

@@ -0,0 +1,15 @@
+import SSE from "./_components/sse";
+
+export const dynamic = "force-static";
+
+export default function Page() {
+	const date = new Date().toISOString();
+
+	return (
+		<main>
+			<h1 data-testid="date">{date}</h1>
+
+			<SSE />
+		</main>
+	);
+}

examples/playground15/e2e/signal.test.ts

@@ -0,0 +1,41 @@
+import { expect, test } from "@playwright/test";
+
+test("Request Signal On Abort", async ({ page }) => {
+	// First, get the initial date
+	await page.goto("/signal");
+	const initialDate = await page.getByTestId("date").textContent();
+	expect(initialDate).toBeTruthy();
+
+	// Start the EventSource
+	await page.getByTestId("start-button").click();
+	const msg0 = page.getByText(`Message 0: {"number":0}`);
+	await expect(msg0).toBeVisible();
+
+	// 2nd message shouldn't arrive yet
+	let msg1 = page.getByText(`Message 1: {"number":1}`);
+	await expect(msg1).not.toBeVisible();
+	await page.waitForTimeout(2_000);
+	// 2nd message should arrive after 2s
+	msg1 = page.getByText(`Message 2: {"number":2}`);
+	await expect(msg1).toBeVisible();
+
+	// 3rd message shouldn't arrive yet
+	let msg3 = page.getByText(`Message 3: {"number":3}`);
+	await expect(msg3).not.toBeVisible();
+	await page.waitForTimeout(2_000);
+	// 3rd message should arrive after 2s
+	msg3 = page.getByText(`Message 3: {"number":3}`);
+	await expect(msg3).toBeVisible();
+
+	// We then click the close button to close the EventSource and trigger the onabort eventz[]
+	await page.getByTestId("close-button").click();
+
+	// Wait for revalidation to finish
+	await page.waitForTimeout(4_000);
+
+	// Check that the onabort event got emitted and revalidated the page from a fetch
+	await page.goto("/signal");
+	const finalDate = await page.getByTestId("date").textContent();
+	expect(finalDate).toBeTruthy();
+	expect(new Date(finalDate!).getTime()).toBeGreaterThan(new Date(initialDate!).getTime());
+});

examples/playground15/open-next.config.ts

@@ -1,9 +1,13 @@
 import { defineCloudflareConfig, type OpenNextConfig } from "@opennextjs/cloudflare";
 import r2IncrementalCache from "@opennextjs/cloudflare/overrides/incremental-cache/r2-incremental-cache";
+import doQueue from "@opennextjs/cloudflare/overrides/queue/do-queue";
+import d1NextTagCache from "@opennextjs/cloudflare/overrides/tag-cache/d1-next-tag-cache";
 
 export default {
 	...defineCloudflareConfig({
 		incrementalCache: r2IncrementalCache,
+		queue: doQueue,
+		tagCache: d1NextTagCache,
 	}),
 	cloudflare: {
 		skewProtection: {

examples/playground15/wrangler.jsonc

@@ -3,7 +3,7 @@
 	"main": ".open-next/worker.js",
 	"name": "playground15",
 	"compatibility_date": "2024-12-30",
-	"compatibility_flags": ["nodejs_compat", "global_fetch_strictly_public"],
+	"compatibility_flags": ["nodejs_compat", "global_fetch_strictly_public", "enable_request_signal"],
 	"assets": {
 		"directory": ".open-next/assets",
 		"binding": "ASSETS",
@@ -17,5 +17,32 @@
 	],
 	"vars": {
 		"hello": "Hello World from the cloudflare context!"
-	}
+	},
+	"services": [
+		{
+			"binding": "WORKER_SELF_REFERENCE",
+			"service": "playground15"
+		}
+	],
+	"durable_objects": {
+		"bindings": [
+			{
+				"name": "NEXT_CACHE_DO_QUEUE",
+				"class_name": "DOQueueHandler"
+			}
+		]
+	},
+	"migrations": [
+		{
+			"tag": "v1",
+			"new_sqlite_classes": ["DOQueueHandler"]
+		}
+	],
+	"d1_databases": [
+		{
+			"binding": "NEXT_TAG_CACHE_D1",
+			"database_id": "db_id",
+			"database_name": "db_name"
+		}
+	]
 }

packages/cloudflare/src/cli/templates/worker.ts

@@ -54,7 +54,7 @@ export default {
 			// @ts-expect-error: resolved by wrangler build
 			const { handler } = await import("./server-functions/default/handler.mjs");
 
-			return handler(reqOrResp, env, ctx);
+			return handler(reqOrResp, env, ctx, request.signal);
 		});
 	},
 } satisfies ExportedHandler<CloudflareEnv>;