Files
PaperClipAI/server/src/shutdown.test.ts
T
DottaandClaude Opus 5.5 c4ba3974a7 fix(heartbeat): close run admission before a graceful shutdown closes its listener (t09) (#20102)
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A graceful restart (K2) drains the control plane: it closes HTTP
admission, suspends running runs, and the next boot resumes them
> - Run admission stayed open while the HTTP listener closed.
Heartbeat's shutdown flag was set only later, inside hot-restart
preparation
> - A run that finished in that window released its issue and promoted
the queued comment wake. That started a new run that had no callback
listener
> - A remote warm continuation then failed with
`local_runtime_api_listener_unavailable`, and the task needed a Retry
(restart chaos t09)
> - This pull request closes run admission before the listener closes,
so the promoted wake stays queued for the next boot
> - The benefit is that a comment sent during a deploy continues on its
own after the restart

## Linked Issues or Issue Description

Refs #19596

**What happened?**
In restart-chaos wave w-20261108T192459-s9501 (Daytona warm, Codex C1,
K2), a K2 SIGTERM started a graceful drain at 19:42:52 UTC, and the HTTP
listener closed first. At 19:43:03 the running turn finished and
released its issue. That promoted the queued comment wake, and
`startNextQueuedRunForAgent` started a new run about 11 s into the
drain. Backend setup resolved the local runtime API URL, found the
listener closed, and threw `local_runtime_api_listener_unavailable`. The
remote target is excluded from the closed-listener handoff, so the run
failed, and the task waited for a manual Retry.

**Expected behavior**
No queued run starts once a graceful shutdown has begun. A wake promoted
during the drain stays queued, and the next boot starts it with no
Retry.

**Steps to reproduce**
1. Run a remote warm task and post a comment while its turn runs, so the
comment wake queues behind it.
2. Send SIGTERM to the control plane just before the turn finishes.
3. Before this change, a new run starts during the drain and fails with
`local_runtime_api_listener_unavailable`.

**Paperclip version or commit**
master at d007afbf86 (the wave ran on c5a172a183).

## What Changed

- `heartbeat/run-admission.ts` (new): a process-wide run admission gate
(`closeProcessRunAdmission`, `isProcessRunAdmissionClosed`). HTTP routes
build their own heartbeat instances, so the gate cannot live in one
instance.
- `heartbeat/queue.ts`: a new optional dependency,
`isRunAdmissionClosed`, which defaults to the process-wide gate.
`startNextQueuedRunForAgent`, `resumeQueuedRuns` and
`resumeQueuedRunsAfterDrain` start nothing (and read nothing) once it
returns true. Wakes are still recorded and promoted; only the start is
held.
- `heartbeat.ts`: also honors its own `shutdownInProgress` flag.
- `shutdown.ts`: `closeHttpListenerForShutdown` takes
`closeRunAdmission` and calls it first, before it checks or closes the
listener. `index.ts` passes `closeProcessRunAdmission`.

## Verification

- Red/green, new tests:
- `queue.test.ts`: "starts no run once shutdown closed run admission"
(for `startNextQueuedRunForAgent` and `resumeQueuedRuns`), and "honors
the process-wide run admission gate in every heartbeat instance".
- `shutdown.test.ts`: "closes run admission before it stops accepting
requests".
- `pnpm exec vitest run src/services/heartbeat/queue.test.ts
src/services/heartbeat/recovery.test.ts
src/__tests__/heartbeat-comment-wake-batching.test.ts
src/services/native-runtime/native-startup-retry.integration.test.ts
src/shutdown.test.ts` (server): 5 files, 277 tests passed.
- `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm typecheck` (server): passed.

## Risks

- Low risk. Only queued-run starts are held during a shutdown that has
already begun; wakes, drains, suspension and finalizers are unchanged.
The next boot's queued-run dispatch starts held runs.
- Known remaining window: a run admitted just before shutdown can still
reach backend setup after the listener closed. A local runner hands off
to restart recovery there. A remote warm continuation still fails,
because the remote target is excluded from that handoff (#19596) and the
bootstrap retry does not cover warm continuations. This change removes
the t09 sequence, where the run started after shutdown had begun; it
does not remove that narrower window.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use
(shell, file edits, test execution, a read-only research subagent).

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-08 14:59:55 -06:00

704 lines
25 KiB
TypeScript

import { EventEmitter } from "node:events";
import { describe, expect, it, vi } from "vitest";
import {
closeHttpListenerForShutdown,
coordinateHeartbeatSchedulerShutdown,
createPreDrainShutdownBudget,
drainRunExecutionFinalizersForShutdown,
finalizeServerShutdown,
finishServiceShutdown,
loadWithoutCoordinatedShutdownSignalHooks,
runBoundedShutdownStep,
} from "./shutdown.js";
function deferred<T = void>() {
let resolve!: (value: T) => void;
let reject!: (reason?: unknown) => void;
const promise = new Promise<T>((res, rej) => {
resolve = res;
reject = rej;
});
return { promise, resolve, reject };
}
function stubLogger() {
return { info: vi.fn(), error: vi.fn() };
}
describe("finalizeServerShutdown", () => {
it("awaits the setup-token cleanup before the database stop and the process exit", async () => {
const order: string[] = [];
// The held promise models the setup-token session cancellation and its
// sandbox lease release. The teardown must not continue while it is pending.
const release = deferred();
const shutdownAppServices = vi.fn(async () => {
order.push("appServices:start");
await release.promise;
order.push("appServices:settled");
});
const closeHttpListener = vi.fn(async () => {
order.push("listener:close");
});
const closeDatabase = vi.fn(async () => {
order.push("database:close");
});
const stopEmbeddedPostgres = vi.fn(async () => {
order.push("postgres:stop");
});
const shutdownInstrumentation = vi.fn(async () => {
order.push("instrumentation:flush");
});
const shutdownSentry = vi.fn(async () => {
order.push("sentry:flush");
});
let exited = false;
const finalize = finalizeServerShutdown({
signal: "SIGTERM",
shutdownAppServices,
closeHttpListener,
closeDatabase,
stopEmbeddedPostgres,
shutdownInstrumentation,
shutdownSentry,
log: stubLogger(),
}).then(() => {
// This models the caller's `process.exit(0)` continuation.
exited = true;
order.push("exit");
});
// The cleanup is in flight. The database stop, the instrumentation flush,
// and the process exit continuation must all wait for it to settle.
await vi.waitFor(() => expect(shutdownAppServices).toHaveBeenCalledOnce());
// The listener already closed: requests are drained while every service
// is still available, and nothing after this point can be reached.
expect(closeHttpListener).toHaveBeenCalledOnce();
expect(closeDatabase).not.toHaveBeenCalled();
expect(stopEmbeddedPostgres).not.toHaveBeenCalled();
expect(shutdownInstrumentation).not.toHaveBeenCalled();
expect(exited).toBe(false);
release.resolve();
await finalize;
expect(exited).toBe(true);
expect(order).toEqual([
"listener:close",
"appServices:start",
"appServices:settled",
"database:close",
"postgres:stop",
"instrumentation:flush",
"sentry:flush",
"exit",
]);
});
it("keeps the teardown durable and still exits when the setup-token release fails", async () => {
const order: string[] = [];
// The held promise rejects, which models a lease release that failed. The
// reaper owns the durable retry, so the teardown must log the failure and
// continue rather than swallow it or block the exit.
const release = deferred();
const releaseError = new Error("lease release failed");
const shutdownAppServices = vi.fn(async () => {
await release.promise;
});
const stopEmbeddedPostgres = vi.fn(async () => {
order.push("postgres:stop");
});
const shutdownInstrumentation = vi.fn(async () => {
order.push("instrumentation:flush");
});
const shutdownSentry = vi.fn(async () => undefined);
const log = stubLogger();
let exited = false;
const finalize = finalizeServerShutdown({
signal: "SIGTERM",
shutdownAppServices,
stopEmbeddedPostgres,
shutdownInstrumentation,
shutdownSentry,
log,
}).then(() => {
exited = true;
});
await vi.waitFor(() => expect(shutdownAppServices).toHaveBeenCalledOnce());
expect(stopEmbeddedPostgres).not.toHaveBeenCalled();
expect(exited).toBe(false);
release.reject(releaseError);
await finalize;
// The teardown surfaced the failure in the log, then finished the ordered
// teardown and reached the exit continuation.
expect(log.error).toHaveBeenCalledWith(
expect.objectContaining({ err: releaseError, signal: "SIGTERM" }),
expect.any(String),
);
expect(order).toEqual(["postgres:stop", "instrumentation:flush"]);
expect(exited).toBe(true);
});
it("logs a failed database close and still stops the provider and exits", async () => {
const order: string[] = [];
const closeError = new Error("pool end timed out");
const closeDatabase = vi.fn(async () => {
order.push("database:close");
throw closeError;
});
const stopEmbeddedPostgres = vi.fn(async () => {
order.push("postgres:stop");
});
const log = stubLogger();
await finalizeServerShutdown({
signal: "SIGTERM",
shutdownAppServices: vi.fn(async () => undefined),
closeDatabase,
stopEmbeddedPostgres,
shutdownInstrumentation: vi.fn(async () => undefined),
shutdownSentry: vi.fn(async () => undefined),
log,
});
expect(order).toEqual(["database:close", "postgres:stop"]);
expect(log.error).toHaveBeenCalledWith(
expect.objectContaining({ err: closeError, signal: "SIGTERM" }),
"Database client shutdown failed",
);
});
it("waits for pending run-failure reports before the database stop and the Sentry flush", async () => {
const order: string[] = [];
const release = deferred();
const drainPendingRunFailureReports = vi.fn(async () => {
order.push("runFailureReports:start");
await release.promise;
order.push("runFailureReports:settled");
});
const closeDatabase = vi.fn(async () => {
order.push("database:close");
});
const shutdownSentry = vi.fn(async () => {
order.push("sentry:flush");
});
const finalize = finalizeServerShutdown({
signal: "SIGTERM",
shutdownAppServices: vi.fn(async () => undefined),
drainPendingRunFailureReports,
closeDatabase,
stopEmbeddedPostgres: null,
shutdownInstrumentation: vi.fn(async () => undefined),
shutdownSentry,
log: stubLogger(),
});
await vi.waitFor(() => expect(drainPendingRunFailureReports).toHaveBeenCalledOnce());
expect(closeDatabase).not.toHaveBeenCalled();
expect(shutdownSentry).not.toHaveBeenCalled();
release.resolve();
await finalize;
expect(order).toEqual([
"runFailureReports:start",
"runFailureReports:settled",
"database:close",
"sentry:flush",
]);
});
it("logs a failed run-failure report drain and still stops the database and exits", async () => {
const drainError = new Error("drain failed");
const drainPendingRunFailureReports = vi.fn(async () => {
throw drainError;
});
const closeDatabase = vi.fn(async () => undefined);
const log = stubLogger();
await finalizeServerShutdown({
signal: "SIGTERM",
shutdownAppServices: vi.fn(async () => undefined),
drainPendingRunFailureReports,
closeDatabase,
stopEmbeddedPostgres: null,
shutdownInstrumentation: vi.fn(async () => undefined),
shutdownSentry: vi.fn(async () => undefined),
log,
});
expect(closeDatabase).toHaveBeenCalledOnce();
expect(log.error).toHaveBeenCalledWith(
expect.objectContaining({ err: drainError, signal: "SIGTERM" }),
"run-failure report drain failed",
);
});
it("skips the database stop when no embedded PostgreSQL runs in this process", async () => {
const shutdownAppServices = vi.fn(async () => undefined);
const shutdownInstrumentation = vi.fn(async () => undefined);
const shutdownSentry = vi.fn(async () => undefined);
const log = stubLogger();
await finalizeServerShutdown({
signal: "SIGINT",
shutdownAppServices,
stopEmbeddedPostgres: null,
shutdownInstrumentation,
shutdownSentry,
log,
});
expect(shutdownAppServices).toHaveBeenCalledOnce();
expect(shutdownInstrumentation).toHaveBeenCalledOnce();
expect(log.info).not.toHaveBeenCalled();
});
});
describe("closeHttpListenerForShutdown", () => {
function fakeServer(input: { listening: boolean; closeDelayMs?: number | null }) {
const closeIdleConnections = vi.fn();
const closeAllConnections = vi.fn();
const close = vi.fn((callback?: (err?: Error) => void) => {
if (input.closeDelayMs === null) return;
setTimeout(() => callback?.(), input.closeDelayMs ?? 0);
});
return { listening: input.listening, close, closeIdleConnections, closeAllConnections };
}
it("stops accepting, closes idle keep-alive sockets, and resolves once the listener closed", async () => {
const server = fakeServer({ listening: true, closeDelayMs: 0 });
await expect(
closeHttpListenerForShutdown({ server, signal: "SIGTERM", timeoutMs: 1_000, log: stubLogger() }),
).resolves.toBe("closed");
expect(server.close).toHaveBeenCalledOnce();
expect(server.closeIdleConnections).toHaveBeenCalledOnce();
expect(server.closeAllConnections).not.toHaveBeenCalled();
});
it("closes run admission before it stops accepting requests", async () => {
const order: string[] = [];
const server = fakeServer({ listening: true, closeDelayMs: 0 });
server.close.mockImplementation((callback?: (err?: Error) => void) => { order.push("listener"); setTimeout(() => callback?.(), 0); });
await closeHttpListenerForShutdown({ server, signal: "SIGTERM", timeoutMs: 1_000, log: stubLogger(),
closeRunAdmission: () => order.push("run_admission") });
expect(order).toEqual(["run_admission", "listener"]);
const stopped = vi.fn();
await expect(closeHttpListenerForShutdown({ server: fakeServer({ listening: false }), signal: "SIGTERM", log: stubLogger(),
closeRunAdmission: stopped })).resolves.toBe("not_listening");
expect(stopped).toHaveBeenCalledOnce();
});
it("closes the remaining connections when the drain outlives the grace period", async () => {
vi.useFakeTimers();
try {
const server = fakeServer({ listening: true, closeDelayMs: null });
const log = stubLogger();
const pending = closeHttpListenerForShutdown({ server, signal: "SIGINT", timeoutMs: 250, log });
await vi.advanceTimersByTimeAsync(250);
await expect(pending).resolves.toBe("timed_out");
expect(server.closeAllConnections).toHaveBeenCalledOnce();
expect(log.info).toHaveBeenCalledWith(
expect.objectContaining({ timeoutMs: 250 }),
expect.stringContaining("timed out"),
);
} finally {
vi.useRealTimers();
}
});
it("waits for HTTP work without waiting for upgraded sockets to close", async () => {
const server = fakeServer({ listening: true, closeDelayMs: null });
const requests = deferred();
let finished = false;
const drain = closeHttpListenerForShutdown({ server, signal: "SIGTERM", log: stubLogger(),
drainRequests: () => requests.promise }).then(result => { finished = true; return result; });
await Promise.resolve();
expect(finished).toBe(false);
requests.resolve();
await expect(drain).resolves.toBe("closed");
expect(server.close).toHaveBeenCalledOnce();
expect(server.closeAllConnections).not.toHaveBeenCalled();
});
it("does nothing when the listener was never bound", async () => {
const server = fakeServer({ listening: false });
await expect(
closeHttpListenerForShutdown({ server, signal: "SIGTERM", log: stubLogger() }),
).resolves.toBe("not_listening");
expect(server.close).not.toHaveBeenCalled();
});
});
describe("drainRunExecutionFinalizersForShutdown", () => {
it("awaits bounded execution finalizers", async () => {
const release = deferred();
const drain = vi.fn(() => release.promise);
const pending = drainRunExecutionFinalizersForShutdown({
signal: "SIGTERM",
drain,
timeoutMs: 1_000,
log: stubLogger(),
});
await vi.waitFor(() => expect(drain).toHaveBeenCalledOnce());
release.resolve();
await expect(pending).resolves.toBe("drained");
});
it("returns after the bounded timeout when an adopted run remains active", async () => {
vi.useFakeTimers();
try {
const log = stubLogger();
const pending = drainRunExecutionFinalizersForShutdown({
signal: "SIGINT",
drain: () => new Promise<void>(() => undefined),
timeoutMs: 250,
log,
});
await vi.advanceTimersByTimeAsync(250);
await expect(pending).resolves.toBe("timed_out");
expect(log.info).toHaveBeenCalledWith(
expect.objectContaining({ timeoutMs: 250 }),
expect.stringContaining("timed out"),
);
} finally {
vi.useRealTimers();
}
});
});
describe("runBoundedShutdownStep", () => {
it("returns the step value and logs its duration", async () => {
const log = stubLogger();
await expect(runBoundedShutdownStep({
step: "delivery_work_stop",
signal: "SIGTERM",
timeoutMs: 1_000,
run: async () => "stopped",
log,
})).resolves.toMatchObject({ outcome: "completed", value: "stopped" });
expect(log.info).toHaveBeenCalledWith(
expect.objectContaining({ step: "delivery_work_stop", signal: "SIGTERM", durationMs: expect.any(Number) }),
"shutdown step completed",
);
});
it("stops waiting for a step that outlives its bound", async () => {
vi.useFakeTimers();
try {
const log = stubLogger();
const pending = runBoundedShutdownStep({
step: "scheduler_idle",
signal: "SIGTERM",
timeoutMs: 250,
run: () => new Promise<void>(() => undefined),
log,
});
await vi.advanceTimersByTimeAsync(250);
await expect(pending).resolves.toMatchObject({ outcome: "timed_out" });
expect(log.info).toHaveBeenCalledWith(
expect.objectContaining({ step: "scheduler_idle", timeoutMs: 250 }),
expect.stringContaining("timed out"),
);
} finally {
vi.useRealTimers();
}
});
it("logs a step that rejects and does not reject the shutdown", async () => {
const log = stubLogger();
const error = new Error("skill sync stop failed");
await expect(runBoundedShutdownStep({
step: "skill_source_sync_stop",
signal: "SIGINT",
timeoutMs: 1_000,
run: async () => { throw error; },
log,
})).resolves.toMatchObject({ outcome: "failed", error });
expect(log.error).toHaveBeenCalledWith(
expect.objectContaining({ step: "skill_source_sync_stop", err: error }),
expect.stringContaining("failed"),
);
});
it("contains a step that throws synchronously", async () => {
const result = await runBoundedShutdownStep({
step: "systemd_notify",
signal: "SIGTERM",
timeoutMs: 1_000,
run: () => { throw new Error("spawn failed"); },
log: stubLogger(),
});
expect(result.outcome).toBe("failed");
});
});
describe("createPreDrainShutdownBudget", () => {
it("gives each step only the time that is left in the shared budget", async () => {
vi.useFakeTimers();
try {
const log = stubLogger();
const budget = createPreDrainShutdownBudget({ signal: "SIGTERM", budgetMs: 1_000, log });
const first = budget.run("skill_source_sync_stop", () =>
new Promise<void>((resolve) => setTimeout(resolve, 700)));
await vi.advanceTimersByTimeAsync(700);
await expect(first).resolves.toMatchObject({ outcome: "completed" });
const second = budget.run("scheduler_idle", () => new Promise<void>(() => undefined));
await vi.advanceTimersByTimeAsync(300);
await expect(second).resolves.toMatchObject({ outcome: "timed_out" });
expect(log.info).toHaveBeenCalledWith(
expect.objectContaining({ step: "scheduler_idle", timeoutMs: 300 }),
expect.stringContaining("timed out"),
);
} finally {
vi.useRealTimers();
}
});
it("reaches the run drain within the budget when every pre-drain step hangs", async () => {
vi.useFakeTimers();
try {
const budget = createPreDrainShutdownBudget({ signal: "SIGTERM", budgetMs: 1_000, log: stubLogger() });
const hang = () => new Promise<void>(() => undefined);
const drain = vi.fn(async () => ({ interrupted: 1 }));
const shutdown = (async () => {
await budget.run("skill_source_sync_stop", hang);
await budget.run("delivery_work_stop", hang);
await coordinateHeartbeatSchedulerShutdown({
signal: "SIGTERM",
prepareHotRestartShutdown: async () => {
const preparation = await budget.run("hot_restart_prepare", () =>
new Promise<{ skipDrain: boolean }>(() => undefined));
if (preparation.outcome === "completed") return preparation.value;
throw new Error("hot-restart shutdown preparation timed out");
},
waitForHeartbeatSchedulerIdle: async () => {
await budget.run("scheduler_idle", hang);
},
});
return drain();
})();
const startedAt = Date.now();
await vi.advanceTimersByTimeAsync(1_000);
await vi.runAllTimersAsync();
await expect(shutdown).resolves.toEqual({ interrupted: 1 });
expect(drain).toHaveBeenCalledOnce();
expect(Date.now() - startedAt).toBeLessThan(1_010);
} finally {
vi.useRealTimers();
}
});
});
describe("loadWithoutCoordinatedShutdownSignalHooks", () => {
it("removes the eager signal handlers from the real embedded-postgres import", async () => {
const before = {
SIGINT: process.rawListeners("SIGINT"),
SIGTERM: process.rawListeners("SIGTERM"),
};
const moduleName = "embedded-postgres";
await loadWithoutCoordinatedShutdownSignalHooks(() => import(moduleName));
expect(process.rawListeners("SIGINT")).toEqual(before.SIGINT);
expect(process.rawListeners("SIGTERM")).toEqual(before.SIGTERM);
});
it("keeps the database available for a marker-backed SIGTERM snapshot", async () => {
const signalTarget = new EventEmitter();
const preexistingSignalListener = vi.fn();
signalTarget.on("SIGTERM", preexistingSignalListener);
let databaseAvailable = true;
const embeddedPostgresExitHook = vi.fn(() => {
databaseAvailable = false;
});
await loadWithoutCoordinatedShutdownSignalHooks(
async () => {
signalTarget.on("SIGINT", embeddedPostgresExitHook);
signalTarget.on("SIGTERM", embeddedPostgresExitHook);
return { default: class EmbeddedPostgres {} };
},
signalTarget,
);
let shutdown: Promise<unknown> | null = null;
let snapshotCaptured = false;
signalTarget.once("SIGTERM", () => {
shutdown = coordinateHeartbeatSchedulerShutdown({
signal: "SIGTERM",
prepareHotRestartShutdown: async () => {
// This models the real failure path: a valid intent exists, and the
// snapshot must query embedded PostgreSQL after SIGTERM is delivered.
expect(databaseAvailable).toBe(true);
snapshotCaptured = true;
return { mode: "hot_restart" as const, skipDrain: true };
},
waitForHeartbeatSchedulerIdle: vi.fn(async () => undefined),
});
});
signalTarget.emit("SIGTERM");
await shutdown;
expect(preexistingSignalListener).toHaveBeenCalledOnce();
expect(embeddedPostgresExitHook).not.toHaveBeenCalled();
expect(snapshotCaptured).toBe(true);
});
});
describe("coordinateHeartbeatSchedulerShutdown", () => {
it("quiesces active scheduler work before capturing a hot-restart snapshot", async () => {
let snapshotCaptured = false;
let releaseScheduler!: () => void;
const schedulerIdle = new Promise<void>((resolve) => {
releaseScheduler = resolve;
});
const waitForHeartbeatSchedulerIdle = vi.fn(() => schedulerIdle);
const shutdown = coordinateHeartbeatSchedulerShutdown({
signal: "SIGTERM",
prepareHotRestartShutdown: vi.fn(async () => {
snapshotCaptured = true;
return { mode: "prepared" as const, skipDrain: true };
}),
waitForHeartbeatSchedulerIdle,
});
await vi.waitFor(() => expect(waitForHeartbeatSchedulerIdle).toHaveBeenCalledOnce());
expect(snapshotCaptured).toBe(false);
releaseScheduler();
const result = await shutdown;
expect(snapshotCaptured).toBe(true);
expect(result).toEqual({
hotRestart: { mode: "prepared", skipDrain: true },
preparationError: null,
waitedForSchedulerIdle: true,
});
});
it("quiesces scheduler work before selecting server-stdio runs to drain", async () => {
const waitForHeartbeatSchedulerIdle = vi.fn(async () => undefined);
const result = await coordinateHeartbeatSchedulerShutdown({
signal: "SIGTERM",
prepareHotRestartShutdown: vi.fn(async () => ({
mode: "acp_drain_required" as const,
skipDrain: false,
drainRunIds: ["acp-run"],
})),
waitForHeartbeatSchedulerIdle,
});
expect(waitForHeartbeatSchedulerIdle).toHaveBeenCalledOnce();
expect(result).toEqual({
hotRestart: {
mode: "acp_drain_required",
skipDrain: false,
drainRunIds: ["acp-run"],
},
preparationError: null,
waitedForSchedulerIdle: true,
});
});
it("preserves the scheduler idle wait for normal graceful shutdown", async () => {
let releaseScheduler!: () => void;
const schedulerIdle = new Promise<void>((resolve) => {
releaseScheduler = resolve;
});
const waitForHeartbeatSchedulerIdle = vi.fn(() => schedulerIdle);
let settled = false;
const shutdown = coordinateHeartbeatSchedulerShutdown({
signal: "SIGTERM",
prepareHotRestartShutdown: vi.fn(async () => ({
mode: "not_requested" as const,
skipDrain: false,
})),
waitForHeartbeatSchedulerIdle,
}).finally(() => {
settled = true;
});
await vi.waitFor(() => expect(waitForHeartbeatSchedulerIdle).toHaveBeenCalledOnce());
expect(settled).toBe(false);
releaseScheduler();
await expect(shutdown).resolves.toEqual({
hotRestart: { mode: "not_requested", skipDrain: false },
preparationError: null,
waitedForSchedulerIdle: true,
});
});
it("waits for scheduler idle when hot-restart preparation is unavailable", async () => {
const waitForHeartbeatSchedulerIdle = vi.fn(async () => undefined);
const result = await coordinateHeartbeatSchedulerShutdown({
signal: "SIGTERM",
prepareHotRestartShutdown: null,
waitForHeartbeatSchedulerIdle,
});
expect(waitForHeartbeatSchedulerIdle).toHaveBeenCalledOnce();
expect(result).toEqual({
hotRestart: null,
preparationError: null,
waitedForSchedulerIdle: true,
});
});
it("falls back to the scheduler idle wait when hot-restart preparation fails", async () => {
const preparationError = new Error("snapshot failed");
const waitForHeartbeatSchedulerIdle = vi.fn(async () => undefined);
const result = await coordinateHeartbeatSchedulerShutdown({
signal: "SIGTERM",
prepareHotRestartShutdown: vi.fn(async () => {
throw preparationError;
}),
waitForHeartbeatSchedulerIdle,
});
expect(waitForHeartbeatSchedulerIdle).toHaveBeenCalledOnce();
expect(result).toEqual({
hotRestart: null,
preparationError,
waitedForSchedulerIdle: true,
});
});
});
describe("service shutdown failure isolation", () => {
it("awaits remaining cleanup before reporting a scheduler failure", async () => {
const failure = new Error("plugin result could not be saved");
const gate = deferred();
const cleanup = vi.fn(() => gate.promise);
let finished = false;
const stopping = finishServiceShutdown(async () => { throw failure; }, cleanup)
.then(() => { finished = true; return undefined; }, error => { finished = true; return error; });
try {
await vi.waitFor(() => expect(cleanup).toHaveBeenCalledOnce());
expect(finished).toBe(false);
} finally { gate.resolve(); }
expect(await stopping).toBe(failure);
});
it("reports both failures after attempting cleanup", async () => {
const stopFailure = new Error("scheduler failure"), cleanupFailure = new Error("worker failure");
const cleanup = vi.fn(async () => { throw cleanupFailure; });
await expect(finishServiceShutdown(async () => { throw stopFailure; }, cleanup))
.rejects.toMatchObject({ errors: [stopFailure, cleanupFailure] });
expect(cleanup).toHaveBeenCalledOnce();
});
});