diff --git a/crates/execution/assets/undici-shims/stream.js b/crates/execution/assets/undici-shims/stream.js index 524c6c21d4..90e053781d 100644 --- a/crates/execution/assets/undici-shims/stream.js +++ b/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,14 +112,14 @@ 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" || @@ -98,12 +127,16 @@ export const finished = (stream, options, callback) => { (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; }; diff --git a/packages/core/tests/network-vm-fetch.test.ts b/packages/core/tests/network-vm-fetch.test.ts index be94f62bef..75ed05310a 100644 --- a/packages/core/tests/network-vm-fetch.test.ts +++ b/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((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);