fix(execution): let an unread fetch body release the guest event loop

crates/execution/assets/undici-shims/stream.js

@@ -45,6 +45,37 @@ const normalizeStreamError = (value) => {
 	return new Error(String(value));
 };
 
+// Web streams expose no way to observe close or error without locking the
+// stream. Node's eos waits on an internal promise that does not keep the event
+// loop alive, so watch web-streams-polyfill `_state` transitions. A polling
+// timer here held the guest process open while a fetch body stayed unread.
+const webStreamSettleListeners = new WeakMap();
+
+const watchWebStreamState = (stream) => {
+	const listeners = new Set();
+	let state = stream._state;
+	Object.defineProperty(stream, "_state", {
+		configurable: true,
+		enumerable: true,
+		get: () => state,
+		set: (next) => {
+			state = next;
+			if (next === "closed" || next === "errored") {
+				for (const listener of listeners) listener();
+			}
+		},
+	});
+	webStreamSettleListeners.set(stream, listeners);
+	return listeners;
+};
+
+const onWebStreamSettled = (stream, listener) => {
+	const listeners =
+		webStreamSettleListeners.get(stream) ?? watchWebStreamState(stream);
+	listeners.add(listener);
+	return () => listeners.delete(listener);
+};
+
 export const finished = (stream, options, callback) => {
 	let normalizedOptions = options;
 	let normalizedCallback = callback;
@@ -65,14 +96,12 @@ export const finished = (stream, options, callback) => {
 	const readableEnabled = normalizedOptions?.readable !== false;
 	const writableEnabled = normalizedOptions?.writable !== false;
 	let cancelled = false;
-	let timer = null;
+	let unsubscribe = null;
 
 	const cleanup = () => {
 		cancelled = true;
-		if (timer !== null) {
-			clearTimeout(timer);
-			timer = null;
-		}
+		unsubscribe?.();
+		unsubscribe = null;
 	};
 
 	const complete = (error = undefined) => {
@@ -83,27 +112,31 @@ export const finished = (stream, options, callback) => {
 		queueMicrotask(() => done(error));
 	};
 
-	const poll = () => {
+	const check = () => {
 		if (cancelled) {
-			return;
+			return true;
 		}
 		const state = stream?._state;
 		if (state === "errored") {
 			complete(normalizeStreamError(stream?._storedError));
-			return;
+			return true;
 		}
 		if (
 			state === "closed" ||
 			(isWebReadableStream(stream) && !readableEnabled) ||
 			(isWebWritableStream(stream) && !writableEnabled)
 		) {
 			complete();
-			return;
+			return true;
 		}
-		timer = setTimeout(poll, 0);
+		return false;
 	};
 
-	poll();
+	if (!check()) {
+		// The polyfill assigns `_storedError` after `_state`, so read it once the
+		// transition has finished.
+		unsubscribe = onWebStreamSettled(stream, () => queueMicrotask(check));
+	}
 	return cleanup;
 };
 

packages/core/tests/network-vm-fetch.test.ts

@@ -69,3 +69,56 @@ test("httpRequest reaches a guest http.createServer listener", async () => {
 		await vm.process.wait(pid).catch(() => {});
 	}
 });
+
+test("a guest fetch with an unread body lets the process exit", async () => {
+	vm = await AgentOs.create({ permissions: { network: "allow" } });
+	// The server keeps the connection open after each response, so the client
+	// holds an idle keep-alive socket and an unread body when its script ends.
+	await vm.writeFile(
+		"/app/server.mjs",
+		[
+			'import net from "node:net";',
+			"const response =",
+			'  "HTTP/1.1 200 OK\\r\\nContent-Length: 2\\r\\nConnection: keep-alive\\r\\nKeep-Alive: timeout=120\\r\\n\\r\\nok";',
+			'const server = net.createServer((socket) => socket.on("data", () => socket.write(response)));',
+			'server.listen(3000, "127.0.0.1", () => console.log("LISTENING"));',
+		].join("\n"),
+	);
+	await vm.writeFile(
+		"/app/client.mjs",
+		'console.log((await fetch("http://127.0.0.1:3000/")).status);',
+	);
+
+	let listening!: () => void;
+	const ready = new Promise<void>((resolve) => {
+		listening = resolve;
+	});
+	const server = await vm.process.spawn("node", ["/app/server.mjs"], {
+		onStdout: (chunk) => {
+			if (textDecoder.decode(chunk).includes("LISTENING")) listening();
+		},
+	});
+	try {
+		await ready;
+		let stdout = "";
+		let stderr = "";
+		const startedAt = Date.now();
+		const client = await vm.process.spawn("node", ["/app/client.mjs"], {
+			onStdout: (chunk) => {
+				stdout += textDecoder.decode(chunk);
+			},
+			onStderr: (chunk) => {
+				stderr += textDecoder.decode(chunk);
+			},
+		});
+		const exit = await vm.process.wait(client.pid);
+		const elapsedMs = Date.now() - startedAt;
+
+		expect(exit.exitCode, stderr).toBe(0);
+		expect(stdout.trim()).toBe("200");
+		expect(elapsedMs).toBeLessThan(5_000);
+	} finally {
+		await vm.process.kill(server.pid);
+		await vm.process.wait(server.pid).catch(() => {});
+	}
+}, 60_000);