Handle the reader error (#5616)

* Handle the reader error

* Add test

* Create lucky-plums-listen.md

Co-authored-by: Rich Harris <hello@rich-harris.dev>

.changeset/lucky-plums-listen.md

@@ -0,0 +1,5 @@
+---
+"@sveltejs/kit": patch
+---
+
+Handle errors in Node streams

packages/kit/src/node/index.js

@@ -112,26 +112,35 @@ export async function setResponse(res, response) {
 
 	res.writeHead(response.status, headers);
 
-	if (response.body) {
-		const reader = response.body.getReader();
+	if (!response.body) {
+		res.end();
+		return;
+	}
 
-		if (res.destroyed) {
-			reader.cancel();
-			return;
-		}
+	const reader = response.body.getReader();
 
-		let cancelled = false;
+	if (res.destroyed) {
+		reader.cancel();
+		return;
+	}
 
-		res.on('close', () => {
-			reader.cancel();
-			cancelled = true;
-		});
+	let cancelled = false;
 
-		res.on('error', (error) => {
-			reader.cancel(error);
-			cancelled = true;
-		});
+	res.on('close', () => {
+		if (cancelled) return;
+		cancelled = true;
+		reader.cancel();
+		res.emit('drain');
+	});
 
+	res.on('error', (error) => {
+		if (cancelled) return;
+		cancelled = true;
+		reader.cancel(error);
+		res.emit('drain');
+	});
+
+	try {
 		for (;;) {
 			const { done, value } = await reader.read();
 
@@ -148,7 +157,8 @@ export async function setResponse(res, response) {
 				await new Promise((fulfil) => res.once('drain', fulfil));
 			}
 		}
-	} else {
-		res.end();
+	} catch (error) {
+		cancelled = true;
+		res.destroy(error instanceof Error ? error : undefined);
 	}
 }

packages/kit/test/apps/basics/src/routes/endpoint-output/stream-throw-error.js

@@ -0,0 +1,13 @@
+/** @type {import('@sveltejs/kit').RequestHandler} */
+export function GET() {
+	return {
+		headers: {
+			'content-type': 'application/octet-stream'
+		},
+		body: new ReadableStream({
+			pull() {
+				throw Error('simulate error');
+			}
+		})
+	};
+}

packages/kit/test/apps/basics/test/server.test.js

@@ -272,6 +272,9 @@ test.describe('Endpoints', () => {
 	});
 
 	test('body can be a binary ReadableStream', async ({ request }) => {
+		const interruptedResponse = request.get('/endpoint-output/stream-throw-error');
+		await expect(interruptedResponse).rejects.toThrow('socket hang up');
+
 		const response = await request.get('/endpoint-output/stream');
 		const body = await response.body();
 		const digest = createHash('sha256').update(body).digest('base64url');