fix(server): bound plugin refreshes and finish teardown after errors

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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 () => {

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<void> | null = null;
   const shutdownAppServices = (): Promise<void> => {
     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;

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<string>();

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");
     },

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();
+  });
+});

server/src/shutdown.ts

@@ -272,3 +272,18 @@ export async function coordinateHeartbeatSchedulerShutdown<
     waitedForSchedulerIdle: true,
   };
 }
+
+export async function finishServiceShutdown(
+  stop: () => Promise<void>,
+  cleanup: () => Promise<void>,
+): Promise<void> {
+  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");
+}