From 88467fd43534be494086908a280129f8a985a002 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 24 Oct 2026 23:29:28 -0500 Subject: [PATCH] fix(server): preserve plugin settlements and concurrent deadlines Co-Authored-By: Paperclip --- doc/idle-plugin-jobs.md | 33 ++++++++-- ...2026-10-24-recurring-server-sleep-stack.md | 10 +++ .../__tests__/plugin-scheduled-work.test.ts | 65 +++++++++++++++++++ server/src/services/plugin-job-scheduler.ts | 9 ++- server/src/services/plugin-registry.ts | 1 + 5 files changed, 111 insertions(+), 7 deletions(-) diff --git a/doc/idle-plugin-jobs.md b/doc/idle-plugin-jobs.md index e1cc4c1f2ad..71f1f163d97 100644 --- a/doc/idle-plugin-jobs.md +++ b/doc/idle-plugin-jobs.md @@ -30,7 +30,9 @@ including a worker becoming available without a database mutation. is not replaced merely because the plugin registered again. A missed schedule dispatches once, then computes its next cron time after completion. - A missing timestamp is initialized from the current cron expression. A changed - cron clears the old timestamp so the new schedule can be initialized. + cron clears the old timestamp so the new schedule can be initialized. Registry + updates lock the job before reading its current deadline, so a concurrent + completion cannot have its newly advanced deadline replaced by a stale value. - Claims recheck the definition under a job-row lock and reject outstanding queued, pending, or running records. Manual and scheduled dispatch share this boundary. Worker RPC calls run outside transactions. @@ -39,10 +41,14 @@ including a worker becoming available without a database mutation. 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. - A concurrent schedule edit is preserved by the settlement check. + A concurrent schedule edit is preserved by the settlement check. Failed + settlements do not prevent independent jobs from using available capacity. + The failed job stays excluded from admission until its exact result is saved. - Stop rejects new admissions, detaches worker subscriptions, and waits for - admitted work. Existing calls retain their RPC timeout. The application then - stops the shared coordinator before ending the database pool. + 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. ## Limits @@ -73,3 +79,22 @@ restart and overdue execution; removed schedules; concurrent claims; failed completion persistence without repeated RPC; capacity refill; drain; and shutdown. Worker responses are simulated. Existing startup, authorization, loader, shared coordinator, decision-expiry, and warm-standby tests cover affected consumers. + +## Review repair evidence + +Real PostgreSQL regressions reproduced three failures: shutdown returned with a +known completed call still marked running; one unwritable completion prevented an +independent job from starting; and a concurrent unchanged registry update restored +an old deadline. All three pass after the repair. The shutdown test initially +expected an unresolved-work hold to disappear, which was incorrect. Its corrected +baseline stops the coordinator, then observes the actual running-row failure. +A further case verifies that persistent shutdown-write failure is reported and +a later stop attempt saves the retained result without repeating the RPC. + +A permanent database outage followed by forced process termination still has the +process-loss limitation above. This change does not claim durable storage for an +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. 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 f1735b251ee..3a446cfdd4a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -966,3 +966,13 @@ Plugin scheduling is published as #17518 at 8b4f551dc2. Its Actions checks passe but review found retained completion writes missing from shutdown and a stale deadline update race. Repair those and preserve independent job progress when one completion write fails before treating the slice as ready. + + +Plugin review repair: real PostgreSQL cases reproduced lost shutdown settlement, +independent-job starvation after a completion write error, and restoration of an +old deadline by a concurrent registry update. Stop now flushes retained results +and reports remaining errors; refresh preserves independent dispatch; registry +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. diff --git a/server/src/__tests__/plugin-scheduled-work.test.ts b/server/src/__tests__/plugin-scheduled-work.test.ts index b50937a2c7a..d61f2160eb5 100644 --- a/server/src/__tests__/plugin-scheduled-work.test.ts +++ b/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(resolve => { release = resolve; }); + const acquired = new Promise(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" }); diff --git a/server/src/services/plugin-job-scheduler.ts b/server/src/services/plugin-job-scheduler.ts index b630aa7e6a3..d9a4c3c8e0b 100644 --- a/server/src/services/plugin-job-scheduler.ts +++ b/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); diff --git a/server/src/services/plugin-registry.ts b/server/src/services/plugin-registry.ts index f03c9006436..f7574bb293d 100644 --- a/server/src/services/plugin-registry.ts +++ b/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) {