Stop aborted request bodies from logging duplicate unhandled rejections in @astrojs/node

.changeset/sharp-doors-vanish.md

@@ -0,0 +1,5 @@
+---
+'@astrojs/node': patch
+---
+
+Fixes aborted request bodies causing duplicate unhandled rejection logs in standalone mode when using `src/fetch.ts`

packages/integrations/node/src/serve-app.ts

@@ -57,16 +57,15 @@ async function readErrorPageFromDisk(
  * Intended to be used in both standalone and middleware mode.
  */
 export function createAppHandler(app: BaseApp, options: Options): RequestHandler {
-	/**
-	 * Keep track of the current request path using AsyncLocalStorage.
-	 * Used to log unhandled rejections with a helpful message.
-	 */
-	const als = new AsyncLocalStorage<string>();
 	const logger = app.adapterLogger;
+	const requestContext = new AsyncLocalStorage<string>();
+	const abortedRequestErrors = new WeakSet<Error>();
+
 	process.on('unhandledRejection', (reason) => {
-		const requestUrl = als.getStore();
-		logger.error(`Unhandled rejection while rendering ${requestUrl}`);
-		console.error(reason);
+		const requestUrl = requestContext.getStore();
+		if (!requestUrl || (reason instanceof Error && abortedRequestErrors.has(reason))) return;
+		const error = reason instanceof Error ? reason.stack || reason.message : String(reason);
+		logger.error(`Unhandled rejection while rendering ${requestUrl}\n${error}`);
 	});
 
 	const client = resolveClientDir(options);
@@ -94,6 +93,13 @@ export function createAppHandler(app: BaseApp, options: Options): RequestHandler
 			: options.bodySizeLimit;
 
 	return async (req, res, next, locals) => {
+		// Tag the exact error object emitted when this request's socket dies so
+		// the unhandledRejection listener can tell client disconnects apart from
+		// genuine errors that happen to carry the same code (e.g. ECONNRESET
+		// from an outbound fetch in user code).
+		req.once('error', (error: NodeJS.ErrnoException) => {
+			if (error.code === 'ECONNRESET') abortedRequestErrors.add(error);
+		});
 		let request: Request;
 		try {
 			request = createRequestFromNodeRequest(req, {
@@ -109,34 +115,54 @@ export function createAppHandler(app: BaseApp, options: Options): RequestHandler
 			return;
 		}
 
-		// Include prerendered routes so static-mode redirects remain dynamic.
-		let routeData = app.match(request, true);
-		// Normal matching can select a lower-priority on-demand route when a prerendered route
-		// matches first.
-		if (routeData?.prerender && PRERENDERED_ROUTE_TYPES.includes(routeData.type)) {
-			routeData = app.match(request);
-		}
-		if (routeData) {
-			const response = await als.run(request.url, () =>
-				app.render(request, {
-					addCookieHeader: true,
-					locals,
-					routeData,
-					prerenderedErrorPageFetch,
-				}),
-			);
-			await writeResponse(response, res);
-		} else if (next) {
-			// Since we're not calling `writeResponse()`, clean up the AbortController and socket listeners
-			const cleanup = getAbortControllerCleanup(req);
-			if (cleanup) cleanup();
-			return next();
-		} else {
-			const response = await app.render(request, {
-				addCookieHeader: true,
-				prerenderedErrorPageFetch,
-			});
-			await writeResponse(response, res);
+		try {
+			// Include prerendered routes so static-mode redirects remain dynamic.
+			let routeData = app.match(request, true);
+			// Normal matching can select a lower-priority on-demand route when a prerendered route
+			// matches first.
+			if (routeData?.prerender && PRERENDERED_ROUTE_TYPES.includes(routeData.type)) {
+				routeData = app.match(request);
+			}
+			if (routeData) {
+				await requestContext.run(request.url, async () => {
+					const response = await app.render(request, {
+						addCookieHeader: true,
+						locals,
+						routeData,
+						prerenderedErrorPageFetch,
+					});
+					await writeResponse(response, res);
+				});
+			} else if (next) {
+				// Since we're not calling `writeResponse()`, clean up the AbortController and socket listeners
+				const cleanup = getAbortControllerCleanup(req);
+				if (cleanup) cleanup();
+				return next();
+			} else {
+				await requestContext.run(request.url, async () => {
+					const response = await app.render(request, {
+						addCookieHeader: true,
+						prerenderedErrorPageFetch,
+					});
+					await writeResponse(response, res);
+				});
+			}
+		} catch (err) {
+			// Client disconnected mid-request: the exact error object is tagged
+			// in `abortedRequestErrors`, and there is nobody left to respond to.
+			if (err instanceof Error && abortedRequestErrors.has(err)) return;
+			// Any other failure is a genuine request error. Log it with the
+			// request URL, then finish the response: a 500 while the headers are
+			// still pending, or terminate the connection once a response has
+			// already started.
+			const error = err instanceof Error ? err.stack || err.message : String(err);
+			logger.error(`Could not render ${request.url}\n${error}`);
+			if (!res.headersSent) {
+				res.statusCode = 500;
+				res.end('Internal Server Error');
+			} else {
+				res.destroy();
+			}
 		}
 	};
 }

packages/integrations/node/test/aborted-request.test.ts

@@ -0,0 +1,139 @@
+import * as assert from 'node:assert/strict';
+import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process';
+import { once } from 'node:events';
+import { createConnection, createServer } from 'node:net';
+import { after, before, describe, it } from 'node:test';
+import { fileURLToPath } from 'node:url';
+import nodejs from '../dist/index.js';
+import { type Fixture, loadFixture } from './test-utils.ts';
+
+const READING_BODY_MARKER = 'reading-json-body';
+
+describe('aborted requests', () => {
+	let fixture: Fixture;
+	let server: ChildProcessWithoutNullStreams;
+	let port: number;
+	let output = '';
+
+	before(async () => {
+		fixture = await loadFixture({
+			root: './fixtures/aborted-request/',
+			output: 'server',
+			logLevel: 'info',
+			adapter: nodejs({ mode: 'standalone' }),
+		});
+		await fixture.build();
+		port = await getAvailablePort();
+		server = spawn(
+			process.execPath,
+			[fileURLToPath(new URL('./fixtures/aborted-request/dist/server/entry.mjs', import.meta.url))],
+			{
+				env: {
+					...process.env,
+					ASTRO_NODE_AUTOSTART: 'enabled',
+					ASTRO_NODE_LOGGING: 'disabled',
+					HOST: '127.0.0.1',
+					PORT: String(port),
+				},
+			},
+		);
+		server.stdout.setEncoding('utf8');
+		server.stderr.setEncoding('utf8');
+		server.stdout.on('data', (data) => (output += data));
+		server.stderr.on('data', (data) => (output += data));
+		await waitFor(
+			async () => {
+				if (server.exitCode !== null) return false;
+				const response = await fetch(`http://127.0.0.1:${port}`);
+				await response.body?.cancel();
+				return true;
+			},
+			() => `Timed out waiting for server to listen:\n${output}`,
+		);
+	});
+
+	after(async () => {
+		server.kill();
+		if (server.exitCode === null) await once(server, 'exit');
+		await fixture.clean();
+	});
+
+	it('does not report an interrupted request body as an unhandled rejection', async () => {
+		const outputStart = output.length;
+		const socket = await connect(port);
+		await writePartialRequest(socket, port);
+		await waitFor(
+			() => output.slice(outputStart).includes(READING_BODY_MARKER),
+			() => 'Timed out waiting for the server to start reading the request body',
+		);
+		socket.destroy();
+		await waitForStableOutput(() => output.length);
+
+		assert.doesNotMatch(output.slice(outputStart), /ECONNRESET|Unhandled rejection/);
+		const response = await fetch(`http://127.0.0.1:${port}`);
+		assert.equal(response.status, 200);
+		assert.equal(await response.text(), 'ok');
+	});
+});
+
+async function getAvailablePort(): Promise<number> {
+	const server = createServer();
+	await new Promise<void>((resolve, reject) => {
+		server.once('error', reject);
+		server.listen(0, '127.0.0.1', resolve);
+	});
+	const address = server.address();
+	assert.ok(address && typeof address === 'object');
+	await new Promise<void>((resolve, reject) =>
+		server.close((error) => (error ? reject(error) : resolve())),
+	);
+	return address.port;
+}
+
+function connect(port: number): Promise<import('node:net').Socket> {
+	return new Promise((resolve, reject) => {
+		const socket = createConnection(port, '127.0.0.1', () => resolve(socket));
+		socket.once('error', reject);
+	});
+}
+
+function writePartialRequest(socket: import('node:net').Socket, port: number): Promise<void> {
+	return new Promise((resolve, reject) =>
+		socket.write(
+			`POST / HTTP/1.1\r\nHost: 127.0.0.1:${port}\r\nContent-Type: application/json\r\nContent-Length: 1000000\r\nConnection: close\r\n\r\n{"partial":`,
+			(error) => (error ? reject(error) : resolve()),
+		),
+	);
+}
+
+async function waitForStableOutput(getLength: () => number): Promise<void> {
+	let lastLength = -1;
+	let stable = 0;
+	for (let attempts = 0; attempts < 40; attempts++) {
+		const length = getLength();
+		if (length === lastLength) {
+			stable++;
+			if (stable >= 3) return;
+		} else {
+			lastLength = length;
+			stable = 0;
+		}
+		await new Promise((resolve) => setTimeout(resolve, 200));
+	}
+	assert.fail('Output never stabilized');
+}
+
+async function waitFor(
+	check: () => boolean | Promise<boolean>,
+	getError: () => string,
+): Promise<void> {
+	for (let attempts = 0; attempts < 100; attempts++) {
+		try {
+			if (await check()) return;
+		} catch {
+			// Keep polling.
+		}
+		await new Promise((resolve) => setTimeout(resolve, 50));
+	}
+	assert.fail(getError());
+}

packages/integrations/node/test/fixtures/aborted-request/astro.config.mjs

@@ -0,0 +1,8 @@
+import { defineConfig, logHandlers } from 'astro/config';
+import node from '@astrojs/node';
+
+export default defineConfig({
+	output: 'server',
+	adapter: node({ mode: 'standalone' }),
+	logger: logHandlers.json(),
+});

packages/integrations/node/test/fixtures/aborted-request/package.json

@@ -0,0 +1,9 @@
+{
+  "name": "@test/aborted-request",
+  "version": "0.0.0",
+  "private": true,
+  "dependencies": {
+    "@astrojs/node": "workspace:*",
+    "astro": "workspace:*"
+  }
+}

packages/integrations/node/test/fixtures/aborted-request/src/fetch.ts

@@ -0,0 +1,9 @@
+export default {
+	async fetch(request: Request) {
+		if (request.method === 'POST') {
+			console.error('reading-json-body');
+			await request.json();
+		}
+		return new Response('ok');
+	},
+};

packages/integrations/node/test/fixtures/aborted-request/src/pages/index.astro

@@ -0,0 +1,3 @@
+---
+---
+ok