Skip to content
2 changes: 2 additions & 0 deletions apps/desktop/src/backend/DesktopBackendConfiguration.ts
Original file line number Diff line number Diff line change
Expand Up @@ -513,6 +513,8 @@ const resolvePrimaryStartConfig = Effect.fn("desktop.backendConfiguration.resolv
desktopTelemetryControlFd: 5,
desktopBrowserFd: 6,
desktopBrowserControlFd: 7,
desktopStopFd: 8,
desktopStopAckFd: 9,
...Option.match(input.resourceMonitorPath, {
onNone: () => ({}),
onSome: (resourceMonitorPath) => ({ resourceMonitorPath }),
Expand Down
74 changes: 74 additions & 0 deletions apps/desktop/src/backend/DesktopBackendManager.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import * as Deferred from "effect/Deferred";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
Expand Down Expand Up @@ -319,6 +320,79 @@ describe("DesktopBackendManager", () => {
),
);

const stopNoticeSpawner = (input: {
readonly events: Array<string>;
readonly acknowledge: boolean;
}) =>
Layer.succeed(
ChildProcessSpawner.ChildProcessSpawner,
ChildProcessSpawner.make((command) =>
Effect.gen(function* () {
const scope = yield* Scope.Scope;
yield* Scope.addFinalizer(
scope,
Effect.sync(() => input.events.push("killed")),
);
const ack = yield* Deferred.make<void>();
const stopFd =
command._tag === "StandardCommand" ? command.options.additionalFds?.fd8 : undefined;
if (stopFd?.type === "input" && stopFd.stream) {
yield* stopFd.stream.pipe(
Stream.runDrain,
Effect.andThen(Effect.sync(() => input.events.push("stop notice"))),
Effect.andThen(input.acknowledge ? Deferred.succeed(ack, undefined) : Effect.void),
Effect.forkIn(scope),
);
}
return makeProcess({
exitCode: Effect.never,
getOutputFd: (fd) =>
fd === 9 ? Stream.fromEffectDrain(Deferred.await(ack)) : Stream.empty,
});
}),
),
);

const runWithStopNotice = (spawnerLayer: Layer.Layer<ChildProcessSpawner.ChildProcessSpawner>) =>
Effect.gen(function* () {
const scope = yield* Scope.make();
yield* DesktopBackendManager.runBackendProcess({
...baseConfig,
bootstrap: { ...baseConfig.bootstrap, desktopStopFd: 8, desktopStopAckFd: 9 },
desktopTelemetryStream: Stream.empty,
}).pipe(
Scope.provide(scope),
Effect.provide(Layer.merge(spawnerLayer, layerHealthyHttpClient)),
Effect.forkChild({ startImmediately: true }),
);
return scope;
});

it.effect("tells the backend it is stopping and waits for its ack before killing it", () =>
Effect.gen(function* () {
const events: Array<string> = [];
const scope = yield* runWithStopNotice(stopNoticeSpawner({ events, acknowledge: true }));
assert.deepEqual(events, []);

yield* Scope.close(scope, Exit.void);
assert.deepEqual(events, ["stop notice", "killed"]);
}),
);

it.effect("kills a backend that never acknowledges the stop notice", () =>
Effect.gen(function* () {
const events: Array<string> = [];
const scope = yield* runWithStopNotice(stopNoticeSpawner({ events, acknowledge: false }));

const closing = yield* Scope.close(scope, Exit.void).pipe(
Effect.forkChild({ startImmediately: true }),
);
yield* TestClock.adjust("1 second");
yield* Fiber.join(closing);
assert.deepEqual(events, ["stop notice", "killed"]);
}),
);

it.effect("reports a SIGKILL exit without an exit code", () =>
Effect.gen(function* () {
const exitCause = PlatformError.systemError({
Expand Down
35 changes: 32 additions & 3 deletions apps/desktop/src/backend/DesktopBackendManager.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import * as Brand from "effect/Brand";
import * as Cause from "effect/Cause";
import * as Deferred from "effect/Deferred";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
Expand Down Expand Up @@ -41,10 +42,11 @@ const DEFAULT_BACKEND_READINESS_INTERVAL = Duration.millis(100);
const DEFAULT_BACKEND_READINESS_REQUEST_TIMEOUT = Duration.seconds(1);
// Match the service launcher's grace period so backends can stop provider sessions.
const DEFAULT_BACKEND_TERMINATE_GRACE = Duration.seconds(5);
// Allow SIGKILL and the spawner exit wait after the SIGTERM grace so quit cannot
// orphan a backend whose shutdown hangs.
const BACKEND_STOP_ACK_TIMEOUT = Duration.seconds(1);
// Allow the stop ack, SIGKILL and the spawner exit wait around the SIGTERM grace so quit
// cannot orphan a backend whose shutdown hangs.
export const BACKEND_QUIT_STOP_TIMEOUT = Duration.sum(
DEFAULT_BACKEND_TERMINATE_GRACE,
Duration.sum(BACKEND_STOP_ACK_TIMEOUT, DEFAULT_BACKEND_TERMINATE_GRACE),
Duration.seconds(2),
);
const DEFAULT_BACKEND_OUTPUT_DRAIN_TIMEOUT = Duration.seconds(5);
Expand Down Expand Up @@ -444,6 +446,20 @@ export const runBackendProcess = Effect.fn("runBackendProcess")(function* (
additionalFds[`fd${options.bootstrap.desktopBrowserControlFd}`] = { type: "output" };
}
}
const stopNotice = yield* Deferred.make<void>();
const stopFds =
options.bootstrapDelivery === "fd3" &&
options.bootstrap.desktopStopFd !== undefined &&
options.bootstrap.desktopStopAckFd !== undefined
? { stop: options.bootstrap.desktopStopFd, ack: options.bootstrap.desktopStopAckFd }
: undefined;
if (stopFds !== undefined) {
additionalFds[`fd${stopFds.stop}`] = {
type: "input",
stream: Stream.fromEffectDrain(Deferred.await(stopNotice)),
};
additionalFds[`fd${stopFds.ack}`] = { type: "output" };
}
const command = ChildProcess.make(options.executablePath, options.args, {
cwd: options.cwd,
env: options.env,
Expand All @@ -469,6 +485,19 @@ export const runBackendProcess = Effect.fn("runBackendProcess")(function* (
}),
),
);
if (stopFds !== undefined) {
// Added after spawn so it runs before the spawner's kill. Windows sends no signal, so the
// backend must hear of the stop before its providers die, or it records their exits as state.
yield* Effect.addFinalizer(() =>
Deferred.succeed(stopNotice, undefined).pipe(
Effect.andThen(
handle
.getOutputFd(stopFds.ack)
.pipe(Stream.runDrain, Effect.timeoutOption(BACKEND_STOP_ACK_TIMEOUT), Effect.ignore),
),
),
);
}
const outputFibers: Array<Fiber.Fiber<void, never>> = [];

yield* options.onStarted?.(handle.pid) ?? Effect.void;
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/cli/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -385,6 +385,8 @@ export const resolveServerConfig = Effect.fn("config.resolveServerConfig")(funct
const desktopTelemetryControlFd = bootstrap?.desktopTelemetryControlFd;
const desktopBrowserFd = bootstrap?.desktopBrowserFd;
const desktopBrowserControlFd = bootstrap?.desktopBrowserControlFd;
const desktopStopFd = bootstrap?.desktopStopFd;
const desktopStopAckFd = bootstrap?.desktopStopAckFd;
const resourceMonitorPath = bootstrap?.resourceMonitorPath;
const autoBootstrapProjectFromCwd = Option.getOrElse(
resolveOptionPrecedence(
Expand Down Expand Up @@ -518,6 +520,8 @@ export const resolveServerConfig = Effect.fn("config.resolveServerConfig")(funct
desktopTelemetryControlFd,
desktopBrowserFd,
desktopBrowserControlFd,
desktopStopFd,
desktopStopAckFd,
resourceMonitorPath,
autoBootstrapProjectFromCwd,
logWebSocketEvents,
Expand Down
8 changes: 6 additions & 2 deletions apps/server/src/cli/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ import { Command, GlobalFlag } from "effect/cli";
import * as CliError from "effect/cli/CliError";

import * as ServerConfig from "../config.ts";
import { awaitLauncherLoss } from "../launcherWatch.ts";
import { awaitLauncherLoss, awaitLauncherStopOnFds } from "../launcherWatch.ts";
import * as DatabaseMaintenance from "../persistence/DatabaseMaintenance.ts";
import * as ProcessRunner from "../processRunner.ts";
import * as AutomaticPort from "./automaticPort.ts";
Expand Down Expand Up @@ -49,9 +49,13 @@ const runServerCommand = Effect.fn("server.runServerCommand")(function* (
process.stdout.on("error", ignoreClosedLauncherPipe);
process.stderr.on("error", ignoreClosedLauncherPipe);
});
const launcherStop =
config.desktopStopFd === undefined || config.desktopStopAckFd === undefined
? Effect.never
: awaitLauncherStopOnFds({ stopFd: config.desktopStopFd, ackFd: config.desktopStopAckFd });
return yield* Effect.raceFirst(
run,
awaitLauncherLoss().pipe(
Effect.raceFirst(awaitLauncherLoss(), launcherStop).pipe(
// End like SIGTERM does (interrupt-only exit, so the runtime forces process exit).
Effect.andThen(Effect.interrupt),
),
Expand Down
4 changes: 4 additions & 0 deletions apps/server/src/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,8 @@ export class ServerConfig extends Context.Service<
readonly desktopTelemetryControlFd?: number | undefined;
readonly desktopBrowserFd?: number | undefined;
readonly desktopBrowserControlFd?: number | undefined;
readonly desktopStopFd?: number | undefined;
readonly desktopStopAckFd?: number | undefined;
readonly resourceMonitorPath?: string | undefined;
readonly autoBootstrapProjectFromCwd: boolean;
readonly logWebSocketEvents: boolean;
Expand Down Expand Up @@ -256,6 +258,8 @@ const makeTest = Effect.fn("ServerConfig.makeTest")(function* (
desktopTelemetryControlFd: undefined,
desktopBrowserFd: undefined,
desktopBrowserControlFd: undefined,
desktopStopFd: undefined,
desktopStopAckFd: undefined,
resourceMonitorPath: undefined,
staticDir: undefined,
devUrl,
Expand Down
55 changes: 53 additions & 2 deletions apps/server/src/launcherWatch.test.ts
Original file line number Diff line number Diff line change
@@ -1,9 +1,13 @@
import { it } from "@effect/vitest";
import { assert, it } from "@effect/vitest";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Fiber from "effect/Fiber";
import * as Stream from "effect/Stream";
import * as TestClock from "effect/testing/TestClock";
import * as HostProcess from "@t3tools/shared/HostProcess";

import { awaitLauncherLoss } from "./launcherWatch.ts";
import { awaitLauncherLoss, awaitLauncherStop } from "./launcherWatch.ts";
import { isShutdownIntentRecorded } from "./shutdownIntent.ts";

const launcherPid = 4100;
// Orphans are adopted by init or a subreaper, so the new parent is not always 1.
Expand All @@ -21,3 +25,50 @@ it.effect("notices a launcher that was already gone when the watch began", () =>
yield* Fiber.join(fiber);
}),
);

it.effect("keeps ingesting when the stop notice channel breaks", () =>
Effect.gen(function* () {
let acknowledged = false;
const fiber = yield* Effect.forkChild(
awaitLauncherStop({
stopNotice: Stream.fail(new Error("EBADF")),
acknowledge: Effect.sync(() => {
acknowledged = true;
}),
}),
);
yield* TestClock.adjust("1 minute");
assert.isFalse(acknowledged);
assert.isUndefined(fiber.pollUnsafe());
}),
);

it.effect("records shutdown intent before acknowledging the desktop's stop notice", () =>
Effect.gen(function* () {
const notice = yield* Deferred.make<void>();
const intentAtAck = yield* Deferred.make<boolean>();
yield* Effect.forkChild(
awaitLauncherStop({
stopNotice: Stream.fromEffectDrain(Deferred.await(notice)),
acknowledge: Effect.suspend(() =>
Deferred.succeed(intentAtAck, isShutdownIntentRecorded()),
),
}),
);
yield* TestClock.adjust("1 minute");
assert.isFalse(isShutdownIntentRecorded());

yield* Deferred.succeed(notice, undefined);
assert.isTrue(yield* Deferred.await(intentAtAck));
}),
);

it.effect("shuts down when the desktop stops but never kills the backend", () =>
Effect.gen(function* () {
const fiber = yield* Effect.forkChild(
awaitLauncherStop({ stopNotice: Stream.empty, acknowledge: Effect.void }),
);
yield* TestClock.adjust("5 seconds");
yield* Fiber.join(fiber);
}),
);
74 changes: 74 additions & 0 deletions apps/server/src/launcherWatch.ts
Original file line number Diff line number Diff line change
@@ -1,8 +1,25 @@
import * as NodeStream from "@effect/platform-node/NodeStream";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
import * as HostProcess from "@t3tools/shared/HostProcess";
import * as NodeNet from "node:net";

import { recordShutdownIntent } from "./shutdownIntent.ts";

class DesktopStopPipeError extends Schema.TaggedError<DesktopStopPipeError>()(
"DesktopStopPipeError",
{ fd: Schema.Number, cause: Schema.Defect() },
) {
override get message(): string {
return `Failed to open desktop stop pipe file descriptor ${this.fd}.`;
}
}

const DEFAULT_POLL_INTERVAL = Duration.seconds(2);
const LAUNCHER_KILL_GRACE = Duration.seconds(5);

// Capture before config and bootstrap reads to detect launcher loss during startup.
const startupParentPid = process.ppid;
Expand Down Expand Up @@ -33,3 +50,60 @@ export const awaitLauncherLoss = Effect.fn("launcherWatch.awaitLauncherLoss")(fu
});
return yield* poll;
});

/**
* Windows has no stop signal, so the desktop ends the stop notice before it kills the backend
* tree. Intent must be recorded before the ack releases that kill. Completes, so the backend
* exits, only if the launcher stopped without killing it.
*/
export const awaitLauncherStop = Effect.fn("launcherWatch.awaitLauncherStop")(function* (input: {
readonly stopNotice: Stream.Stream<unknown, Error>;
readonly acknowledge: Effect.Effect<void>;
}): Effect.fn.Return<void> {
const notice = yield* Stream.runDrain(input.stopNotice).pipe(Effect.exit);
if (Exit.isFailure(notice)) {
yield* Effect.logWarning("desktop stop notice failed; only signals can stop the backend", {
cause: notice.cause,
});
return yield* Effect.never;
}
recordShutdownIntent();
yield* input.acknowledge;
yield* Effect.sleep(LAUNCHER_KILL_GRACE);
yield* Effect.logWarning("desktop launcher stopped without ending the backend; shutting down");
});

export const awaitLauncherStopOnFds = (fds: { readonly stopFd: number; readonly ackFd: number }) =>
Effect.acquireRelease(
// Socket reads stay cancellable; filesystem pipe reads keep Node alive after scope shutdown.
Effect.try({
try: () => new NodeNet.Socket({ fd: fds.stopFd, readable: true, writable: false }),
catch: (cause) => new DesktopStopPipeError({ fd: fds.stopFd, cause }),
}),
(socket) => Effect.sync(() => socket.destroy()),
).pipe(
Effect.matchCauseEffect({
onFailure: (cause) =>
Effect.logWarning("desktop stop notice is unavailable", { cause }).pipe(
Effect.andThen(Effect.never),
),
onSuccess: (socket) =>
awaitLauncherStop({
stopNotice: NodeStream.fromReadable<Uint8Array, Error>({
evaluate: () => socket,
closeOnDone: true,
onError: (cause) => (cause instanceof Error ? cause : new Error(String(cause))),
}),
acknowledge: Effect.try({
try: () =>
new NodeNet.Socket({ fd: fds.ackFd, readable: false, writable: true }).destroy(),
catch: (cause) => new DesktopStopPipeError({ fd: fds.ackFd, cause }),
}).pipe(
Effect.catchCause((cause) =>
Effect.logWarning("could not acknowledge the desktop stop notice", { cause }),
),
),
}),
}),
Effect.scoped,
);
Loading
Loading