From 6c080b3ecc320b17d20507f31713e89eaafb42d8 Mon Sep 17 00:00:00 2001 From: Lakshmi Tanmay Date: Sat, 10 Oct 2026 00:02:38 +0530 Subject: [PATCH] fix(server): require the bootstrap envelope before startup --- apps/server/src/bootstrap.test.ts | 169 +++++++++++++++++++++-------- apps/server/src/bootstrap.ts | 80 +++++++------- apps/server/src/cli/config.test.ts | 38 +++++++ apps/server/src/cli/config.ts | 5 +- 4 files changed, 209 insertions(+), 83 deletions(-) diff --git a/apps/server/src/bootstrap.test.ts b/apps/server/src/bootstrap.test.ts index d532f71c9fb8..7292bb188d48 100644 --- a/apps/server/src/bootstrap.test.ts +++ b/apps/server/src/bootstrap.test.ts @@ -1,12 +1,10 @@ // @effect-diagnostics nodeBuiltinImport:off import * as NodeFS from "node:fs"; -import * as NodePath from "node:path"; -import * as NodeChildProcess from "node:child_process"; +import * as NodeStream from "node:stream"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import * as FileSystem from "effect/FileSystem"; import * as Schema from "effect/Schema"; -import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as TestClock from "effect/testing/TestClock"; @@ -15,22 +13,27 @@ import { HostProcessPlatform } from "@t3tools/shared/hostProcess"; import { BootstrapEnvelopeDecodeError, + BootstrapEnvelopeMissingError, + BootstrapEnvelopeReadError, + BootstrapEnvelopeTimeoutError, BootstrapFdStatError, BootstrapInputStreamOpenError, readBootstrapEnvelope, } from "./bootstrap.ts"; -import { assertNone, assertSome } from "@effect/vitest/utils"; const openSyncInterceptor = vi.hoisted(() => ({ failPath: null as string | null, errorCode: "ENXIO", })); const fstatSyncInterceptor = vi.hoisted(() => ({ failFd: null as number | null })); +const readStreamInterceptor = vi.hoisted(() => ({ stream: null as NodeStream.Readable | null })); vi.mock("node:fs", async (importOriginal) => { const actual = await importOriginal(); return { ...actual, + createReadStream: (...args: Parameters) => + readStreamInterceptor.stream ?? actual.createReadStream(...args), openSync: (...args: Parameters) => { const [filePath, flags] = args; if ( @@ -76,6 +79,22 @@ const openBootstrapInputFd = (filePath: string) => const TestEnvelopeSchema = Schema.Struct({ mode: Schema.String }); const encodeTestEnvelopeSchema = Schema.encodeEffect(Schema.fromJsonString(TestEnvelopeSchema)); +// Control pipe delivery independently of the virtual timeout, on every host. +const openControlledBootstrapInput = Effect.acquireRelease( + Effect.sync(() => { + const fd = NodeFS.openSync(nullDevice, "r"); + const stream = new NodeStream.PassThrough(); + readStreamInterceptor.stream = stream; + return { fd, stream }; + }), + ({ fd, stream }) => + Effect.sync(() => { + readStreamInterceptor.stream = null; + stream.destroy(); + closeIfOpen(fd); + }), +); + it.layer(NodeServices.layer)("readBootstrapEnvelope", (it) => { it.effect("reads a bootstrap envelope from a provided fd", () => Effect.gen(function* () { @@ -90,7 +109,7 @@ it.layer(NodeServices.layer)("readBootstrapEnvelope", (it) => { const fd = yield* openBootstrapInputFd(filePath); const payload = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd, { timeoutMs: 100 }); - assertSome(payload, { + assert.deepEqual(payload, { mode: "desktop", }); }), @@ -117,7 +136,7 @@ it.layer(NodeServices.layer)("readBootstrapEnvelope", (it) => { const payload = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd, { timeoutMs: 100, }).pipe(Effect.provideService(HostProcessPlatform, "linux")); - assertSome(payload, { + assert.deepEqual(payload, { mode: "desktop", }); } finally { @@ -159,13 +178,16 @@ it.layer(NodeServices.layer)("readBootstrapEnvelope", (it) => { }), ); - it.effect("returns none when the fd is unavailable", () => + it.effect("fails when the explicitly provided fd is unavailable", () => Effect.gen(function* () { const fd = NodeFS.openSync(nullDevice, "r"); NodeFS.closeSync(fd); - const payload = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd, { timeoutMs: 100 }); - assertNone(payload); + const error = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd, { + timeoutMs: 100, + }).pipe(Effect.flip); + assert.instanceOf(error, BootstrapFdStatError); + assert.equal(error.fd, fd); }), ); @@ -213,43 +235,104 @@ it.layer(NodeServices.layer)("readBootstrapEnvelope", (it) => { }), ); - // Needs a FIFO, which mkfifo creates; Windows has neither. - it.effect.skipIf(windowsHost)( - "returns none when the bootstrap read times out before any value arrives", - () => - Effect.gen(function* () { - const fs = yield* FileSystem.FileSystem; - const tempDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-bootstrap-" }); - const fifoPath = NodePath.join(tempDir, "bootstrap.pipe"); - - yield* Effect.sync(() => NodeChildProcess.execFileSync("mkfifo", [fifoPath])); - - const _writer = yield* Effect.acquireRelease( - Effect.sync(() => - NodeChildProcess.spawn("sh", ["-c", 'exec 3>"$1"; sleep 60', "sh", fifoPath], { - stdio: ["ignore", "ignore", "ignore"], - }), - ), - (writer) => - Effect.sync(() => { - writer.kill("SIGKILL"); - }), - ); + it.effect("accepts an envelope arriving after the former one-second deadline", () => + Effect.gen(function* () { + const { fd, stream } = yield* openControlledBootstrapInput; + const fiber = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd).pipe( + Effect.provideService(HostProcessPlatform, "win32"), + Effect.forkScoped, + ); - const fd = yield* Effect.acquireRelease( - Effect.sync(() => NodeFS.openSync(fifoPath, "r")), - (fd) => Effect.sync(() => closeIfOpen(fd)), - ); + yield* Effect.yieldNow; + yield* TestClock.adjust(1_001); + assert.isUndefined(fiber.pollUnsafe()); + stream.end('{"mode":"desktop"}\n'); - const fiber = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd, { - timeoutMs: 100, - }).pipe(Effect.forkScoped); + const payload = yield* Fiber.join(fiber); + assert.deepEqual(payload, { mode: "desktop" }); + assert.isTrue(stream.destroyed); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("fails with the fd and deadline when no envelope arrives", () => + Effect.gen(function* () { + const { fd, stream } = yield* openControlledBootstrapInput; + const fiber = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd).pipe( + Effect.provideService(HostProcessPlatform, "win32"), + Effect.flip, + Effect.forkScoped, + ); + + yield* Effect.yieldNow; + yield* TestClock.adjust(30_000); + + const error = yield* Fiber.join(fiber); + assert.instanceOf(error, BootstrapEnvelopeTimeoutError); + assert.equal(error.fd, fd); + assert.equal(error.timeoutMs, 30_000); + assert.equal( + error.message, + `Timed out waiting for bootstrap envelope from file descriptor ${fd} after 30000 ms.`, + ); + assert.isTrue(stream.destroyed); + }).pipe(Effect.provide(TestClock.layer())), + ); - yield* Effect.yieldNow; - yield* TestClock.adjust(Duration.millis(100)); + it.effect("fails when the pipe closes without an envelope", () => + Effect.gen(function* () { + const { fd, stream } = yield* openControlledBootstrapInput; + const fiber = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd).pipe( + Effect.provideService(HostProcessPlatform, "win32"), + Effect.flip, + Effect.forkScoped, + ); + + yield* Effect.yieldNow; + stream.end(); + + const error = yield* Fiber.join(fiber); + assert.instanceOf(error, BootstrapEnvelopeMissingError); + assert.equal(error.fd, fd); + assert.equal( + error.message, + `Bootstrap input on file descriptor ${fd} closed without an envelope.`, + ); + assert.isTrue(stream.destroyed); + }), + ); - const payload = yield* Fiber.join(fiber); - assertNone(payload); - }).pipe(Effect.provide(TestClock.layer())), + it.effect("preserves read errors instead of treating an unavailable pipe as optional", () => + Effect.gen(function* () { + const { fd, stream } = yield* openControlledBootstrapInput; + const cause = Object.assign(new Error("closed fd"), { code: "EBADF" }); + const fiber = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd).pipe( + Effect.provideService(HostProcessPlatform, "win32"), + Effect.flip, + Effect.forkScoped, + ); + + yield* Effect.yieldNow; + stream.destroy(cause); + + const error = yield* Fiber.join(fiber); + assert.instanceOf(error, BootstrapEnvelopeReadError); + assert.equal(error.fd, fd); + assert.equal(error.cause, cause); + }), + ); + + it.effect("destroys the bootstrap stream when the read is interrupted", () => + Effect.gen(function* () { + const { fd, stream } = yield* openControlledBootstrapInput; + const fiber = yield* readBootstrapEnvelope(TestEnvelopeSchema, fd).pipe( + Effect.provideService(HostProcessPlatform, "win32"), + Effect.forkScoped, + ); + + yield* Effect.yieldNow; + yield* Fiber.interrupt(fiber); + + assert.isTrue(stream.destroyed); + }), ); }); diff --git a/apps/server/src/bootstrap.ts b/apps/server/src/bootstrap.ts index 71dae0dc3096..aa9a603f28b0 100644 --- a/apps/server/src/bootstrap.ts +++ b/apps/server/src/bootstrap.ts @@ -5,7 +5,6 @@ import * as NodeReadline from "node:readline"; import type * as NodeStream from "node:stream"; import * as Effect from "effect/Effect"; -import * as Option from "effect/Option"; import * as Predicate from "effect/Predicate"; import * as Result from "effect/Result"; import * as Schema from "effect/Schema"; @@ -63,11 +62,31 @@ export class BootstrapEnvelopeDecodeError extends Schema.TaggedError()( + "BootstrapEnvelopeMissingError", + { fd: Schema.Number }, +) { + override get message(): string { + return `Bootstrap input on file descriptor ${this.fd} closed without an envelope.`; + } +} + +export class BootstrapEnvelopeTimeoutError extends Schema.TaggedError()( + "BootstrapEnvelopeTimeoutError", + { fd: Schema.Number, timeoutMs: Schema.Number }, +) { + override get message(): string { + return `Timed out waiting for bootstrap envelope from file descriptor ${this.fd} after ${this.timeoutMs} ms.`; + } +} + export const BootstrapError = Schema.Union([ BootstrapFdStatError, BootstrapInputStreamOpenError, BootstrapEnvelopeReadError, BootstrapEnvelopeDecodeError, + BootstrapEnvelopeMissingError, + BootstrapEnvelopeTimeoutError, ]); export type BootstrapError = typeof BootstrapError.Type; @@ -77,17 +96,21 @@ export const readBootstrapEnvelope = Effect.fn("readBootstrapEnvelope")(function options?: { timeoutMs?: number; }, -): Effect.fn.Return, BootstrapError> { - const fdReady = yield* isFdReady(fd); - if (!fdReady) return Option.none(); +): Effect.fn.Return { + yield* Effect.try({ + try: () => NodeFS.fstatSync(fd), + catch: (cause) => new BootstrapFdStatError({ fd, cause }), + }); const stream = yield* makeBootstrapInputStream(fd); - const timeoutMs = options?.timeoutMs ?? 1000; + // An explicit bootstrap fd is required startup input. Allow cold reads time + // to complete, but fail before the desktop's one-minute readiness deadline. + const timeoutMs = options?.timeoutMs ?? 30_000; return yield* Effect.callback< - Option.Option, - BootstrapEnvelopeReadError | BootstrapEnvelopeDecodeError + A, + BootstrapEnvelopeReadError | BootstrapEnvelopeDecodeError | BootstrapEnvelopeMissingError >((resume) => { const input = NodeReadline.createInterface({ input: stream, @@ -95,19 +118,18 @@ export const readBootstrapEnvelope = Effect.fn("readBootstrapEnvelope")(function }); const cleanup = () => { - stream.removeListener("error", handleError); + input.removeListener("error", handleError); input.removeListener("line", handleLine); input.removeListener("close", handleClose); input.close(); stream.destroy(); }; + const finish = (result: Parameters[0]) => + resume(result.pipe(Effect.ensuring(Effect.sync(cleanup)))); + const handleError = (error: Error) => { - if (isUnavailableBootstrapFdError(error)) { - resume(Effect.succeedNone); - return; - } - resume( + finish( Effect.fail( new BootstrapEnvelopeReadError({ fd, @@ -120,9 +142,9 @@ export const readBootstrapEnvelope = Effect.fn("readBootstrapEnvelope")(function const handleLine = (line: string) => { const parsed = decodeJsonResult(schema)(line); if (Result.isSuccess(parsed)) { - resume(Effect.succeedSome(parsed.success)); + finish(Effect.succeed(parsed.success)); } else { - resume( + finish( Effect.fail( new BootstrapEnvelopeDecodeError({ fd, @@ -134,37 +156,21 @@ export const readBootstrapEnvelope = Effect.fn("readBootstrapEnvelope")(function }; const handleClose = () => { - resume(Effect.succeedNone); + finish(Effect.fail(new BootstrapEnvelopeMissingError({ fd }))); }; - stream.once("error", handleError); + input.once("error", handleError); input.once("line", handleLine); input.once("close", handleClose); return Effect.sync(cleanup); - }).pipe(Effect.timeoutOption(timeoutMs), Effect.map(Option.flatten)); -}); - -const isUnavailableBootstrapFdError = Predicate.compose( - Predicate.hasProperty("code"), - (_) => _.code === "EBADF" || _.code === "ENOENT", -); - -const isFdReady = (fd: number) => - Effect.try({ - try: () => NodeFS.fstatSync(fd), - catch: (error) => - new BootstrapFdStatError({ - fd, - cause: error, - }), }).pipe( - Effect.as(true), - Effect.catchTags({ - BootstrapFdStatError: (error) => - isUnavailableBootstrapFdError(error.cause) ? Effect.succeed(false) : Effect.fail(error), + Effect.timeoutOrElse({ + duration: timeoutMs, + orElse: () => Effect.fail(new BootstrapEnvelopeTimeoutError({ fd, timeoutMs })), }), ); +}); const makeBootstrapInputStream = (fd: number) => Effect.gen(function* () { diff --git a/apps/server/src/cli/config.test.ts b/apps/server/src/cli/config.test.ts index 04f6c3ca7a8c..68b8b93743e4 100644 --- a/apps/server/src/cli/config.test.ts +++ b/apps/server/src/cli/config.test.ts @@ -20,6 +20,7 @@ import * as NetService from "@t3tools/shared/Net"; import { DEFAULT_SIGNAL_EXPORT } from "@t3tools/shared/observability"; import * as OtelEnvironment from "@t3tools/shared/otelEnvironment"; import * as NodeServices from "@effect/platform-node/NodeServices"; +import { BootstrapFdStatError } from "../bootstrap.ts"; import { deriveServerPaths } from "../config.ts"; import { resolveServerConfig } from "./config.ts"; @@ -1057,6 +1058,43 @@ it.layer(NodeServices.layer)("cli config resolution", (it) => { tailscaleServePort: Option.none(), }); + it.effect.each(["flag", "env"] as const)( + "rejects an unavailable bootstrap fd from %s before creating server state", + (source) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-cli-bootstrap-required-" }); + const baseDir = path.join(root, "server-home"); + const fd = 2_147_483_647; + + const error = yield* resolveServerConfig( + { + ...minimalWebFlags(baseDir), + mode: Option.none(), + bootstrapFd: source === "flag" ? Option.some(fd) : Option.none(), + }, + Option.none(), + ).pipe( + Effect.provide( + Layer.mergeAll( + NetService.layer, + ConfigProvider.layer( + ConfigProvider.fromEnv({ + env: source === "env" ? { T3CODE_BOOTSTRAP_FD: String(fd) } : {}, + }), + ), + ), + ), + Effect.flip, + ); + + assert.instanceOf(error, BootstrapFdStatError); + assert.equal(error.fd, fd); + expect(yield* fs.exists(baseDir)).toBe(false); + }), + ); + it.effect( "resolves each signal's endpoint through T3CODE_OTLP_*_URL, an OTEL endpoint, the bootstrap envelope, and persisted Settings, in that order", () => diff --git a/apps/server/src/cli/config.ts b/apps/server/src/cli/config.ts index 841962f0dc8e..c441e7293be7 100644 --- a/apps/server/src/cli/config.ts +++ b/apps/server/src/cli/config.ts @@ -279,11 +279,10 @@ export const resolveServerConfig = ( tailscaleServePort: flags.tailscaleServePort ?? Option.none(), } satisfies CliServerFlags; const bootstrapFd = Option.getOrUndefined(normalizedFlags.bootstrapFd) ?? env.bootstrapFd; - const bootstrapEnvelope = + const bootstrap = bootstrapFd !== undefined ? yield* readBootstrapEnvelope(DesktopBackendBootstrap, bootstrapFd) - : Option.none(); - const bootstrap = Option.getOrUndefined(bootstrapEnvelope); + : undefined; const mode: ServerConfig.RuntimeMode = Option.getOrElse( resolveOptionPrecedence(