fix(server): preserve plugin settlements and concurrent deadlines

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

server/src/__tests__/plugin-scheduled-work.test.ts

@@ -151,6 +151,71 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0;
     await completed(); expect(rpc).toHaveBeenCalledTimes(1);
   });
+  it("flushes a retained completion before graceful shutdown finishes", 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));
+    draining = true;
+    await coordinator!.stop();
+    await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0;
+    await scheduler!.stop();
+    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));
+    draining = true; await coordinator!.stop(); errors.length = 0;
+    try {
+      await expect(scheduler!.stop()).rejects.toThrow("Plugin job completion persistence failed during shutdown");
+      expect((await runs())[0].status).toBe("running");
+    } finally { await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); }
+    await scheduler!.stop();
+    expect((await runs())[0].status).toBe("succeeded");
+    expect(rpc).toHaveBeenCalledTimes(1);
+  });
+  it("dispatches independent jobs while one completion remains unwritable", async () => {
+    const blocked = await due("blocked");
+    await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') and new.job_id::text = TG_ARGV[0] 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(${sql.raw("'" + blocked.id + "'")})`);
+    try {
+      await start();
+      await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0));
+      const independent = await due("independent");
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(await runs()).toEqual(expect.arrayContaining([expect.objectContaining({ jobId: independent.id, status: "succeeded" })]));
+      }, { timeout: 5000 });
+      expect(rpc).toHaveBeenCalledTimes(2);
+    } finally {
+      await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0;
+      await completed(2); await settled(); errors.length = 0;
+    }
+  });
+  it("preserves a deadline advanced during a concurrent unchanged declaration update", async () => {
+    const row = await due();
+    const advanced = new Date(Date.now() + 600_000);
+    let release!: () => void, locked!: () => void;
+    const barrier = new Promise<void>(resolve => { release = resolve; });
+    const acquired = new Promise<void>(resolve => { locked = resolve; });
+    const advancing = db.transaction(async tx => {
+      await pluginJobStore(tx).updateRunTimestamps(row.id, new Date(), advanced);
+      locked(); await barrier;
+    });
+    await acquired;
+    const updating = pluginRegistryService(db).upsertJob(pluginId, row.jobKey, { schedule: row.schedule });
+    try {
+      await vi.waitFor(async () => {
+        const [waiting] = await db.execute<{ count: number }>(sql`select count(*)::int as count from pg_stat_activity
+          where datname = current_database() and state = 'active' and wait_event_type = 'Lock'`);
+        expect(waiting.count).toBeGreaterThan(0);
+      });
+    } finally { release(); await advancing; await updating; }
+    expect((await pluginJobStore(db).getJobById(row.id))!.nextRunAt).toEqual(advanced);
+  });
   it("records provider failure and advances the cron obligation", async () => {
     rpc.mockRejectedValueOnce(new Error("plugin rejected job")); await due(); await start(); await completed();
     expect((await runs())[0]).toMatchObject({ status: "failed", error: "plugin rejected job" });

server/src/services/plugin-job-scheduler.ts

@@ -135,10 +135,10 @@ export function createPluginJobScheduler(options: PluginJobSchedulerOptions): Pl
   }
 
   async function refresh(signal?: AbortSignal) {
-    for (const jobId of pendingSettlements.keys()) await settle(jobId);
+    const settlements = await Promise.allSettled([...pendingSettlements.keys()].map(settle));
     const jobs = await db.select().from(pluginJobs).where(eq(pluginJobs.status, "active"));
     const runnable = jobs.filter(job => !unregistered.has(job.pluginId) && workerManager.isRunning(job.pluginId)
-      && !activeJobs.has(job.id) && job.schedule && !validateCron(job.schedule));
+      && !activeJobs.has(job.id) && !pendingSettlements.has(job.id) && job.schedule && !validateCron(job.schedule));
     const outstanding = runnable.length === 0 ? [] : await db.select({ jobId: pluginJobRuns.jobId })
       .from(pluginJobRuns).where(and(inArray(pluginJobRuns.jobId, runnable.map(job => job.id)),
         inArray(pluginJobRuns.status, ["pending", "queued", "running"])));
@@ -168,7 +168,7 @@ export function createPluginJobScheduler(options: PluginJobSchedulerOptions): Pl
       }
     }
     const results = await Promise.allSettled(dispatches);
-    const failed = results.find(result => result.status === "rejected");
+    const failed = [...settlements, ...results].find(result => result.status === "rejected");
     if (failed?.status === "rejected") throw failed.reason;
   }
 
@@ -195,6 +195,9 @@ export function createPluginJobScheduler(options: PluginJobSchedulerOptions): Pl
       subscriptions.clear();
       await Promise.allSettled([ticking]);
       while (inFlight.size) await Promise.allSettled([...inFlight]);
+      const results = await track(Promise.allSettled([...pendingSettlements.keys()].map(settle)));
+      const failures = results.filter(result => result.status === "rejected");
+      if (failures.length) throw new AggregateError(failures.map(result => result.reason), "Plugin job completion persistence failed during shutdown");
     },
     async registerPlugin(pluginId) {
       unregistered.delete(pluginId);

server/src/services/plugin-registry.ts

@@ -678,6 +678,7 @@ export function pluginRegistryService(db: Db) {
           .select()
           .from(pluginJobs)
           .where(and(eq(pluginJobs.pluginId, pluginId), eq(pluginJobs.jobKey, jobKey)))
+          .for("update")
           .then((rows) => rows[0] ?? null);
 
         if (existing) {