From c9fa1367caff4cd6595386814e1f8bb950d3d3ef Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Thu, 8 Oct 2026 22:30:05 -0500 Subject: [PATCH] fix(desktop): cancel backend pipe reads to avoid slow shutdown (#17386) --- .../src/preview/DesktopBrowserChannel.test.ts | 109 +++++++++--------- .../src/preview/DesktopBrowserChannel.ts | 6 +- .../testing/DesktopPipeLifecycle.fixture.ts | 57 +++++++++ .../DesktopTelemetryReceiver.ts | 9 +- 4 files changed, 120 insertions(+), 61 deletions(-) create mode 100644 apps/server/src/preview/testing/DesktopPipeLifecycle.fixture.ts diff --git a/apps/server/src/preview/DesktopBrowserChannel.test.ts b/apps/server/src/preview/DesktopBrowserChannel.test.ts index 4fd769ab3fc6542ad72ed60e9740e3f777a00964..59cc5e1701a17a3b0a03155b58385a3bc84e7be6 100644 --- a/apps/server/src/preview/DesktopBrowserChannel.test.ts +++ b/apps/server/src/preview/DesktopBrowserChannel.test.ts @@ -1,58 +1,59 @@ -// @effect-diagnostics nodeBuiltinImport:off - The channel reads real file descriptors. -import * as NodeFS from "node:fs"; -import * as NodeOS from "node:os"; -import * as NodePath from "node:path"; +// @effect-diagnostics nodeBuiltinImport:off - Exercises the real inherited descriptor lifecycle in a subprocess. +import * as NodeChildProcess from "node:child_process"; +import * as NodeURL from "node:url"; -import * as NodeServices from "@effect/platform-node/NodeServices"; -import { expect, it } from "@effect/vitest"; -import * as Context from "effect/Context"; -import * as Effect from "effect/Effect"; -import * as Exit from "effect/Exit"; -import * as Layer from "effect/Layer"; +import { expect, it } from "vite-plus/test"; -import * as ServerConfig from "../config.ts"; -import * as DesktopBrowserChannel from "./DesktopBrowserChannel.ts"; - -const key = { threadId: "thread-1", tabId: "tab-1" }; - -/** The channel over two files: what the desktop sent, and where commands go. */ -const channelOver = (events: ReadonlyArray>) => - Effect.gen(function* () { - const directory = yield* Effect.acquireRelease( - Effect.sync(() => - NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-desktop-browser-channel-")), - ), - (directory) => Effect.sync(() => NodeFS.rmSync(directory, { recursive: true, force: true })), - ); - const inbound = NodePath.join(directory, "events.ndjson"); - NodeFS.writeFileSync(inbound, events.map((event) => `${JSON.stringify(event)}\n`).join("")); - const control = yield* Effect.acquireRelease( - Effect.sync(() => NodeFS.openSync(NodePath.join(directory, "commands.ndjson"), "w")), - (fd) => Effect.sync(() => NodeFS.closeSync(fd)), - ); - const base = yield* ServerConfig.ServerConfig.pipe( - Effect.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-desktop-browser-" })), - ); - const config = Layer.succeed(ServerConfig.ServerConfig, { - ...base, - desktopBrowserFd: NodeFS.openSync(inbound, "r"), - desktopBrowserControlFd: control, - }); - // Built in the caller's scope, so the channel outlives this setup. - const context = yield* Layer.build(DesktopBrowserChannel.layer.pipe(Layer.provide(config))); - return Context.get(context, DesktopBrowserChannel.DesktopBrowserChannel); - }); - -it.layer(NodeServices.layer)("DesktopBrowserChannel", (it) => { - it.effect("refuses an endpoint for a tab that is no longer attached", () => - Effect.gen(function* () { - const channel = yield* channelOver([ - { type: "attached", ...key }, - { type: "detached", ...key }, - ]); - // Whether or not the reader has reached these lines yet, the tab is not attached. - const exit = yield* Effect.exit(Effect.scoped(channel.endpoint(key))); - expect(Exit.isFailure(exit)).toBe(true); - }).pipe(Effect.scoped), +it("receives desktop messages and exits while the parent keeps both input pipes open", async () => { + const child = NodeChildProcess.spawn( + process.execPath, + [NodeURL.fileURLToPath(new URL("./testing/DesktopPipeLifecycle.fixture.ts", import.meta.url))], + { stdio: ["ignore", "pipe", "pipe", "pipe", "pipe", "pipe", "pipe"] }, ); + // A failed exit must not leave a test process running. This never fires on success. + // @effect-diagnostics-next-line globalTimers:off -- Bounds the native subprocess on failure; success waits for process exit. + const watchdog = setTimeout(() => child.kill("SIGKILL"), 8_000); + let output = ""; + let errors = ""; + let verified = false; + child.stderr?.on("data", (chunk: Buffer) => { + errors += chunk.toString(); + }); + const write = (fd: number, value: Record) => { + const stream = child.stdio[fd]; + if (!stream || !("write" in stream)) throw new Error(`Missing pipe ${fd}`); + stream.write(`${JSON.stringify(value)}\n`); + }; + const key = { threadId: "thread-1", tabId: "tab-1" }; + child.stdout?.on("data", (chunk: Buffer) => { + output += chunk.toString(); + let end: number; + while ((end = output.indexOf("\n")) >= 0) { + const line = output.slice(0, end); + output = output.slice(end + 1); + if (line === "ready") { + write(3, { type: "attached", ...key }); + write(5, { version: 1, type: "desktopTelemetryHello", electronPid: process.pid }); + } else if (line === "attached") { + write(3, { type: "detached", ...key }); + } else if (line === "verified") { + verified = true; + } + } + }); + try { + const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>( + (resolve, reject) => { + child.once("error", reject); + child.once("exit", (code, signal) => resolve({ code, signal })); + }, + ); + expect(errors).not.toContain("Error"); + expect(verified).toBe(true); + expect(result).toEqual({ code: 0, signal: null }); + } finally { + clearTimeout(watchdog); + if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); + for (const stream of child.stdio) stream?.destroy(); + } }); diff --git a/apps/server/src/preview/DesktopBrowserChannel.ts b/apps/server/src/preview/DesktopBrowserChannel.ts index 99f89f3544c362a8008b8ab68bf6c5ce9ad064f8..000fe950802744ce56147fe28f1082a22d85cc92 100644 --- a/apps/server/src/preview/DesktopBrowserChannel.ts +++ b/apps/server/src/preview/DesktopBrowserChannel.ts @@ -27,7 +27,7 @@ import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as Ndjson from "effect/encoding/Ndjson"; import * as NodeCrypto from "node:crypto"; -import * as NodeFS from "node:fs"; +import * as NodeNet from "node:net"; import * as ServerConfig from "../config.ts"; import { writeAllToFileDescriptor } from "../resourceTelemetry/DesktopTelemetryReceiver.ts"; @@ -102,8 +102,10 @@ const make = Effect.gen(function* () { ), ); + // Socket reads can be cancelled while the desktop keeps its write end open. + // Filesystem reads on this pipe keep Node alive after scope shutdown. const readable = yield* Effect.acquireRelease( - Effect.sync(() => NodeFS.createReadStream("", { fd: inputFd, autoClose: true })), + Effect.sync(() => new NodeNet.Socket({ fd: inputFd, readable: true, writable: false })), (stream) => Effect.sync(() => stream.destroy()), ); yield* NodeStream.fromReadable({ diff --git a/apps/server/src/preview/testing/DesktopPipeLifecycle.fixture.ts b/apps/server/src/preview/testing/DesktopPipeLifecycle.fixture.ts new file mode 100644 index 0000000000000000000000000000000000000000..b9fee70bf39fc2665207562b112b798dff255bd8 --- /dev/null +++ b/apps/server/src/preview/testing/DesktopPipeLifecycle.fixture.ts @@ -0,0 +1,57 @@ +// @effect-diagnostics nodeBuiltinImport:off - Entry point for the inherited-pipe shutdown regression test. +import * as NodeRuntime from "@effect/platform-node/NodeRuntime"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Console from "effect/Console"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import * as ServerSettings from "../../serverSettings.ts"; + +import * as ServerConfig from "../../config.ts"; +import * as DesktopTelemetryReceiver from "../../resourceTelemetry/DesktopTelemetryReceiver.ts"; +import * as DesktopBrowserChannel from "../DesktopBrowserChannel.ts"; + +const key = { threadId: "thread-1", tabId: "tab-1" }; + +Effect.gen(function* () { + const base = yield* ServerConfig.ServerConfig.pipe( + Effect.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-pipe-lifecycle-" })), + ); + const context = yield* Layer.build( + Layer.merge(DesktopBrowserChannel.layer, DesktopTelemetryReceiver.layer).pipe( + Layer.provide(ServerSettings.layerTest()), + Layer.provide( + Layer.succeed(ServerConfig.ServerConfig, { + ...base, + desktopBrowserFd: 3, + desktopBrowserControlFd: 4, + desktopTelemetryFd: 5, + desktopTelemetryControlFd: 6, + }), + ), + ), + ); + const browser = Context.get(context, DesktopBrowserChannel.DesktopBrowserChannel); + const telemetry = Context.get(context, DesktopTelemetryReceiver.DesktopTelemetryReceiver); + const health = yield* telemetry.subscribeHealth; + const healthy = yield* health.changes.pipe( + Stream.filter((value) => value.status === "healthy"), + Stream.runHead, + Effect.forkScoped, + ); + const detached = yield* browser.detached.pipe(Stream.runHead, Effect.forkScoped); + yield* Console.log("ready"); + if (!(yield* browser.awaitAttached(key, "5 seconds"))) { + return yield* Effect.die("The browser attach message was not received."); + } + yield* Fiber.join(healthy); + yield* Console.log("attached"); + yield* Fiber.join(detached); + const endpoint = yield* Effect.exit(Effect.scoped(browser.endpoint(key))); + if (Exit.isSuccess(endpoint)) return yield* Effect.die("A detached tab received an endpoint."); + yield* Console.log("verified"); +}).pipe(Effect.scoped, Effect.provide(NodeServices.layer), NodeRuntime.runMain); diff --git a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts index cb9bee58f7cd3466fd5bf8c892f7a1b0dda646e0..01a6be1c86f319ec7eaa900bf93d2ee86ecffb11 100644 --- a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts +++ b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts @@ -1,5 +1,6 @@ // @effect-diagnostics nodeBuiltinImport:off import * as NodeFS from "node:fs"; +import * as NodeNet from "node:net"; import * as NodeStream from "@effect/platform-node/NodeStream"; import { @@ -463,13 +464,11 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make") if (config.desktopTelemetryFd !== undefined) { const fd = config.desktopTelemetryFd; + // The parent waits for this server before closing its pipe. Socket reads + // can be cancelled; filesystem reads would keep process.exit waiting. const readable = yield* Effect.acquireRelease( Effect.try({ - try: () => - NodeFS.createReadStream("", { - fd, - autoClose: true, - }), + try: () => new NodeNet.Socket({ fd, readable: true, writable: false }), catch: (cause) => new DesktopTelemetryStreamFailed({ fd, cause }), }), (stream) => -- 2.54.0