authorgravatar for saiansh2525@gmail.comYash Singh <saiansh2525@gmail.com> 2026-10-08 22:30:05-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-08 22:30:05-05:00
logc9fa1367caff4cd6595386814e1f8bb950d3d3ef
treed41c9eaae68bea87ad147bc40c9e5c3ca37d9c9f
parent42c6623d492850c747b7ba06c1d0f1e1559d0ad5
signature Signed by PGP key B5690EEEBB952194

fix(desktop): cancel backend pipe reads to avoid slow shutdown (#17386)

4 files changed, 120 insertions(+), 61 deletions(-)

apps/server/src/preview/DesktopBrowserChannel.test.ts+55-54
...@@ -1,58 +1,59 @@...@@ -1,58 +1,59 @@
1// @effect-diagnostics nodeBuiltinImport:off - The channel reads real file descriptors.1// @effect-diagnostics nodeBuiltinImport:off - Exercises the real inherited descriptor lifecycle in a subprocess.
2import * as NodeFS from "node:fs";2import * as NodeChildProcess from "node:child_process";
3import * as NodeOS from "node:os";3import * as NodeURL from "node:url";
4import * as NodePath from "node:path";
54
6import * as NodeServices from "@effect/platform-node/NodeServices";5import { expect, it } from "vite-plus/test";
7import { expect, it } from "@effect/vitest";
8import * as Context from "effect/Context";
9import * as Effect from "effect/Effect";
10import * as Exit from "effect/Exit";
11import * as Layer from "effect/Layer";
126
13import * as ServerConfig from "../config.ts";7it("receives desktop messages and exits while the parent keeps both input pipes open", async () => {
14import * as DesktopBrowserChannel from "./DesktopBrowserChannel.ts";8 const child = NodeChildProcess.spawn(
159 process.execPath,
16const key = { threadId: "thread-1", tabId: "tab-1" };10 [NodeURL.fileURLToPath(new URL("./testing/DesktopPipeLifecycle.fixture.ts", import.meta.url))],
1711 { stdio: ["ignore", "pipe", "pipe", "pipe", "pipe", "pipe", "pipe"] },
18/** The channel over two files: what the desktop sent, and where commands go. */
19const channelOver = (events: ReadonlyArray<Record<string, unknown>>) =>
20 Effect.gen(function* () {
21 const directory = yield* Effect.acquireRelease(
22 Effect.sync(() =>
23 NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-desktop-browser-channel-")),
24 ),
25 (directory) => Effect.sync(() => NodeFS.rmSync(directory, { recursive: true, force: true })),
26 );
27 const inbound = NodePath.join(directory, "events.ndjson");
28 NodeFS.writeFileSync(inbound, events.map((event) => `${JSON.stringify(event)}\n`).join(""));
29 const control = yield* Effect.acquireRelease(
30 Effect.sync(() => NodeFS.openSync(NodePath.join(directory, "commands.ndjson"), "w")),
31 (fd) => Effect.sync(() => NodeFS.closeSync(fd)),
32 );
33 const base = yield* ServerConfig.ServerConfig.pipe(
34 Effect.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-desktop-browser-" })),
35 );
36 const config = Layer.succeed(ServerConfig.ServerConfig, {
37 ...base,
38 desktopBrowserFd: NodeFS.openSync(inbound, "r"),
39 desktopBrowserControlFd: control,
40 });
41 // Built in the caller's scope, so the channel outlives this setup.
42 const context = yield* Layer.build(DesktopBrowserChannel.layer.pipe(Layer.provide(config)));
43 return Context.get(context, DesktopBrowserChannel.DesktopBrowserChannel);
44 });
45
46it.layer(NodeServices.layer)("DesktopBrowserChannel", (it) => {
47 it.effect("refuses an endpoint for a tab that is no longer attached", () =>
48 Effect.gen(function* () {
49 const channel = yield* channelOver([
50 { type: "attached", ...key },
51 { type: "detached", ...key },
52 ]);
53 // Whether or not the reader has reached these lines yet, the tab is not attached.
54 const exit = yield* Effect.exit(Effect.scoped(channel.endpoint(key)));
55 expect(Exit.isFailure(exit)).toBe(true);
56 }).pipe(Effect.scoped),
57 );12 );
13 // A failed exit must not leave a test process running. This never fires on success.
14 // @effect-diagnostics-next-line globalTimers:off -- Bounds the native subprocess on failure; success waits for process exit.
15 const watchdog = setTimeout(() => child.kill("SIGKILL"), 8_000);
16 let output = "";
17 let errors = "";
18 let verified = false;
19 child.stderr?.on("data", (chunk: Buffer) => {
20 errors += chunk.toString();
21 });
22 const write = (fd: number, value: Record<string, unknown>) => {
23 const stream = child.stdio[fd];
24 if (!stream || !("write" in stream)) throw new Error(`Missing pipe ${fd}`);
25 stream.write(`${JSON.stringify(value)}\n`);
26 };
27 const key = { threadId: "thread-1", tabId: "tab-1" };
28 child.stdout?.on("data", (chunk: Buffer) => {
29 output += chunk.toString();
30 let end: number;
31 while ((end = output.indexOf("\n")) >= 0) {
32 const line = output.slice(0, end);
33 output = output.slice(end + 1);
34 if (line === "ready") {
35 write(3, { type: "attached", ...key });
36 write(5, { version: 1, type: "desktopTelemetryHello", electronPid: process.pid });
37 } else if (line === "attached") {
38 write(3, { type: "detached", ...key });
39 } else if (line === "verified") {
40 verified = true;
41 }
42 }
43 });
44 try {
45 const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>(
46 (resolve, reject) => {
47 child.once("error", reject);
48 child.once("exit", (code, signal) => resolve({ code, signal }));
49 },
50 );
51 expect(errors).not.toContain("Error");
52 expect(verified).toBe(true);
53 expect(result).toEqual({ code: 0, signal: null });
54 } finally {
55 clearTimeout(watchdog);
56 if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL");
57 for (const stream of child.stdio) stream?.destroy();
58 }
58});59});
apps/server/src/preview/DesktopBrowserChannel.ts+4-2
...@@ -27,7 +27,7 @@ import * as Semaphore from "effect/Semaphore";...@@ -27,7 +27,7 @@ import * as Semaphore from "effect/Semaphore";
27import * as Stream from "effect/Stream";27import * as Stream from "effect/Stream";
28import * as Ndjson from "effect/encoding/Ndjson";28import * as Ndjson from "effect/encoding/Ndjson";
29import * as NodeCrypto from "node:crypto";29import * as NodeCrypto from "node:crypto";
30import * as NodeFS from "node:fs";30import * as NodeNet from "node:net";
3131
32import * as ServerConfig from "../config.ts";32import * as ServerConfig from "../config.ts";
33import { writeAllToFileDescriptor } from "../resourceTelemetry/DesktopTelemetryReceiver.ts";33import { writeAllToFileDescriptor } from "../resourceTelemetry/DesktopTelemetryReceiver.ts";
...@@ -102,8 +102,10 @@ const make = Effect.gen(function* () {...@@ -102,8 +102,10 @@ const make = Effect.gen(function* () {
102 ),102 ),
103 );103 );
104104
105 // Socket reads can be cancelled while the desktop keeps its write end open.
106 // Filesystem reads on this pipe keep Node alive after scope shutdown.
105 const readable = yield* Effect.acquireRelease(107 const readable = yield* Effect.acquireRelease(
106 Effect.sync(() => NodeFS.createReadStream("", { fd: inputFd, autoClose: true })),108 Effect.sync(() => new NodeNet.Socket({ fd: inputFd, readable: true, writable: false })),
107 (stream) => Effect.sync(() => stream.destroy()),109 (stream) => Effect.sync(() => stream.destroy()),
108 );110 );
109 yield* NodeStream.fromReadable<Uint8Array, Error>({111 yield* NodeStream.fromReadable<Uint8Array, Error>({
apps/server/src/preview/testing/DesktopPipeLifecycle.fixture.ts created+57
...@@ -0,0 +1,57 @@
1// @effect-diagnostics nodeBuiltinImport:off - Entry point for the inherited-pipe shutdown regression test.
2import * as NodeRuntime from "@effect/platform-node/NodeRuntime";
3import * as NodeServices from "@effect/platform-node/NodeServices";
4import * as Console from "effect/Console";
5import * as Context from "effect/Context";
6import * as Effect from "effect/Effect";
7import * as Exit from "effect/Exit";
8import * as Fiber from "effect/Fiber";
9import * as Layer from "effect/Layer";
10import * as Stream from "effect/Stream";
11
12import * as ServerSettings from "../../serverSettings.ts";
13
14import * as ServerConfig from "../../config.ts";
15import * as DesktopTelemetryReceiver from "../../resourceTelemetry/DesktopTelemetryReceiver.ts";
16import * as DesktopBrowserChannel from "../DesktopBrowserChannel.ts";
17
18const key = { threadId: "thread-1", tabId: "tab-1" };
19
20Effect.gen(function* () {
21 const base = yield* ServerConfig.ServerConfig.pipe(
22 Effect.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-pipe-lifecycle-" })),
23 );
24 const context = yield* Layer.build(
25 Layer.merge(DesktopBrowserChannel.layer, DesktopTelemetryReceiver.layer).pipe(
26 Layer.provide(ServerSettings.layerTest()),
27 Layer.provide(
28 Layer.succeed(ServerConfig.ServerConfig, {
29 ...base,
30 desktopBrowserFd: 3,
31 desktopBrowserControlFd: 4,
32 desktopTelemetryFd: 5,
33 desktopTelemetryControlFd: 6,
34 }),
35 ),
36 ),
37 );
38 const browser = Context.get(context, DesktopBrowserChannel.DesktopBrowserChannel);
39 const telemetry = Context.get(context, DesktopTelemetryReceiver.DesktopTelemetryReceiver);
40 const health = yield* telemetry.subscribeHealth;
41 const healthy = yield* health.changes.pipe(
42 Stream.filter((value) => value.status === "healthy"),
43 Stream.runHead,
44 Effect.forkScoped,
45 );
46 const detached = yield* browser.detached.pipe(Stream.runHead, Effect.forkScoped);
47 yield* Console.log("ready");
48 if (!(yield* browser.awaitAttached(key, "5 seconds"))) {
49 return yield* Effect.die("The browser attach message was not received.");
50 }
51 yield* Fiber.join(healthy);
52 yield* Console.log("attached");
53 yield* Fiber.join(detached);
54 const endpoint = yield* Effect.exit(Effect.scoped(browser.endpoint(key)));
55 if (Exit.isSuccess(endpoint)) return yield* Effect.die("A detached tab received an endpoint.");
56 yield* Console.log("verified");
57}).pipe(Effect.scoped, Effect.provide(NodeServices.layer), NodeRuntime.runMain);
apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts+4-5
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1// @effect-diagnostics nodeBuiltinImport:off1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeFS from "node:fs";2import * as NodeFS from "node:fs";
3import * as NodeNet from "node:net";
34
4import * as NodeStream from "@effect/platform-node/NodeStream";5import * as NodeStream from "@effect/platform-node/NodeStream";
5import {6import {
...@@ -463,13 +464,11 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make")...@@ -463,13 +464,11 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make")
463464
464 if (config.desktopTelemetryFd !== undefined) {465 if (config.desktopTelemetryFd !== undefined) {
465 const fd = config.desktopTelemetryFd;466 const fd = config.desktopTelemetryFd;
467 // The parent waits for this server before closing its pipe. Socket reads
468 // can be cancelled; filesystem reads would keep process.exit waiting.
466 const readable = yield* Effect.acquireRelease(469 const readable = yield* Effect.acquireRelease(
467 Effect.try({470 Effect.try({
468 try: () =>471 try: () => new NodeNet.Socket({ fd, readable: true, writable: false }),
469 NodeFS.createReadStream("", {
470 fd,
471 autoClose: true,
472 }),
473 catch: (cause) => new DesktopTelemetryStreamFailed({ fd, cause }),472 catch: (cause) => new DesktopTelemetryStreamFailed({ fd, cause }),
474 }),473 }),
475 (stream) =>474 (stream) =>