Skip to content

fix(server): require the bootstrap envelope before startup #17568

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
voltcrash wants to merge 1 commit into
base: main
Choose a base branch
from
+209 −83
Open

fix(server): require the bootstrap envelope before startup #17568

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 126 additions & 43 deletions apps/server/src/bootstrap.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";
Expand All @@ -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<typeof import("node:fs")>();
return {
...actual,
createReadStream: (...args: Parameters<typeof actual.createReadStream>) =>
readStreamInterceptor.stream ?? actual.createReadStream(...args),
openSync: (...args: Parameters<typeof actual.openSync>) => {
const [filePath, flags] = args;
if (
Expand Down Expand Up @@ -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* () {
Expand All @@ -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",
});
}),
Expand All @@ -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 {
Expand Down Expand Up @@ -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);
}),
);

Expand Down Expand Up @@ -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);
}),
);
});
80 changes: 43 additions & 37 deletions apps/server/src/bootstrap.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -63,11 +62,31 @@ export class BootstrapEnvelopeDecodeError extends Schema.TaggedError<BootstrapEn
}
}

export class BootstrapEnvelopeMissingError extends Schema.TaggedError<BootstrapEnvelopeMissingError>()(
"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>()(
"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;

Expand All @@ -77,37 +96,40 @@ export const readBootstrapEnvelope = Effect.fn("readBootstrapEnvelope")(function
options?: {
timeoutMs?: number;
},
): Effect.fn.Return<Option.Option<A>, BootstrapError> {
const fdReady = yield* isFdReady(fd);
if (!fdReady) return Option.none();
): Effect.fn.Return<A, BootstrapError> {
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<A>,
BootstrapEnvelopeReadError | BootstrapEnvelopeDecodeError
A,
BootstrapEnvelopeReadError | BootstrapEnvelopeDecodeError | BootstrapEnvelopeMissingError
>((resume) => {
const input = NodeReadline.createInterface({
input: stream,
crlfDelay: Infinity,
});

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<typeof resume>[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,
Expand All @@ -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,
Expand All @@ -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* () {
Expand Down
Loading
Loading