mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-12 14:19:00 +01:00
## 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 atd007afbf86(the wave ran onc5a172a183). ## 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>
704 lines
25 KiB
TypeScript
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();
|
|
});
|
|
});
|