From 8efb21f55ca55ac8df24ec69ac49651732ea40f7 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Sun, 1 Nov 2026 03:55:48 +0000 Subject: [PATCH] fix(server): close abandoned workspace service connections (#8421) * fix(server): close the upstream connection when a proxied client leaves The service proxy piped the service's response to the client but never destroyed the upstream request when the client went away. The upstream socket stayed open and unread, so a Server-Sent Events service kept writing into it and grew in memory. The proxy now destroys the upstream request when the client response closes before it finished, without logging that deliberate abort as an unreachable upstream, and a WebSocket upgrade closes the service socket when the client socket closes. Fixes #8149 * test(server): publish terminal activity fixture atomically File existence did not mean the child had finished writing its JSON. Publish the completed fixture by rename so a concurrent reader cannot parse an empty file. Co-authored-by: Dominic Anaeto * test(server): keep rollback listener ports distinct Selecting both free ports before binding either lets the OS return the same port twice. Hold the intentionally occupied main port before selecting the standalone port so the rollback probe cannot reach the wrong server. --------- Co-authored-by: Dominic Anaeto Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> --- .../server/src/server/bootstrap.smoke.test.ts | 2 +- .../server/src/server/service-proxy.test.ts | 277 ++++++++++++++++-- packages/server/src/server/service-proxy.ts | 10 + .../server/terminal-activity-route.test.ts | 8 +- 4 files changed, 264 insertions(+), 33 deletions(-) diff --git a/packages/server/src/server/bootstrap.smoke.test.ts b/packages/server/src/server/bootstrap.smoke.test.ts index 76fa8cc7d1..ac045741f1 100644 --- a/packages/server/src/server/bootstrap.smoke.test.ts +++ b/packages/server/src/server/bootstrap.smoke.test.ts @@ -670,11 +670,11 @@ describe("paseo daemon bootstrap", () => { test("rolls back standalone listener without starting plugins when main listen fails", async () => { const mainPort = await findFreePort(); - const standalonePort = await findFreePort(); const occupiedMain = http.createServer((_req, res) => { res.end("occupied-main"); }); await new Promise((resolve) => occupiedMain.listen(mainPort, "127.0.0.1", resolve)); + const standalonePort = await findFreePort(); const paseoHomeRoot = await mkdtemp(path.join(os.tmpdir(), "paseo-main-rollback-")); const paseoHome = path.join(paseoHomeRoot, ".paseo"); diff --git a/packages/server/src/server/service-proxy.test.ts b/packages/server/src/server/service-proxy.test.ts index 5764b19539..eb679d5629 100644 --- a/packages/server/src/server/service-proxy.test.ts +++ b/packages/server/src/server/service-proxy.test.ts @@ -1,7 +1,9 @@ +import { type EventEmitter, once } from "node:events"; import { readdirSync, readFileSync, statSync } from "node:fs"; import path from "node:path"; import http from "node:http"; import net from "node:net"; +import type { Duplex } from "node:stream"; import express from "express"; import { describe, expect, it } from "vitest"; import pino from "pino"; @@ -332,48 +334,39 @@ describe("service proxy subsystem shape", () => { }); }); -interface ForwardedFixture { +interface ProxiedServiceFixture { upstreamPort: number; daemonPort: number; + daemon: http.Server; hostname: string; close(): Promise; } /** - * Runs a real workspace service behind a real daemon listener. The upstream - * echoes the headers it received so tests can assert what actually crossed the - * proxy, not what a helper returned. + * Runs a real workspace service behind a real daemon listener that serves the + * service proxy for both HTTP and upgrades. */ -async function startForwardedHeadersFixture(upstreamHost = "127.0.0.1"): Promise { - const upstream = http.createServer((req, res) => { - let body = ""; - req.on("data", (chunk: Buffer) => { - body += chunk.toString(); +async function startProxiedService( + upstream: http.Server, + options: { upstreamHost?: string; logger?: pino.Logger } = {}, +): Promise { + const { upstreamHost = "127.0.0.1", logger: proxyLogger = logger } = options; + // Upgraded sockets keep the server alive; close() alone would hang. + const sockets = new Set(); + function trackConnections(server: http.Server) { + server.on("connection", (socket: net.Socket) => { + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); }); - req.on("end", () => { - res.writeHead(200, { "content-type": "application/json" }); - res.end(JSON.stringify({ ...req.headers, method: req.method, url: req.url, body })); - }); - }); - const upgradeSockets: net.Socket[] = []; - upstream.on("upgrade", (req, socket) => { - upgradeSockets.push(socket); - // The client resets this connection once it has the echo, so the reset - // reaches us as an 'error'. A raw upgrade socket has no default handler — - // without this the event goes unhandled and takes down the test process. - socket.on("error", () => socket.destroy()); - const payload = JSON.stringify(req.headers); - socket.write( - `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nX-Echo-Length: ${payload.length}\r\n\r\n${payload}`, - ); - }); + } + trackConnections(upstream); await new Promise((resolve, reject) => { upstream.once("error", reject); upstream.listen(0, upstreamHost, resolve); }); const upstreamPort = (upstream.address() as net.AddressInfo).port; - const serviceProxy = createServiceProxySubsystem({ logger }); + const serviceProxy = createServiceProxySubsystem({ logger: proxyLogger }); const route = serviceProxy.registerWorkspaceService({ workspaceId: "workspace-a", projectSlug: "repo", @@ -390,24 +383,51 @@ async function startForwardedHeadersFixture(upstreamHost = "127.0.0.1"): Promise res.status(404).send("404 Not Found"); }); const daemon = http.createServer(app); + trackConnections(daemon); daemon.on("upgrade", serviceProxy.upgradeHandler({ passthroughUnknown: false })); await new Promise((resolve) => daemon.listen(daemonPort, "127.0.0.1", resolve)); return { upstreamPort, daemonPort, + daemon, hostname: route.hostname, async close() { - // Upgraded sockets keep the server alive; close() alone would hang. - for (const socket of upgradeSockets) socket.destroy(); - daemon.closeAllConnections(); + for (const socket of sockets) socket.destroy(); await new Promise((resolve) => daemon.close(() => resolve())); - upstream.closeAllConnections(); await new Promise((resolve) => upstream.close(() => resolve())); }, }; } +/** + * The upstream echoes the headers it received so tests can assert what + * actually crossed the proxy, not what a helper returned. + */ +function startForwardedHeadersFixture(upstreamHost = "127.0.0.1"): Promise { + const upstream = http.createServer((req, res) => { + let body = ""; + req.on("data", (chunk: Buffer) => { + body += chunk.toString(); + }); + req.on("end", () => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ...req.headers, method: req.method, url: req.url, body })); + }); + }); + upstream.on("upgrade", (req, socket) => { + // The client resets this connection once it has the echo, so the reset + // reaches us as an 'error'. A raw upgrade socket has no default handler — + // without this the event goes unhandled and takes down the test process. + socket.on("error", () => socket.destroy()); + const payload = JSON.stringify(req.headers); + socket.write( + `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nX-Echo-Length: ${payload.length}\r\n\r\n${payload}`, + ); + }); + return startProxiedService(upstream, { upstreamHost }); +} + function upgradeThroughProxy( port: number, host: string, @@ -651,3 +671,198 @@ describe("service proxy forwarded headers", () => { } }); }); + +function closed(emitter: EventEmitter): Promise { + return new Promise((resolve) => { + emitter.once("close", () => resolve()); + }); +} + +async function within(promise: Promise, ms: number, failure: string): Promise { + let timer: NodeJS.Timeout | undefined; + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(`${failure} after ${ms}ms`)), ms); + }); + try { + return await Promise.race([promise, timeout]); + } finally { + clearTimeout(timer); + } +} + +function requestThroughProxy( + fixture: ProxiedServiceFixture, + requestPath: string, +): http.ClientRequest { + return http.get({ + hostname: "127.0.0.1", + port: fixture.daemonPort, + path: requestPath, + headers: { host: fixture.hostname }, + }); +} + +const STRAY_WARNING_GRACE_MS = 100; + +function createUpgradeUpstream(onUpgrade: (socket: Duplex) => void): http.Server { + const upstream = http.createServer(); + upstream.on("upgrade", (_req, socket) => { + socket.on("error", () => socket.destroy()); + socket.on("end", () => socket.end()); + socket.write( + "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n", + ); + onUpgrade(socket); + }); + return upstream; +} + +function openUpgradeThroughProxy(fixture: ProxiedServiceFixture): Promise { + return new Promise((resolve, reject) => { + const socket = net.connect({ host: "127.0.0.1", port: fixture.daemonPort }, () => { + const lines = [ + "GET /ws HTTP/1.1", + `Host: ${fixture.hostname}`, + "Upgrade: websocket", + "Connection: Upgrade", + "", + "", + ]; + socket.write(lines.join("\r\n")); + }); + let raw = ""; + socket.on("data", (chunk: Buffer) => { + raw += chunk.toString(); + if (raw.includes("\r\n\r\n")) resolve(socket); + }); + socket.on("close", () => reject(new Error("upgrade closed before the 101 arrived"))); + socket.on("error", reject); + }); +} + +describe("service proxy connection teardown", () => { + it("closes the upstream connection when a client leaves an event stream", async () => { + const upstreamResponse = Promise.withResolvers(); + const upstream = http.createServer((_req, res) => { + res.writeHead(200, { "content-type": "text/event-stream" }); + res.write("data: open\n\n"); + upstreamResponse.resolve(res); + }); + const fixture = await startProxiedService(upstream); + try { + const request = requestThroughProxy(fixture, "/events"); + const [response] = (await once(request, "response")) as [http.IncomingMessage]; + await once(response, "data"); + const upstreamClosed = closed(await upstreamResponse.promise); + + request.destroy(); + + await within(upstreamClosed, 1000, "the service still holds the event stream"); + } finally { + await fixture.close(); + } + }, 5000); + + it("aborts the upstream request when a client leaves before the response starts", async () => { + const warnings: unknown[] = []; + const proxyLogger = pino( + { level: "warn" }, + { write: (line: string) => warnings.push(JSON.parse(line)) }, + ); + const upstreamResponse = Promise.withResolvers(); + const upstream = http.createServer((_req, res) => { + upstreamResponse.resolve(res); + }); + const fixture = await startProxiedService(upstream, { logger: proxyLogger }); + try { + const request = requestThroughProxy(fixture, "/slow"); + request.on("error", () => undefined); + const upstreamClosed = closed(await upstreamResponse.promise); + + request.destroy(); + + await within(upstreamClosed, 1000, "the service still holds the abandoned request"); + await new Promise((resolve) => setTimeout(resolve, STRAY_WARNING_GRACE_MS)); + expect(warnings).toEqual([]); + } finally { + await fixture.close(); + } + }, 5000); + + it("lets a complete response finish without resetting the service", async () => { + const upstreamResponse = Promise.withResolvers(); + const upstream = http.createServer((_req, res) => { + res.writeHead(200, { "content-type": "text/plain" }); + res.write("first;"); + upstreamResponse.resolve(res); + }); + const fixture = await startProxiedService(upstream); + try { + const response = httpGet(fixture.daemonPort, fixture.hostname, { path: "/" }); + const res = await upstreamResponse.promise; + const upstreamClosed = closed(res); + + res.end("second"); + + await expect(response).resolves.toEqual({ status: 200, body: "first;second" }); + await within(upstreamClosed, 1000, "the service response never closed"); + expect(res.writableFinished).toBe(true); + } finally { + await fixture.close(); + } + }, 5000); + + it("closes the upstream WebSocket when the daemon drops the client socket", async () => { + const upstreamSocket = Promise.withResolvers(); + const fixture = await startProxiedService( + createUpgradeUpstream((socket) => upstreamSocket.resolve(socket)), + ); + const daemonSide = Promise.withResolvers(); + fixture.daemon.on("upgrade", (_req, socket) => daemonSide.resolve(socket)); + try { + const client = await openUpgradeThroughProxy(fixture); + const upstreamClosed = closed(await upstreamSocket.promise); + + (await daemonSide.promise).destroy(); + + await within(upstreamClosed, 1000, "the service still holds the WebSocket"); + client.destroy(); + } finally { + await fixture.close(); + } + }, 5000); + + it("closes the upstream WebSocket when the client disconnects", async () => { + const upstreamSocket = Promise.withResolvers(); + const fixture = await startProxiedService( + createUpgradeUpstream((socket) => upstreamSocket.resolve(socket)), + ); + try { + const client = await openUpgradeThroughProxy(fixture); + const upstreamClosed = closed(await upstreamSocket.promise); + + client.destroy(); + + await within(upstreamClosed, 1000, "the service still holds the WebSocket"); + } finally { + await fixture.close(); + } + }, 5000); + + it("closes the client WebSocket when the service closes it", async () => { + const upstreamSocket = Promise.withResolvers(); + const fixture = await startProxiedService( + createUpgradeUpstream((socket) => upstreamSocket.resolve(socket)), + ); + try { + const client = await openUpgradeThroughProxy(fixture); + const clientClosed = closed(client); + + (await upstreamSocket.promise).destroy(); + + await within(clientClosed, 1000, "the client WebSocket is still open"); + } finally { + await fixture.close(); + } + }, 5000); +}); diff --git a/packages/server/src/server/service-proxy.ts b/packages/server/src/server/service-proxy.ts index ef93a27f9a..24bfd80c93 100644 --- a/packages/server/src/server/service-proxy.ts +++ b/packages/server/src/server/service-proxy.ts @@ -342,7 +342,14 @@ function proxyHttpRequest({ proxyRes.pipe(res, { end: true }); }, ); + let clientClosed = false; + res.on("close", () => { + if (res.writableFinished) return; + clientClosed = true; + proxyReq.destroy(); + }); proxyReq.on("error", (err) => { + if (clientClosed) return; logger.warn( { err, hostname: route.hostname, port: route.port }, "Service proxy: upstream unreachable", @@ -399,6 +406,9 @@ function proxyUpgradeRequest({ socket.on("error", () => { targetSocket.destroy(); }); + socket.on("close", () => { + targetSocket.destroy(); + }); } function sameRouteOwner(left: ServiceProxyRouteEntry, right: ServiceProxyRouteEntry): boolean { diff --git a/packages/server/src/server/terminal-activity-route.test.ts b/packages/server/src/server/terminal-activity-route.test.ts index afc97d77e8..e047c56ab9 100644 --- a/packages/server/src/server/terminal-activity-route.test.ts +++ b/packages/server/src/server/terminal-activity-route.test.ts @@ -95,7 +95,13 @@ it("accepts terminalId and token reports through the route into the tracker", as command: process.execPath, args: [ "-e", - `require("node:fs").writeFileSync(${JSON.stringify(envPath)}, JSON.stringify({ terminalId: process.env.PASEO_TERMINAL_ID, token: process.env.PASEO_ACTIVITY_TOKEN, url: process.env.PASEO_TERMINAL_ACTIVITY_URL })); setInterval(() => {}, 1000);`, + ` + const fs = require("node:fs"); + const envPath = ${JSON.stringify(envPath)}; + fs.writeFileSync(envPath + ".tmp", JSON.stringify({ terminalId: process.env.PASEO_TERMINAL_ID, token: process.env.PASEO_ACTIVITY_TOKEN, url: process.env.PASEO_TERMINAL_ACTIVITY_URL })); + fs.renameSync(envPath + ".tmp", envPath); + setInterval(() => {}, 1000); + `, ], }); await waitForCondition(() => existsSync(envPath), 10000);