Cancel response stream when connection closes

.changeset/modern-ways-develop.md

@@ -0,0 +1,5 @@
+---
+'@astrojs/node': patch
+---
+
+Cancel response stream when connection closes

packages/integrations/node/src/nodeMiddleware.ts

@@ -1,8 +1,6 @@
 import type { NodeApp } from 'astro/app/node';
 import type { ServerResponse } from 'node:http';
-import type { Readable } from 'stream';
 import { createOutgoingHttpHeaders } from './createOutgoingHttpHeaders.js';
-import { responseIterator } from './response-iterator.js';
 import type { ErrorHandlerParams, Options, RequestHandlerParams } from './types.js';
 
 // Disable no-unused-vars to avoid breaking signature change
@@ -79,8 +77,14 @@ async function writeWebResponse(app: NodeApp, res: ServerResponse, webResponse:
 	res.writeHead(status, nodeHeaders);
 	if (webResponse.body) {
 		try {
-			for await (const chunk of responseIterator(webResponse) as unknown as Readable) {
-				res.write(chunk);
+			const reader = webResponse.body.getReader();
+			res.on("close", () => {
+				reader.cancel();
+			})
+			let result = await reader.read();
+			while (!result.done) {
+				res.write(result.value);
+				result = await reader.read();
 			}
 		} catch (err: any) {
 			console.error(err?.stack || err?.message || String(err));

packages/integrations/node/test/api-route.test.js

@@ -89,4 +89,23 @@ describe('API routes', () => {
 		let [out] = await done;
 		expect(new Uint8Array(out.buffer)).to.deep.equal(expectedDigest);
 	});
+
+	it('Can bail on streaming', async () => {
+		const { handler } = await import('./fixtures/api-route/dist/server/entry.mjs');
+		let { req, res, done } = createRequestAndResponse({
+			url: '/streaming',
+		});
+
+		let locals = { cancelledByTheServer: false };
+
+		handler(req, res, () => {}, locals);
+		req.send();
+
+		await new Promise((resolve) => setTimeout(resolve, 500));
+		res.emit("close");
+
+		await done;
+
+		expect(locals).to.deep.include({ cancelledByTheServer: true });
+	});
 });

packages/integrations/node/test/fixtures/api-route/src/pages/hash.ts

@@ -1,6 +1,6 @@
 import crypto from 'node:crypto';
 
-export async function post({ request }: { request: Request }) {
+export async function POST({ request }: { request: Request }) {
 	const hash = crypto.createHash('sha256');
 
     const iterable = request.body as unknown as AsyncIterable<Uint8Array>;

packages/integrations/node/test/fixtures/api-route/src/pages/streaming.ts

@@ -0,0 +1,22 @@
+export const GET = ({ locals }) => {
+    let sentChunks = 0;
+
+    const readableStream = new ReadableStream({
+        async pull(controller) {
+            if (sentChunks === 3) return controller.close();
+            else sentChunks++;
+
+            await new Promise(resolve => setTimeout(resolve, 1000));
+            controller.enqueue(new TextEncoder().encode('hello\n'));
+        },
+        cancel() {
+            locals.cancelledByTheServer = true;
+        }
+    });
+
+    return new Response(readableStream, {
+        headers: {
+            "Content-Type": "text/event-stream"
+        }
+    });
+}