diff --git a/doc/idle-plugin-jobs.md b/doc/idle-plugin-jobs.md index 71f1f163d97..3fb6575c54d 100644 --- a/doc/idle-plugin-jobs.md +++ b/doc/idle-plugin-jobs.md @@ -37,7 +37,7 @@ including a worker becoming available without a database mutation. queued, pending, or running records. Manual and scheduled dispatch share this boundary. Worker RPC calls run outside transactions. - Active calls consume concurrency slots and remain tracked until settlement. - Their completion wakes waiting work without an interval. Slow calls do not + Their completion requests a coalesced refresh for waiting work. Slow calls do not prevent other available slots from being used. - Completion and the scheduled job's next timestamp commit together. Failed persistence retries the retained settlement, not the plugin side effect. @@ -48,7 +48,16 @@ including a worker becoming available without a database mutation. admitted work, then flushes retained completion results without another RPC. If persistence still fails, stop rejects and retains the result for another attempt; it does not report successful shutdown. Existing calls retain their - RPC timeout. Application teardown must complete before the pool is ended. + RPC timeout. A failed stop still allows application and plugin-worker cleanup; + teardown then retries the retained settlement once before reporting any remaining + error, while the database pool is open. + +Producer and lifecycle wakes share a 30-second minimum refresh spacing, matching +the former scan budget. Sequential run-status writes coalesce instead of each +causing a full inventory scan. This is a bound on requested refreshes, not a +recurring timer: settled workers without a future obligation remain disarmed. +Manual dispatch still uses its direct claim path. New schedules and completion- +released capacity can wait up to that spacing before scheduled dispatch. ## Limits @@ -98,3 +107,23 @@ external effect that the database has never acknowledged. The final focused run passes 62 cases across plugin scheduling, loader recovery, warm standby, and the delivery coordinator. Server typechecking passes. Remote checks and a fresh review remain required for the repaired head. + + +A second review reproduced 16 full inventory reads during fifteen sequential run +lifecycles on forty manual jobs. Restoring the prior 30-second spacing bounds that +burst to one read, then reconciliation returns to zero SQL. Shutdown regressions +also reproduced skipped cleanup after a settlement error. The shared teardown +boundary now always attempts remaining cleanup and retries a failed stop once; +both application teardown and plugin-loader shutdown use it. A PostgreSQL trigger +case verifies the retained result is saved after cleanup removes the write fault, +without a second worker RPC. Persistent failures remain visible. + +The first expanded timing run failed fifteen assertions because the test assumed +one second was sufficient to settle a coalesced wake. Tests now wait for the actual +null/future deadline before measuring three minutes of SQL silence. The corrected +run passed all 75 cases; the additional post-cleanup retry case is verified separately +below. No production timeout or quiet assertion was relaxed. + +The post-cleanup retry run passes all 48 shutdown/plugin cases (28 plugin and +20 shutdown). Together with the unchanged consumer selection this covers 76 +unique cases. Server typecheck and build pass on the final round-two repair. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 3a446cfdd4a..c0633d791f7 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -976,3 +976,16 @@ updates take the job lock. All 62 focused lifecycle/consumer cases and server typechecking pass. #17510 at 400f4c0fc8 cleared the security gate and previous CI failures, but environment-cleanup pagination has a 32-versus-33 assertion race in server shard 1. Diagnose and repair it before accepting the stack. + + +Plugin round-two review repair restores a 30-second minimum between requested +full refreshes without adding idle polling. A burst regression failed at 16 +reads before repair and passes at at most one afterward. Failed plugin stop +no longer skips application/worker teardown; retained settlement is retried once +after cleanup while the pool is open, with persistent errors still reported. +The final 48 shutdown/plugin cases and earlier 75-case consumer selection pass +(76 unique cases), with server types/build passing. Fresh remote gates remain. +Approval #17510 has passing latest-head CI and Greptile 5/5 at 0c7109c103, but +Paperclip review identified persistent-failure backoff and duplicated PR matching +that still require repair. Tool health remains the next new migration after +these owning-PR repairs. diff --git a/server/src/__tests__/plugin-scheduled-work.test.ts b/server/src/__tests__/plugin-scheduled-work.test.ts index d61f2160eb5..48fb5c41b84 100644 --- a/server/src/__tests__/plugin-scheduled-work.test.ts +++ b/server/src/__tests__/plugin-scheduled-work.test.ts @@ -1,3 +1,4 @@ +import { finishServiceShutdown } from "../shutdown.js"; import { EventEmitter } from "node:events"; import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -61,7 +62,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu }, { timeout: 5000 }); } async function quiet() { - await settled(); expect(coordinator!.nextWakeAt()).toBeNull(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute"); await vi.advanceTimersByTimeAsync(180_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); @@ -80,6 +85,23 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu } it("does not query repeatedly after an empty startup reconciliation", async () => { await start(); await quiet(); }); + it("bounds full refreshes while many job status writes arrive sequentially", async () => { + await pluginJobStore(db).syncJobDeclarations(pluginId, Array.from({ length: 40 }, (_, i) => ({ jobKey: `manual-${i}`, schedule: "" }))); + const jobs = await db.select().from(pluginJobs).where(eq(pluginJobs.pluginId, pluginId)); + await start(); await quiet(); + const reads = vi.spyOn(db, "select"); + for (const job of jobs.slice(0, 15)) { + const run = await pluginJobStore(db).createRun({ jobId: job.id, pluginId, trigger: "manual" }); + await pluginJobStore(db).markRunning(run.id); + await pluginJobStore(db).completeRun(run.id, { status: "succeeded" }); + await vi.advanceTimersByTimeAsync(1100); + } + expect(reads.mock.calls.length).toBeLessThanOrEqual(1); + reads.mockRestore(); + await quiet(); + expect(await runs()).toHaveLength(15); + expect(rpc).not.toHaveBeenCalled(); + }); it("keeps configured manual-only and invalid schedules quiet", async () => { await job("manual", ""); await job("invalid", "not cron"); await start(); await quiet(); expect(rpc).not.toHaveBeenCalled(); }); @@ -147,7 +169,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`); await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`); await due(); await start(); - await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); expect(rpc).toHaveBeenCalledTimes(1); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); expect(rpc).toHaveBeenCalledTimes(1); await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0; await completed(); expect(rpc).toHaveBeenCalledTimes(1); }); @@ -155,7 +177,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`); await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`); await due(); await start(); - await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); draining = true; await coordinator!.stop(); await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0; @@ -163,11 +185,28 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect((await runs())[0].status).toBe("succeeded"); expect(rpc).toHaveBeenCalledTimes(1); }); + it("finishes service cleanup and retries a failed settlement before closing the pool", async () => { + await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`); + await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`); + await due(); await start(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); + draining = true; await coordinator!.stop(); errors.length = 0; + const cleanup = vi.fn(async () => { + expect((await runs())[0].status).toBe("running"); + await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); + }); + const stop = vi.fn(() => scheduler!.stop()); + await finishServiceShutdown(stop, cleanup); + expect(cleanup).toHaveBeenCalledOnce(); + expect(stop).toHaveBeenCalledTimes(2); + expect((await runs())[0].status).toBe("succeeded"); + expect(rpc).toHaveBeenCalledTimes(1); + }); it("reports shutdown persistence failure and retains the result for another stop attempt", async () => { await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`); await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`); await due(); await start(); - await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); draining = true; await coordinator!.stop(); errors.length = 0; try { await expect(scheduler!.stop()).rejects.toThrow("Plugin job completion persistence failed during shutdown"); @@ -183,7 +222,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure(${sql.raw("'" + blocked.id + "'")})`); try { await start(); - await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); const independent = await due("independent"); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); @@ -244,7 +283,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu it("does no SQL while waiting for a future configured deadline", async () => { const row = await job(); const at = new Date(Date.now() + 600_000); await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), at); - await start(); await settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + await start(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + }, { timeout: 5000 }); const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); await vi.advanceTimersByTimeAsync(180_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); @@ -254,7 +297,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu const row = await job(); const at = new Date(Date.now() + 600_000); await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), at); await start(); await settled(); await scheduler!.stop(); await coordinator!.stop(); - await start(); await settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + await start(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + }, { timeout: 5000 }); await vi.advanceTimersByTimeAsync(at.getTime() - Date.now()); await completed(); expect(rpc).toHaveBeenCalledTimes(1); }); it("rechecks a paused definition after waiting for its lock", async () => { diff --git a/server/src/app.ts b/server/src/app.ts index d3dfddd0a9f..2205bd47203 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1,3 +1,4 @@ +import { finishServiceShutdown } from "./shutdown.js"; import { promptInjectionPreventionRoutes } from "./routes/prompt-injection-prevention.js"; import { voiceSessionRoutes, voiceWebhookRoutes } from "./routes/voice-sessions.js"; import { eq } from "drizzle-orm"; @@ -1455,10 +1456,7 @@ export async function createApp( let appServicesShutdown: Promise | null = null; const shutdownAppServices = (): Promise => { if (appServicesShutdown) return appServicesShutdown; - appServicesShutdown = (async () => { - // The scheduler tick queries the database. Stop it here, inside the - // awaited teardown, so no tick runs after the caller ends the pool. - await scheduler.stop(); + appServicesShutdown = finishServiceShutdown(() => scheduler.stop(), async () => { await agentLifecycle.stop(); await publicMcpEvents?.stop(); await dotMcpEvents?.stop(); @@ -1493,7 +1491,7 @@ export async function createApp( // caller stops the database and the provider. A lease release that // fails stays a durable record for the startup reaper. await setupTokenLoginService?.shutdown(); - })(); + }); return appServicesShutdown; }; app.locals.paperclipShutdown = shutdownAppServices; diff --git a/server/src/services/plugin-job-scheduler.ts b/server/src/services/plugin-job-scheduler.ts index d9a4c3c8e0b..827600f019b 100644 --- a/server/src/services/plugin-job-scheduler.ts +++ b/server/src/services/plugin-job-scheduler.ts @@ -45,7 +45,7 @@ function nextCron(schedule: string, now: Date): Date | null { } export function createPluginJobScheduler(options: PluginJobSchedulerOptions): PluginJobScheduler { - const { db, jobStore, workerManager, tickIntervalMs = 1_000, + const { db, jobStore, workerManager, tickIntervalMs = 30_000, jobTimeoutMs = 5 * 60_000, maxConcurrentJobs = 10 } = options; const log = logger.child({ service: "plugin-job-scheduler" }); const activeJobs = new Set(); diff --git a/server/src/services/plugin-loader.ts b/server/src/services/plugin-loader.ts index 283cbc6c401..d00f683a84b 100644 --- a/server/src/services/plugin-loader.ts +++ b/server/src/services/plugin-loader.ts @@ -1,3 +1,4 @@ +import { finishServiceShutdown } from "../shutdown.js"; /** * PluginLoader — discovery, installation, and runtime activation of plugins. * @@ -2143,11 +2144,7 @@ export function pluginLoader( const { workerManager, jobScheduler } = runtimeServices; - // 1. Stop the job scheduler tick loop - await jobScheduler.stop(); - - // 2. Stop all worker processes - await workerManager.stopAll(); + await finishServiceShutdown(() => jobScheduler.stop(), () => workerManager.stopAll()); log.info("plugin-loader: all plugins shut down"); }, diff --git a/server/src/shutdown.test.ts b/server/src/shutdown.test.ts index c9d118a8095..754a99fb1e1 100644 --- a/server/src/shutdown.test.ts +++ b/server/src/shutdown.test.ts @@ -5,6 +5,7 @@ import { coordinateHeartbeatSchedulerShutdown, drainRunExecutionFinalizersForShutdown, finalizeServerShutdown, + finishServiceShutdown, loadWithoutCoordinatedShutdownSignalHooks, } from "./shutdown.js"; @@ -523,3 +524,26 @@ describe("coordinateHeartbeatSchedulerShutdown", () => { }); }); }); + +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(); + }); +}); diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index cda7499392b..ea927cbd6e1 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -272,3 +272,18 @@ export async function coordinateHeartbeatSchedulerShutdown< waitedForSchedulerIdle: true, }; } + +export async function finishServiceShutdown( + stop: () => Promise, + cleanup: () => Promise, +): Promise { + const failures: unknown[] = []; + let retryStop = false; + try { await stop(); } catch { retryStop = true; } + try { await cleanup(); } catch (error) { failures.push(error); } + if (retryStop) { + try { await stop(); } catch (error) { failures.unshift(error); } + } + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) throw new AggregateError(failures, "Service shutdown failed"); +}