From 96727b3cfefb61f85ffe1fbc29a9b6dd6837efe8 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 21:07:10 -0500 Subject: [PATCH] fix(server): schedule plugin jobs from saved deadlines (#17518) ## Thinking Path > - Paperclip manages agents and plugin work. > - Plugins can declare cron jobs with saved deadlines. > - The existing scheduler queries for due jobs every 30 seconds, even when none exist. > - Empty queries keep an otherwise idle database active. > - This PR uses the shared scheduler and committed job changes to discover work. > - Existing job and run rows preserve deadlines and execution history. ## Linked Issues or Issue Description **What happened?** The plugin scheduler on the base of this stack runs empty periodic SQL. Its registration path also advances overdue timestamps. An empty replacement manifest can leave removed schedules active. **Expected behavior** No recurring SQL when no eligible jobs exist. Configured jobs wait for their saved deadline. Changes reactivate scheduling after commit. Work remains protected through drain and shutdown. **Steps to reproduce** Start the plugin scheduler with no jobs. After startup, observe queries during a three-minute idle window. The new PostgreSQL regression failed on the baseline and passes with this change. **Paperclip version or commit** Stacked on #17510 at `4d15c08428`. Deployment: source checkout, local PostgreSQL, controlled plugin RPC. Related: #7905 covers empty-manifest sync; this migration requires that behavior on both activation paths. #11571 proposes company-scoped jobs; this PR preserves the existing instance-scoped job contract. ## What Changed - Install the concurrency test barrier before due job notifications can dispatch. - Make dispatch the sole reservation owner and preserve retained or concurrently committed results when unregistering a plugin. - Replace the private interval with the existing shared deadline worker. - Signal job definition and run changes inside their owning transactions. No new tables. - Wake from plugin registration and worker status events. Retain explicit scheduler wakes received during drain. - Preserve overdue work and initialize only missing or changed schedules. - Recheck claims under a job-row lock. Keep plugin calls outside transactions and refill capacity on completion. - Commit completion and schedule advancement together. Retry failed persistence without repeating the plugin side effect. - Track admitted calls through shutdown and synchronize empty manifests. ## Verification This PR now targets master after #17510 merged as `4e27a82a4265c01f3ab4f05aaecc010a0026de75`. The normal landed-parent merge preserves the entire verified `df78ca3b5e` tree byte for byte; only final publication docs change afterwards. The PR still differs in its original seventeen paths and has no database schema or migration change. Primary test revision: `36be0bd73363680d7666d94ff5f1dd28b27f4235`, on parent `ba80060407`. Final integration of parent `51668b8240` preserves all plugin/runtime/shutdown sources from that proof and passes all 63 cases in complete plugin, approval-seeding and scoped-deadline suites at `56ce7c1d2a`, plus server types/build. Only six parent seeding/doc files changed; this is not a rerun of all eleven suites. Only the original seventeen plugin-scheduling paths differ from that parent; all other tracked files, including migration history, match it. - **247 cases pass in eleven complete suites:** plugin scheduler, shutdown, warm standby, plugin environment-driver readiness, delivery coordinator, both retention workers, approval seeding, decision expiry, scoped deadlines and startup. Server typecheck and production build pass. - The new real PostgreSQL lock test failed before the repair: shutdown remained pending while the completion update waited for a held run row. After repair, shutdown reports the statement timeout while the lock remains held; retry after release commits the saved result with exactly one plugin invocation. All 34 scheduler cases pass. - Existing regressions preserve real committed schedule changes, saved deadlines, manual/scheduled reservation, retained success and failure results during unregister, completion/unregister lock order, independent progress, rollback/outer commit, quiet configured states, drain/resume and shutdown cleanup. - The parent's complete seeding/producer run passed 522 cases at its stated revision. Earlier duplicate reservations, overwritten retained results, hourly-clock fixture failures and combined-stack CI/setup failures remain recorded in `doc/idle-plugin-jobs.md`, its dated audit and the stack ledger; they are not erased or recounted as this candidate's tests. - The completion timeout is transaction-local and defaults to thirty seconds per PostgreSQL statement. It does not bound every shutdown/network operation or make plugin side effects safe to replay. No full local repository-suite or deployed host-sleep claim is made. Commands: server Vitest for the eleven suites above; server typecheck/build. Fresh exact-head CI, Greptile and Ponytail reviews remain required before merge. ## Risks - This uses process-local notifications. Independent SQL writers and other replicas need their own wake integration. - A run left unresolved after process loss is not automatically replayed: the external effect may have occurred. Existing lifecycle/operator reconciliation must resolve it. Live-process settlement failures retry without replay. - Job definitions remain instance-scoped. This PR does not implement company-scoped job dispatch. - Cloud's plugin sleep blockers remain. Internal deadlines do not add durable external wake coverage or prove deployed sleep. - The shared deadline helper now retains direct lifecycle wakes through drain; existing coordinator and decision deadline consumers are tested. ## Model Used OpenAI Codex, GPT-6-based agent, with repository tools, code execution, and GitHub access. The exact serving model version and context-window size are not exposed in this session. ## 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 --------- Co-authored-by: Paperclip --- doc/idle-plugin-jobs.md | 203 ++++ .../2026-10-24-plugin-job-sleep-audit.md | 116 +++ ...2026-10-24-recurring-server-sleep-stack.md | 68 ++ .../src/__tests__/cloud-warm-standby.test.ts | 6 +- ...-environment-driver-ready-recovery.test.ts | 4 +- .../__tests__/plugin-scheduled-work.test.ts | 478 +++++++++ server/src/app.ts | 10 +- server/src/services/deadline-work.ts | 3 +- .../services/delivery-work-notifications.ts | 1 + server/src/services/plugin-job-coordinator.ts | 2 +- server/src/services/plugin-job-scheduler.ts | 905 ++++-------------- server/src/services/plugin-job-store.ts | 198 ++-- .../services/plugin-job-work-notifications.ts | 11 + server/src/services/plugin-loader.ts | 11 +- server/src/services/plugin-registry.ts | 121 +-- server/src/shutdown.test.ts | 24 + server/src/shutdown.ts | 15 + 17 files changed, 1313 insertions(+), 863 deletions(-) create mode 100644 doc/idle-plugin-jobs.md create mode 100644 doc/plans/2026-10-24-plugin-job-sleep-audit.md create mode 100644 server/src/__tests__/plugin-scheduled-work.test.ts create mode 100644 server/src/services/plugin-job-work-notifications.ts diff --git a/doc/idle-plugin-jobs.md b/doc/idle-plugin-jobs.md new file mode 100644 index 00000000000..4643e241cff --- /dev/null +++ b/doc/idle-plugin-jobs.md @@ -0,0 +1,203 @@ +# Plugin job scheduling without idle polling + +Plugin cron scheduling uses the instance delivery coordinator. It no longer +queries `plugin_jobs` every 30 seconds when there is no eligible work. + +## Ownership and durable state + +The plugin scheduler owns cron interpretation, dispatch eligibility, concurrency, +and run settlement. `plugin_jobs` stores schedules and their next deadlines; +`plugin_job_runs` records admitted calls and their outcomes. No table or migration +is added. The application passes its existing coordinator to `scheduler.start`. +The coordinator owns timer execution, transaction uncertainty, and drain gating. + +The job store and registry signal before changes inside their transactions. +Definition sync, pause/resume, changed schedules, run creation/settlement, job +removal, and plugin removal invalidate the scheduler. Nested transactions wake +only after their outer transaction settles. An empty manifest is a real sync: +both the loader and lifecycle coordinator pause removed definitions. +Registration and worker status changes provide local lifecycle wakes. The shared +`registerDeadlineWork` helper retains explicit wakes received during drain, +including a worker becoming available without a database mutation. + +## Idle and dispatch behavior + +- Empty instances, manual-only jobs, invalid cron definitions, paused jobs, and + unavailable workers have no recurring scheduler SQL after reconciliation. +- Configured future jobs arm their saved next deadline. No SQL is needed between + reconciliation and that deadline unless a supported producer changes state. +- Startup reconstructs deadlines. An overdue saved timestamp is owed work and + 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. 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. +- Active calls consume concurrency slots and remain tracked until settlement. + 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. + 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, 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. 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 + +This is the existing single-process writer model. Independent SQL writers and +other app replicas do not emit these local signals. The row lock prevents an +overlapping claim, but does not establish a supported multi-replica wake protocol. + +A process crash after a plugin call but before durable completion leaves an +unresolved run. The scheduler does not assume that an unknown external effect is +safe to repeat. Such a run remains outstanding until explicit lifecycle/operator +reconciliation. During a live process, failed completion writes retain their +exact result and retry. This does not invent exactly-once external execution. + +Plugin definitions remain instance-scoped, as on the base branch. The separate +company-scoped job proposal is not implemented here. Plugin configuration and +worker authorization retain their existing owners. + +The Cloud sleep proof remains conservative for plugin work. An internal cron +deadline cannot wake a stopped host; this PR does not add external alarm coverage, +allow configured plugins to bypass idle checks, or claim measured host sleep. + +## Verification + +`plugin-scheduled-work.test.ts` uses real PostgreSQL and controlled worker calls. +It checks idle SQL silence with no jobs and configured future/manual-only jobs; +first writes, pause/resume, worker availability, outer commit and rollback; +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. + + +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. + + +## Reservation and unregister races + +A scheduled refresh can yield after selecting eligible jobs while a manual trigger +reserves one of those jobs. Previously the stale scheduled attempt reused the +same set entry, found the manual run under its row lock, and removed that entry. +The manual provider call was still running but no longer consumed scheduler +capacity. Dispatch now performs the synchronous reservation check and insertion +at one boundary; callers cannot remove a reservation they did not create. Refresh +also rechecks active and retained-settlement state after its database read. + +Unregister now preserves runs whose real result is retained for settlement. For +other unfinished runs, it locks and rereads the current run before cancelling, so +a completion committed since discovery cannot be overwritten. The existing job +store still owns mutation notifications. Provider calls stay outside transactions; +no new periodic work or durable storage is introduced. + +Real PostgreSQL regressions first reproduced both incorrect cancellation of saved +success/failure results and loss of a manual reservation after a delayed query +result. The corrected suite also checks concurrent completion during unregister, +manual/scheduled capacity, retries without another provider call, drain/resume, +startup, shutdown and quiet states. All 33 complete scheduler cases and server +TypeScript checks pass on the owning slice. The first full run crossed the next +hourly cron boundary during a retry fixture and legitimately produced extra runs; +its fixed clock now isolates that test from wall-clock start time while preserving +its assertions. Stack integration and latest-head CI are separate gates. + + +## Bounded completion writes + +A retained completion transaction now sets a PostgreSQL statement timeout local +to that transaction, thirty seconds by default. A database row lock cannot hold +the completion attempt forever. Both ordinary reconciliation and the final +shutdown flush use the same retained writer. On timeout, the result remains in +memory and the existing recovery owner can retry it without invoking the plugin +again. Shutdown reports the persistence failure and continues through the existing +cleanup/retry boundary. The timeout resets when the transaction ends; no shared +pool setting or polling loop is added. + +The bound applies to PostgreSQL statement execution and lock waits. It is not a +new bound on the plugin RPC, every shutdown operation, or a network failure before +PostgreSQL receives a statement. A retained result is process-local; after process +loss, existing durable run rows and their existing orphan policy remain the source +of truth. This change does not make arbitrary plugin side effects replayable. + +A real PostgreSQL regression first observed a completion waiting for a held run +row, then demonstrated that shutdown remained pending. With the repair, PostgreSQL +cancels the write, shutdown reports failure while the lock is still held, and the +saved completion commits after release without a second plugin call. All 34 +scheduler cases pass. All 247 cases in eleven complete scheduler/shutdown/startup, +warm-standby, driver-readiness, coordinator, scoped-deadline, approval/join +retention, approval seeding and decision-expiry suites pass, along with server +typecheck and production build, at `36be0bd733`. Earlier failures and their stated +revisions remain above. Fresh final-parent integration and exact-head remote +checks/reviews remain required. + + +Final parent `51668b8240` changes only approval-seeding behavior/tests and two +documents relative to the 247-case candidate. All plugin implementation, plugin +tests, shared deadline/coordinator and shutdown sources remain identical to that +run. After normal integration, all 63 cases in complete plugin-scheduler, +approval-seeding and scoped-deadline suites pass at `56ce7c1d2a`, plus server +typecheck/build. The original seventeen-file plugin boundary and exact parent +migration metadata are preserved. Publication waits for the parent to land so +final master reconciliation and remote checks can run together. + + +Approval seeding #17510 merged as `4e27a82a42`, with both final-head 5/5 reviews +and all 58 accepted checks. Plugin scheduling now integrates that actual master +parent. The complete integrated tree is byte-identical to `df78ca3b5e`, so the +247-case and final 63-case verification above retain their exact source scope. +No local checks were repeated on unchanged code. Only final publication docs +change afterwards; fresh exact-head CI and reviews remain required against master. diff --git a/doc/plans/2026-10-24-plugin-job-sleep-audit.md b/doc/plans/2026-10-24-plugin-job-sleep-audit.md new file mode 100644 index 00000000000..b31bf007355 --- /dev/null +++ b/doc/plans/2026-10-24-plugin-job-sleep-audit.md @@ -0,0 +1,116 @@ +# Plugin job scheduling migration audit + +Baseline: approval queue slice `e04a0375dc` (#17510), with fetched master +`a3675ddcdf`. This is the next independent recurring-server obligation; it does +not replace the other outstanding inventory items. + +## Current owner and behavior + +`plugin-job-scheduler.ts` starts a private 30-second interval from `app.ts`. +Each tick selects active jobs with a due `nextRunAt`, even on an empty instance. +Cron calculation and dispatch already belong to this service; the migration +should retain that ownership and register deadlines with the existing delivery +coordinator. It must not create a second shared scheduler or one table per job. + +Saved job definitions and run records are the durable truth. Plugin lifecycle +registration initializes next-run timestamps. Job pause/resume, manifest sync, +job deletion, and run completion can all change the next eligible obligation. +The scheduler currently catches tick errors; its scheduled entry point must not +swallow errors that the shared coordinator needs for recovery. + +## Producer and lifecycle audit + +- `plugin-job-store.ts`: declaration sync, definition status changes, timestamp + updates, and deletion. These are the primary persistence boundary. +- `plugin-registry.ts`: its separate declaration synchronization also directly + inserts/updates jobs. Do not assume the job store covers all writers. +- `plugin-job-coordinator.ts`: loaded/disabled/unloaded events connect the store + and scheduler. The loaded handler skips declaration sync when the new manifest + has no jobs, which can leave old definitions. Preserve or explicitly repair + that behavior before claiming removed schedules are quiet. +- `plugin-loader.ts`: also registers/unregisters jobs and stops the scheduler. + Inspect actual worker start/restart events; a worker becoming available must + wake overdue jobs without relying on periodic discovery. +- `app.ts`: creates the scheduler before the delivery coordinator and starts it + just before coordinator creation. Wiring must pass the existing coordinator + after creation and retain correct shutdown ordering. + +## Correctness questions to settle before implementation + +1. Registration currently advances overdue timestamps as well as missing ones. + An observation must not silently move a real due deadline. Establish and test + startup/catch-up behavior rather than copying this as deadline discovery. +2. Scheduled dispatch tracks jobs in memory, while manual dispatch also checks + saved running records. Replacing polling must not replay an unknown prior + provider effect or lose delivery; audit interrupted run recovery separately. +3. Concurrency limits and a temporarily unavailable worker defer actual work. + Completion/availability signals should wake it. They are not permission for + an empty recurring scan. +4. Invalid or missing cron expressions and disabled definitions must not cause + an endless past-due deadline. Preserve manual trigger behavior. +5. Shutdown currently stops its timer while in-flight jobs finish naturally. + The shared owner must still track and await those admitted operations, with + durable results and schedule advancement preserved. +6. The current status update and declaration-sync methods need outer-commit + signals. Never put worker RPC calls inside a database transaction. + +Prove empty, configured/manual-only, paused, and completed-history quiet; +first actual declaration/status changes; future deadlines; restart/catch-up; +worker unavailability and recovery; failures; drain/resume; concurrent writes; +and shutdown. Cloud alarm coverage for plugin jobs remains unsupported and +conservative sleep admission must stay intact. + +## Other audit observations retained for later slices + +Tool connection health checks have a distinct 15-minute proactive policy in +`tool-access/catalog/health.ts`; the aggregate tick repeatedly discovers its due +rows. About 94 direct tool-connection write sites exist, many for excluded chat +or AI transports. Classify those writers instead of adding wrappers indiscriminately. +The current health checker returns immediately for AI-purpose connections, while +the sweep does not explicitly exclude that purpose. Audit inert due rows and +pre-probe setup failures before deriving deadlines from `healthCheckedAt`. +No tool-health migration is claimed by this plugin-job slice. + +## Implementation decisions + +The migration now uses existing job/run rows and the shared deadline worker. +Local lifecycle wakes needed the helper to retain an explicit wake across drain; +the plugin tests exercise that condition through a real status event. The worker +handle already exposes status subscriptions, so no new worker-manager API is +needed. The job store accepts the query/transaction subset shared by root clients +and transactions instead of casting transaction handles to full pools. + +Claims recheck current state with a job-row lock. Worker calls stay outside the +transaction. Completion and schedule advancement commit atomically; failed +persistence retries only settlement. Unresolved old runs are not blindly replayed. +This preserves the conservative Cloud sleep blocker for unknown external effects. + +Related proposals: #7905 covers empty manifest sync; this migration needs that +behavior on both activation paths. #11571 proposes company-scoped jobs; this +migration retains the current instance-scoped contract and does not implement it. + +## Refresh/manual-dispatch test ordering (2026-10-27) + +CI job 122104710882 timed out in the manual-reservation concurrency case. Its +setup created due work before installing the outstanding-run snapshot barrier. +The creation notification could start dispatch first, leaving no runnable job +for the later refresh to query and no read-barrier arrival. An isolated local +run passed, so that retry alone did not establish the cause of the CI failure. + +The test now pauses background admission during setup and resumes it only after +the barrier is installed. It still holds a real database snapshot across the +manual dispatch, verifies the active reservation and concurrency bound, and +checks that all three jobs finish. All 33 plugin scheduling cases pass with +this ordering. Production scheduling, assertions, and test timeouts are unchanged. + + +## October 28 review repair + +The final retained completion transaction uses a local thirty-second PostgreSQL +statement timeout (configurable through the internal constructor for tests). It +bounds lock/query execution in both refresh and stop, preserves pending results +on error and never repeats the plugin RPC to repair persistence. A two-connection +lock regression failed before this repair, then passed. All 34 scheduler cases and +247 cases across eleven complete affected/lifecycle suites pass at `36be0bd733`; +server types/build pass. No schema change or new periodic activity is introduced. +See `doc/idle-plugin-jobs.md` for the exact scope and network/process-loss limits. 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 56ecd467c84..8d0416e6961 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1176,6 +1176,26 @@ and #17497 were each rerun once to clear stale cancelled contexts. The #17490 retry has a new E2E shard 3 failure under investigation; do not call it green or rerun again without a repair. #17497's retry was still running at this observation. +### Plugin scheduling implementation, 2026-10-24 + +The next slice replaces the plugin scheduler's private 30-second query loop with +shared deadlines and transaction-aware producer wakes. It uses existing job/run +rows, preserves overdue schedules, handles empty manifest sync, and tracks calls +through durable settlement and shutdown. See [the implementation contract](../idle-plugin-jobs.md) +and [producer audit](2026-10-24-plugin-job-sleep-audit.md). + +The baseline empty-instance regression failed before the migration. The first +expanded local run passed 59 tests across plugin scheduling, warm standby, the +shared coordinator, and decision expiry. The consumer run exposed an incomplete +loader mock after empty-manifest synchronization became mandatory; that fixture +was corrected; all 14 loader recovery cases passed. Workspace typechecks and the +production build passed. Final server typecheck and build also passed after the +lifecycle wiring edits. Remote review/checks remain open. + +Separate stack gates remain: #17490 E2E cleanup hit a cancellation lock timeout; +#17510 has a security finding about unscoped approval queue scans plus CI failures. +Neither PR is claimed green. Approval retention #17497 is green at `2c5babcabc`. +The complete original inventory remains in scope; this slice does not finish it. Slice 16 security review reproduced cross-company work amplification from normal producer writes. The repair carries company hints through all supported producers @@ -1192,6 +1212,29 @@ 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. + + +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. + + Approval queue follow-up: persistent projection failure reproduced 120 failed transactions in two simulated minutes. Per-scope exponential backoff now caps retries at one minute without discarding debt or letting repeated signals reset @@ -1223,3 +1266,28 @@ cases passed on three fresh fixtures, and the final added guard passes all 56 cases on those migrated fixtures. Server types/build pass at `7948136ef8`. The earlier cleanup-expectation failure and 522-case integration evidence retain their own revisions. Fresh exact-head reviews and CI are the remaining merge gate. + +Slice 17 current review repair: a retained plugin completion can no longer wait +indefinitely for a PostgreSQL statement/row lock. The transaction-local timeout +preserves the saved result for retry and allows existing shutdown cleanup to +continue; the provider is not called again. A real lock regression failed then +passed. All 34 scheduler cases and 247 cases in eleven complete lifecycle and +existing-worker suites pass at `36be0bd733`, plus server types/build. This is not +a guarantee about every network or shutdown operation. Final parent integration, +fresh remote review and all CI gates remain required. + + +Plugin slice 17 integrates final seeding parent `51668b8240`. Its original +seventeen-file boundary is preserved, and every plugin/runtime/shutdown source +is unchanged from the 247-case proof. All 63 cases across complete plugin, +seeding and scoped-deadline suites now pass at `56ce7c1d2a`, plus server +types/build. Hold its final publication for the parent's accepted merge, then +reconcile the landed master parent and require fresh exact-head review/CI. + + +Approval queue seeding #17510 merged as `4e27a82a42` after final-head 5/5 +Greptile/Ponytail and all 58 accepted checks. Plugin scheduling now targets master +and integrates that accepted parent. Normal ancestry reconciliation preserves +its entire verified tree (`df78ca3b5e`) byte for byte; only publication docs change +subsequently. All original remaining duties remain in scope, beyond the 65 +published numbered slices, including native backups and remaining aggregate owners. diff --git a/server/src/__tests__/cloud-warm-standby.test.ts b/server/src/__tests__/cloud-warm-standby.test.ts index e5d8bbfdcd5..4a73563341b 100644 --- a/server/src/__tests__/cloud-warm-standby.test.ts +++ b/server/src/__tests__/cloud-warm-standby.test.ts @@ -140,16 +140,18 @@ describe("unclaimed Cloud background work", () => { tickIntervalMs: 1000, isBackgroundWorkEnabled: () => enabled, }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, onError: () => {} }); try { email.start(); - scheduler.start(); + await scheduler.start(coordinator); await vi.advanceTimersByTimeAsync(10 * 60_000); expect(select).not.toHaveBeenCalled(); enabled = true; await vi.advanceTimersByTimeAsync(1000); expect(select.mock.calls.length).toBeGreaterThanOrEqual(2); } finally { - scheduler.stop(); + await scheduler.stop(); + await coordinator.stop(); await email.shutdown(); } }); diff --git a/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts b/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts index 766a9e4fc54..533df782fc0 100644 --- a/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts +++ b/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts @@ -439,7 +439,7 @@ describe("listReadyPluginEnvironmentDrivers worker recovery", () => { clearPlugin: vi.fn(), subscriptionCount: vi.fn(() => 0), }; - const jobScheduler = { unregisterPlugin: vi.fn().mockResolvedValue(undefined) }; + const jobScheduler = { registerPlugin: vi.fn().mockResolvedValue(undefined), unregisterPlugin: vi.fn().mockResolvedValue(undefined) }; const toolDispatcher = { unregisterPluginTools: vi.fn(), registerPluginTools: vi.fn() }; const loader = pluginLoader( @@ -454,7 +454,7 @@ describe("listReadyPluginEnvironmentDrivers worker recovery", () => { workerManager, eventBus, jobScheduler, - jobStore: {}, + jobStore: { syncJobDeclarations: vi.fn().mockResolvedValue(undefined) }, toolDispatcher, buildHostHandlers: vi.fn(() => ({})), instanceInfo: { diff --git a/server/src/__tests__/plugin-scheduled-work.test.ts b/server/src/__tests__/plugin-scheduled-work.test.ts new file mode 100644 index 00000000000..ca25ef2d73b --- /dev/null +++ b/server/src/__tests__/plugin-scheduled-work.test.ts @@ -0,0 +1,478 @@ +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"; +import { eq, sql } from "drizzle-orm"; +import { createDb, plugins, pluginJobs, pluginJobRuns } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createPluginJobScheduler } from "../services/plugin-job-scheduler.js"; +import { pluginJobStore } from "../services/plugin-job-store.js"; +import { pluginRegistryService } from "../services/plugin-registry.js"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PLUGIN_SCHEDULE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("plugin job scheduling without idle polling", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, pluginId: string; + let scheduler: ReturnType | undefined; + let coordinator: ReturnType | undefined; + let available = true, draining = false; + let events: EventEmitter; + const rpc = vi.fn<(...args: unknown[]) => Promise>(); + const errors: unknown[] = []; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("plugin-schedule-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 30_000); + beforeEach(async () => { + pluginId = randomUUID(); available = true; draining = false; events = new EventEmitter(); errors.length = 0; + rpc.mockReset(); rpc.mockResolvedValue({}); + await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: pluginId, version: "1.0.0", manifestJson: {} as never }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + // Retry tests must not accidentally cross the next real hourly cron tick. + vi.setSystemTime(new Date("2026-10-27T12:05:00Z")); + }); + afterEach(async () => { + await scheduler?.stop(); await coordinator?.stop(); scheduler = undefined; coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`drop trigger if exists plugin_settlement_failure on plugin_job_runs`); + await db.execute(sql`drop function if exists plugin_settlement_failure()`); + await db.delete(plugins).where(eq(plugins.id, pluginId)); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + async function start(maxConcurrentJobs = 10, settlementTimeoutMs = 30_000) { + const workerManager = { isRunning: () => available, call: rpc, getWorker: () => events } as unknown as PluginWorkerManager; + scheduler = createPluginJobScheduler({ db, jobStore: pluginJobStore(db), workerManager, + isBackgroundWorkEnabled: () => !draining, maxConcurrentJobs, settlementTimeoutMs }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, + onError: error => errors.push(error) }); + await scheduler.start(coordinator); + await scheduler.registerPlugin(pluginId); + } + const job = (key = "hourly", schedule = "0 * * * *") => pluginRegistryService(db).upsertJob(pluginId, key, { schedule }); + const runs = () => db.select().from(pluginJobRuns).where(eq(pluginJobRuns.pluginId, pluginId)); + async function settled() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + } + async function quiet() { + 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(); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); + } + async function due(key = "hourly") { + const row = await job(key); + await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), new Date(Date.now() - 1000)); + return row; + } + async function completed(count = 1) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect((await runs()).filter(run => run.status === "succeeded" || run.status === "failed")).toHaveLength(count); + }, { timeout: 5000 }); + } + + 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("honors a saved deadline inside the notification batching window", async () => { + const row = await job(); + const deadline = new Date(Date.now() + 10_000); + await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), deadline); + await start(); + // registerPlugin produces a local wake immediately after startup. It may + // batch the reload but must not replace the known ten-second deadline. + await settled(); + expect(coordinator!.nextWakeAt()).toBe(deadline.getTime()); + expect(rpc).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(Math.max(0, deadline.getTime() - Date.now())); + await completed(); + expect(rpc).toHaveBeenCalledOnce(); + expect(rpc.mock.calls[0][2]).toMatchObject({ job: { scheduledAt: deadline.toISOString() } }); + }); + 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(); + }); + it("initializes a first declaration after empty startup and runs only at its deadline", async () => { + await start(); await quiet(); const row = await job(); await settled(); + const current = await pluginJobStore(db).getJobById(row.id); + expect(current?.nextRunAt!.getTime()).toBeGreaterThan(Date.now()); + expect(rpc).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(current!.nextRunAt!.getTime() - Date.now()); await completed(); expect(rpc).toHaveBeenCalledTimes(1); + }); + it("catches up an overdue startup job without moving its owed timestamp", async () => { + const row = await due(); const at = (await pluginJobStore(db).getJobById(row.id))!.nextRunAt!; + await start(); await completed(); + expect(rpc.mock.calls[0][2]).toMatchObject({ job: { scheduledAt: at.toISOString(), trigger: "schedule" } }); + expect((await pluginJobStore(db).getJobById(row.id))!.nextRunAt!.getTime()).toBeGreaterThan(Date.now()); + }); + it("pauses a due schedule and reactivates from the real status producer", async () => { + const row = await due(); await pluginJobStore(db).updateJobStatus(row.id, "paused"); await start(); await quiet(); + await pluginJobStore(db).updateJobStatus(row.id, "active"); await completed(); expect(rpc).toHaveBeenCalledTimes(1); + }); + it("parks an unavailable worker and wakes overdue work from its lifecycle event", async () => { + await due(); available = false; await start(); await quiet(); + available = true; events.emit("status", { status: "ready" }); await completed(); expect(rpc).toHaveBeenCalledTimes(1); + }); + it("retains a worker-availability wake received during drain", async () => { + await due(); available = false; await start(); await quiet(); + draining = true; available = true; events.emit("status", { status: "ready" }); + await vi.advanceTimersByTimeAsync(10_000); expect(rpc).not.toHaveBeenCalled(); + draining = false; await completed(); + }); + it("retains the first committed job created during drain", async () => { + await start(); await quiet(); draining = true; await due(); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(10_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect(rpc).not.toHaveBeenCalled(); draining = false; await completed(); + }); + it("does not dispatch a nested write until the outer transaction commits", async () => { + await start(); await quiet(); + await db.transaction(async tx => { + await pluginJobStore(tx).syncJobDeclarations(pluginId, [{ jobKey: "outer", schedule: "0 * * * *" }]); + const row = await pluginJobStore(tx).getJobByKey(pluginId, "outer"); + await pluginJobStore(tx).updateRunTimestamps(row!.id, new Date(0), new Date(Date.now() - 1000)); + await vi.advanceTimersByTimeAsync(5000); expect(rpc).not.toHaveBeenCalled(); + }); + await completed(); + }); + it("disarms after a rolled-back declaration", async () => { + await start(); await quiet(); + await expect(db.transaction(async tx => { + await pluginJobStore(tx).syncJobDeclarations(pluginId, [{ jobKey: "rollback", schedule: "* * * * *" }]); + throw new Error("abort"); + })).rejects.toThrow("abort"); + await quiet(); expect(rpc).not.toHaveBeenCalled(); + }); + it("parks removed declarations while retaining their history", async () => { + await due(); await start(); await completed(); + await pluginJobStore(db).syncJobDeclarations(pluginId, []); await quiet(); expect(await runs()).toHaveLength(1); + }); + it("recomputes a changed cron schedule", async () => { + await job("changed", "0 0 1 1 *"); await start(); await settled(); + await job("changed", "* * * * *"); await settled(); + expect(coordinator!.nextWakeAt()! - Date.now()).toBeLessThanOrEqual(60_000); + await vi.advanceTimersByTimeAsync(60_000); await completed(); + }); + it("retries failed completion persistence without repeating the plugin side effect", 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 }); 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); + }); + 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(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; + await scheduler!.stop(); + 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(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"); + 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(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); + 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.each(["succeeded", "failed"] as const)("retains a completed %s result when unregistering during a settlement failure", async outcome => { + 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()`); + if (outcome === "failed") rpc.mockRejectedValueOnce(new Error("Plugin failed before unregister")); + await due(); await start(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(errors.length).toBeGreaterThan(0); + expect(scheduler!.diagnostics().activeJobCount).toBe(0); + }, { timeout: 5000 }); + try { + await scheduler!.unregisterPlugin(pluginId); + expect((await runs())[0].status).toBe("running"); + } finally { + await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); + errors.length = 0; + await scheduler!.tick(); await settled(); errors.length = 0; + } + expect((await runs())[0].status).toBe(outcome); + expect(rpc).toHaveBeenCalledOnce(); + await quiet(); + }); + + it("preserves a completion committed while unregister waits on the discovered run", async () => { + const row = await job("finishing", ""); + const run = await pluginJobStore(db).createRun({ jobId: row.id, pluginId, trigger: "manual" }); + await pluginJobStore(db).markRunning(run.id); + await start(); await quiet(); + let release!: () => void, locked!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const acquired = new Promise(resolve => { locked = resolve; }); + const finishing = db.transaction(async tx => { + await pluginJobStore(tx).completeRun(run.id, { status: "succeeded" }); + locked(); await gate; + }); + await acquired; + const unregistering = scheduler!.unregisterPlugin(pluginId); + 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' + and query like '%plugin_job_runs%'`); + expect(waiting.count).toBeGreaterThan(0); + }); + } finally { release(); await finishing; await unregistering; } + expect((await runs())[0].status).toBe("succeeded"); + expect(rpc).not.toHaveBeenCalled(); + await quiet(); + }); + + it("retains the manual reservation when a refresh read finishes after manual dispatch", async () => { + await start(2); await quiet(); + // Creating due work also wakes the scheduler. Keep dispatch paused until + // the snapshot barrier is installed, regardless of notification timing. + draining = true; + const row = await due("raced"); + let resumeRead!: () => void, enteredRead!: () => void, releaseCall!: () => void; + const readGate = new Promise(resolve => { resumeRead = resolve; }); + const readEntered = new Promise(resolve => { enteredRead = resolve; }); + const callGate = new Promise(resolve => { releaseCall = resolve; }); + rpc.mockImplementation(() => callGate); + const select = db.select.bind(db); + let held = false; + // Hold delivery of one real outstanding-run query result. The manual run + // commits after this snapshot, before the scheduled dispatch resumes. + const readSpy = vi.spyOn(db, "select").mockImplementation((fields) => { + const builder = select(fields); + if (fields?.jobId !== pluginJobRuns.jobId || held) return builder; + held = true; + const from = builder.from.bind(builder); + vi.spyOn(builder, "from").mockImplementation((table) => { + const query = from(table); + const then = query.then.bind(query); + query.then = ((fulfilled, rejected) => then(async rows => { + enteredRead(); await readGate; return rows; + }).then(fulfilled, rejected)) as typeof query.then; + return query; + }); + return builder; + }); + draining = false; + const refreshing = scheduler!.tick(); + try { + await readEntered; + await scheduler!.triggerJob(row.id); + await vi.waitFor(() => expect(rpc).toHaveBeenCalledOnce()); + resumeRead(); await refreshing; + expect(scheduler!.diagnostics().activeJobIds).toEqual([row.id]); + await due("waiting-one"); await due("waiting-two"); await scheduler!.tick(); + await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(2)); + expect(scheduler!.diagnostics().activeJobCount).toBe(2); + } finally { + resumeRead(); readSpy.mockRestore(); + await pluginJobStore(db).updateJobStatus(row.id, "paused"); + releaseCall(); await refreshing; await settled(); + } + await completed(3); + expect(rpc).toHaveBeenCalledTimes(3); + }); + + 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" }); + await settled(); expect(coordinator!.nextWakeAt()! > Date.now()).toBe(true); + }); + it("does not replay an unresolved run left by another process", async () => { + const row = await due(); await pluginJobStore(db).createRun({ jobId: row.id, pluginId, trigger: "schedule" }); + await start(); await quiet(); expect(rpc).not.toHaveBeenCalled(); + const [run] = await runs(); await pluginJobStore(db).completeRun(run.id, { status: "cancelled" }); await completed(); + }); + it("fills released capacity from completion without a poll", async () => { + await due("one"); await due("two"); let release!: () => void; + rpc.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + const starting = start(1); await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(1)); + release(); await starting; await completed(2); expect(rpc).toHaveBeenCalledTimes(2); + }); + it("tracks manual work and waits for admitted calls during shutdown", async () => { + const row = await job("manual", ""); await start(); await quiet(); let release!: () => void; + rpc.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + await scheduler!.triggerJob(row.id); await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(1)); + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + let stopped = false; const stopping = scheduler!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false); + release(); await stopping; expect((await runs())[0].status).toBe("succeeded"); + }); + 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 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(); + reads.mockRestore(); writes.mockRestore(); expect(rpc).not.toHaveBeenCalled(); + }); + it("reconstructs the saved deadline after scheduler restart", 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(); await scheduler!.stop(); await coordinator!.stop(); + 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 () => { + const row = await due(); let startup!: Promise; + await db.transaction(async tx => { + await pluginJobStore(tx).updateJobStatus(row.id, "paused"); + startup = start(); await vi.advanceTimersByTimeAsync(1000); expect(rpc).not.toHaveBeenCalled(); + }); + await startup; await quiet(); expect(rpc).not.toHaveBeenCalled(); + }); + it("prevents overlapping concurrent manual triggers", async () => { + const row = await job("manual", ""); await start(); await quiet(); let release!: () => void; + rpc.mockImplementation(() => new Promise(resolve => { release = resolve; })); + const attempts = await Promise.allSettled([scheduler!.triggerJob(row.id), scheduler!.triggerJob(row.id)]); + await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(1)); + release(); await completed(); expect(attempts.filter(result => result.status === "fulfilled")).toHaveLength(1); + }); + it("bounds a locked completion write through shutdown and retries without rerunning the plugin", async () => { + const row = await job("manual", ""); await start(10, 250); await quiet(); + let finishRpc!: () => void; + rpc.mockImplementation(() => new Promise(resolve => { finishRpc = resolve; })); + const { runId } = await scheduler!.triggerJob(row.id); + await vi.waitFor(() => expect(rpc).toHaveBeenCalledOnce()); + let unlock!: () => void, acquired!: () => void; + const gate = new Promise(resolve => { unlock = resolve; }); + const admitted = new Promise(resolve => { acquired = resolve; }); + const holder = db.transaction(async tx => { + await tx.select().from(pluginJobRuns).where(eq(pluginJobRuns.id, runId)).for("update"); + acquired(); await gate; + }); + await admitted; + let stop: Promise | undefined, stopped = false, failure: unknown; + try { + draining = true; finishRpc(); + await vi.waitFor(async () => { + const waits = await db.execute(sql`select 1 from pg_stat_activity where datname = current_database() + and pid <> pg_backend_pid() and wait_event_type = 'Lock' and query like 'update "plugin_job_runs"%'`); + expect(waits.length).toBe(1); + }); + stop = scheduler!.stop().then(() => { stopped = true; }, error => { failure = error; stopped = true; }); + await vi.waitFor(() => expect(stopped).toBe(true), { timeout: 5000 }); + expect(failure).toBeInstanceOf(AggregateError); + expect((failure as AggregateError).errors).toEqual([expect.objectContaining({ cause: expect.objectContaining({ code: "57014" }) })]); + expect(await runs()).toMatchObject([{ id: runId, status: "running" }]); + } finally { + unlock(); await holder; await stop; + } + await scheduler!.stop(); + expect(await runs()).toMatchObject([{ id: runId, status: "succeeded" }]); + expect(rpc).toHaveBeenCalledOnce(); + expect(errors).toEqual([]); + }); + + it("deleting the last job cancels its future deadline", async () => { + await job(); await start(); await settled(); await pluginJobStore(db).deleteAllJobs(pluginId); await quiet(); + }); +}); diff --git a/server/src/app.ts b/server/src/app.ts index f5f083ebf38..ad36ece1182 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"; @@ -1267,7 +1268,6 @@ export async function createApp( app.use(errorHandler); jobCoordinator.start(); - scheduler.start(); const deliveryWork = createDeliveryWorkCoordinator({ owner: db, canRun: () => !isWarmStandby() && !isIdleTaskDrainActive(), @@ -1275,6 +1275,7 @@ export async function createApp( onError: (err, queue) => logger.error({ err, queue }, "Delivery reconciliation failed"), }); app.locals.deliveryWork = deliveryWork; + await scheduler.start(deliveryWork); const agentLifecycleDelivery = deliveryWork.register(DELIVERY_QUEUES.agentLifecycle, { retryMs: 1_000, run: () => agentLifecycle.sweep(), @@ -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. - 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/deadline-work.ts b/server/src/services/deadline-work.ts index 31482ccd8a0..50df3a48db5 100644 --- a/server/src/services/deadline-work.ts +++ b/server/src/services/deadline-work.ts @@ -18,7 +18,7 @@ export function registerDeadlineWork( let next: number | null = null; let earliest = 0; let coalesceUntil = 0; - return coordinator.register(queue, { + const worker = coordinator.register(queue, { retryMs: task.retryMs, pausePassiveMaintenance: true, onSettled: () => { requested++; }, @@ -45,4 +45,5 @@ export function registerDeadlineWork( hasPending: async () => requested !== reconciled, nextRunAt: () => next, }); + return { ...worker, wake: () => { requested++; worker.wake(); } }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 9a42db08f16..f4f3a3f613c 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -62,6 +62,7 @@ export const DELIVERY_QUEUES = { gitRestorePoints: "git-restore-points", nativeWorkspaceSyncRetention: "native-workspace-sync-retention", question: "question-responses", + pluginSchedules: "plugin-schedules", pluginLifecycle: "plugin-resource-lifecycle", toolAction: "tool-action-receipts", terminalWorkspaces: "terminal-workspaces", diff --git a/server/src/services/plugin-job-coordinator.ts b/server/src/services/plugin-job-coordinator.ts index bb0df0836e7..47ef9fab690 100644 --- a/server/src/services/plugin-job-coordinator.ts +++ b/server/src/services/plugin-job-coordinator.ts @@ -130,8 +130,8 @@ export function createPluginJobCoordinator( { pluginId, pluginKey, jobCount: jobDeclarations.length }, "syncing job declarations from manifest", ); - await jobStore.syncJobDeclarations(pluginId, jobDeclarations); } + await jobStore.syncJobDeclarations(pluginId, jobDeclarations); // Register with the scheduler (computes nextRunAt for active jobs) await scheduler.registerPlugin(pluginId); diff --git a/server/src/services/plugin-job-scheduler.ts b/server/src/services/plugin-job-scheduler.ts index a7044f2e8dd..9081521fb87 100644 --- a/server/src/services/plugin-job-scheduler.ts +++ b/server/src/services/plugin-job-scheduler.ts @@ -1,755 +1,264 @@ -/** - * PluginJobScheduler — tick-based scheduler for plugin scheduled jobs. - * - * The scheduler is the central coordinator for all plugin cron jobs. It - * periodically ticks (default every 30 seconds), queries the `plugin_jobs` - * table for jobs whose `nextRunAt` has passed, dispatches `runJob` RPC calls - * to the appropriate worker processes, records each execution in the - * `plugin_job_runs` table, and advances the scheduling pointer. - * - * ## Responsibilities - * - * 1. **Tick loop** — A `setInterval`-based loop fires every `tickIntervalMs` - * (default 30s). Each tick scans for due jobs and dispatches them. - * - * 2. **Cron parsing & next-run calculation** — Uses the lightweight built-in - * cron parser ({@link parseCron}, {@link nextCronTick}) to compute the - * `nextRunAt` timestamp after each run or when a new job is registered. - * - * 3. **Overlap prevention** — Before dispatching a job, the scheduler checks - * for an existing `running` run for the same job. If one exists, the job - * is skipped for that tick. - * - * 4. **Job run recording** — Every execution creates a `plugin_job_runs` row: - * `queued` → `running` → `succeeded` | `failed`. Duration and error are - * captured. - * - * 5. **Lifecycle integration** — The scheduler exposes `registerPlugin()` and - * `unregisterPlugin()` so the host lifecycle manager can wire up job - * scheduling when plugins start/stop. On registration, the scheduler - * computes `nextRunAt` for all active jobs that don't already have one. - * - * @see PLUGIN_SPEC.md §17 — Scheduled Jobs - * @see ./plugin-job-store.ts — Persistence layer - * @see ./cron.ts — Cron parsing utilities - */ - -import { and, eq, lte, or } from "drizzle-orm"; -import type { Db } from "@paperclipai/db"; -import { pluginJobs, pluginJobRuns } from "@paperclipai/db"; -import type { PluginJobStore } from "./plugin-job-store.js"; +import { and, eq, inArray, sql } from "drizzle-orm"; +import { pluginJobs, pluginJobRuns, type Db } from "@paperclipai/db"; +import { pluginJobStore, type PluginJobStore } from "./plugin-job-store.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { parseCron, nextCronTick, validateCron } from "./cron.js"; +import { trackIdleWork } from "./task-admission.js"; import { logger } from "../middleware/logger.js"; -// --------------------------------------------------------------------------- -// Constants -// --------------------------------------------------------------------------- +type Job = typeof pluginJobs.$inferSelect; +type Trigger = "schedule" | "manual" | "retry"; +type Coordinator = ReturnType; -/** Default interval between scheduler ticks (30 seconds). */ -const DEFAULT_TICK_INTERVAL_MS = 30_000; - -/** Default timeout for a runJob RPC call (5 minutes). */ -const DEFAULT_JOB_TIMEOUT_MS = 5 * 60 * 1_000; - -/** Maximum number of concurrent job executions across all plugins. */ -const DEFAULT_MAX_CONCURRENT_JOBS = 10; - -// --------------------------------------------------------------------------- -// Types -// --------------------------------------------------------------------------- - -/** - * Options for creating a PluginJobScheduler. - */ export interface PluginJobSchedulerOptions { - /** Suppress periodic database work while an unclaimed Cloud app stands by. */ isBackgroundWorkEnabled?: () => boolean; - /** Drizzle database instance. */ db: Db; - /** Persistence layer for jobs and runs. */ jobStore: PluginJobStore; - /** Worker process manager for RPC calls. */ workerManager: PluginWorkerManager; - /** Interval between scheduler ticks in ms (default: 30s). */ tickIntervalMs?: number; - /** Timeout for individual job RPC calls in ms (default: 5min). */ jobTimeoutMs?: number; - /** Maximum number of concurrent job executions (default: 10). */ + settlementTimeoutMs?: number; maxConcurrentJobs?: number; } - -/** - * Result of a manual job trigger. - */ -export interface TriggerJobResult { - /** The created run ID. */ - runId: string; - /** The job ID that was triggered. */ - jobId: string; -} - -/** - * Diagnostic information about the scheduler. - */ +export interface TriggerJobResult { runId: string; jobId: string } export interface SchedulerDiagnostics { - /** Whether the tick loop is running. */ running: boolean; - /** Number of jobs currently executing. */ activeJobCount: number; - /** Set of job IDs currently in-flight. */ activeJobIds: string[]; - /** Total number of ticks executed since start. */ tickCount: number; - /** Timestamp of the last tick (ISO 8601). */ lastTickAt: string | null; } - -// --------------------------------------------------------------------------- -// Scheduler -// --------------------------------------------------------------------------- - -/** - * The public interface of the job scheduler. - */ export interface PluginJobScheduler { - /** - * Start the scheduler tick loop. - * - * Safe to call multiple times — subsequent calls are no-ops. - */ - start(): void; - - /** - * Stop the scheduler tick loop. - * - * In-flight job runs are NOT cancelled — they are allowed to finish - * naturally. The tick loop simply stops firing. - */ - stop(): void; - - /** - * Register a plugin with the scheduler. - * - * Computes `nextRunAt` for all active jobs that are missing it. This is - * typically called after a plugin's worker process starts and - * `syncJobDeclarations()` has been called. - * - * @param pluginId - UUID of the plugin - */ + start(coordinator: Coordinator): Promise; + stop(): Promise; registerPlugin(pluginId: string): Promise; - - /** - * Unregister a plugin from the scheduler. - * - * Cancels any in-flight runs for the plugin and removes tracking state. - * - * @param pluginId - UUID of the plugin - */ unregisterPlugin(pluginId: string): Promise; - - /** - * Manually trigger a specific job (outside of the cron schedule). - * - * Creates a run with `trigger: "manual"` and dispatches immediately, - * respecting the overlap prevention check. - * - * @param jobId - UUID of the job to trigger - * @param trigger - What triggered this run (default: "manual") - * @returns The created run info - * @throws {Error} if the job is not found, not active, or already running - */ triggerJob(jobId: string, trigger?: "manual" | "retry"): Promise; - - /** - * Run a single scheduler tick immediately (for testing). - * - * @internal - */ - tick(): Promise; - - /** - * Get diagnostic information about the scheduler state. - */ + tick(signal?: AbortSignal): Promise; diagnostics(): SchedulerDiagnostics; } -// --------------------------------------------------------------------------- -// Implementation -// --------------------------------------------------------------------------- - -/** - * Create a new PluginJobScheduler. - * - * @example - * ```ts - * const scheduler = createPluginJobScheduler({ - * db, - * jobStore, - * workerManager, - * }); - * - * // Start the tick loop - * scheduler.start(); - * - * // When a plugin comes online, register it - * await scheduler.registerPlugin(pluginId); - * - * // Manually trigger a job - * const { runId } = await scheduler.triggerJob(jobId); - * - * // On server shutdown - * scheduler.stop(); - * ``` - */ -export function createPluginJobScheduler( - options: PluginJobSchedulerOptions, -): PluginJobScheduler { - const { - db, - jobStore, - workerManager, - tickIntervalMs = DEFAULT_TICK_INTERVAL_MS, - jobTimeoutMs = DEFAULT_JOB_TIMEOUT_MS, - maxConcurrentJobs = DEFAULT_MAX_CONCURRENT_JOBS, - } = options; +function nextCron(schedule: string, now: Date): Date | null { + return schedule && !validateCron(schedule) ? nextCronTick(parseCron(schedule), now) : null; +} +export function createPluginJobScheduler(options: PluginJobSchedulerOptions): PluginJobScheduler { + const { db, jobStore, workerManager, tickIntervalMs = 30_000, + jobTimeoutMs = 5 * 60_000, settlementTimeoutMs = 30_000, maxConcurrentJobs = 10 } = options; const log = logger.child({ service: "plugin-job-scheduler" }); - - // ----------------------------------------------------------------------- - // State - // ----------------------------------------------------------------------- - - /** Timer handle for the tick loop. */ - let tickTimer: ReturnType | null = null; - - /** Whether the scheduler is running. */ - let running = false; - - /** Set of job IDs currently being executed (for overlap prevention). */ const activeJobs = new Set(); - - /** Total number of ticks since start. */ + const pendingSettlements = new Map Promise; running?: Promise }>(); + const inFlight = new Set>(); + const subscriptions = new Map void>(); + const unregistered = new Set(); + let worker: ReturnType | undefined; + let running = false; + let next: number | null = null; let tickCount = 0; - - /** Timestamp of the last tick. */ let lastTickAt: Date | null = null; + let ticking: Promise | undefined; + const canRun = () => running && options.isBackgroundWorkEnabled?.() !== false; - /** Guard against concurrent tick execution. */ - let tickInProgress = false; - - // ----------------------------------------------------------------------- - // Core: tick - // ----------------------------------------------------------------------- - - /** - * A single scheduler tick. Queries for due jobs and dispatches them. - */ - async function tick(): Promise { - if (options.isBackgroundWorkEnabled?.() === false) return; - // Prevent overlapping ticks (in case a tick takes longer than the interval) - if (tickInProgress) { - log.debug("skipping tick — previous tick still in progress"); - return; - } - - tickInProgress = true; - tickCount++; - lastTickAt = new Date(); - - try { - const now = new Date(); - - // Query for jobs whose nextRunAt has passed and are active. - // We include jobs with null nextRunAt since they may have just been - // registered and need their first run calculated. - const dueJobs = await db - .select() - .from(pluginJobs) - .where( - and( - eq(pluginJobs.status, "active"), - lte(pluginJobs.nextRunAt, now), - ), - ); - - if (dueJobs.length === 0) { - return; - } - - log.debug({ count: dueJobs.length }, "found due jobs"); - - // Dispatch each due job (respecting concurrency limits) - const dispatches: Promise[] = []; - - for (const job of dueJobs) { - // Concurrency limit - if (activeJobs.size >= maxConcurrentJobs) { - log.warn( - { maxConcurrentJobs, activeJobCount: activeJobs.size }, - "max concurrent jobs reached, deferring remaining jobs", - ); - break; - } - - // Overlap prevention: skip if this job is already running - if (activeJobs.has(job.id)) { - log.debug( - { jobId: job.id, jobKey: job.jobKey, pluginId: job.pluginId }, - "skipping job — already running (overlap prevention)", - ); - continue; - } - - // Check if the worker is available - if (!workerManager.isRunning(job.pluginId)) { - log.debug( - { jobId: job.id, pluginId: job.pluginId }, - "skipping job — worker not running", - ); - continue; - } - - // Validate cron expression before dispatching - if (!job.schedule) { - log.warn( - { jobId: job.id, jobKey: job.jobKey }, - "skipping job — no schedule defined", - ); - continue; - } - - dispatches.push(dispatchJob(job)); - } - - if (dispatches.length > 0) { - await Promise.allSettled(dispatches); - } - } catch (err) { - log.error( - { err: err instanceof Error ? err.message : String(err) }, - "scheduler tick error", - ); - } finally { - tickInProgress = false; - } + function track(promise: Promise): Promise { + const tracked = trackIdleWork(promise); + inFlight.add(tracked); + void tracked.then(() => inFlight.delete(tracked), () => inFlight.delete(tracked)); + return tracked; } - // ----------------------------------------------------------------------- - // Core: dispatch a single job - // ----------------------------------------------------------------------- + async function settle(jobId: string) { + const entry = pendingSettlements.get(jobId); + if (!entry) return; + entry.running ??= entry.persist().finally(() => { entry.running = undefined; }); + await entry.running; + if (pendingSettlements.get(jobId) === entry) pendingSettlements.delete(jobId); + } - /** - * Dispatch a single job run — create the run record, call the worker, - * record the result, and advance the schedule pointer. - */ - async function dispatchJob( - job: typeof pluginJobs.$inferSelect, - ): Promise { - const { id: jobId, pluginId, jobKey, schedule } = job; - const jobLog = log.child({ jobId, pluginId, jobKey }); - - // Mark as active (overlap prevention) - activeJobs.add(jobId); - - let runId: string | undefined; + async function execute(job: Job, runId: string, trigger: Trigger) { const startedAt = Date.now(); - - try { - // 1. Create run record - const run = await jobStore.createRun({ - jobId, - pluginId, - trigger: "schedule", - }); - runId = run.id; - - jobLog.info({ runId }, "dispatching scheduled job"); - - // 2. Mark run as running - await jobStore.markRunning(runId); - - // 3. Call worker via RPC - await workerManager.call( - pluginId, - "runJob", - { - job: { - jobKey, - runId, - trigger: "schedule" as const, - scheduledAt: (job.nextRunAt ?? new Date()).toISOString(), - }, - }, - jobTimeoutMs, - ); - - // 4. Mark run as succeeded - const durationMs = Date.now() - startedAt; - await jobStore.completeRun(runId, { - status: "succeeded", - durationMs, - }); - - jobLog.info({ runId, durationMs }, "job completed successfully"); - } catch (err) { - const durationMs = Date.now() - startedAt; - const errorMessage = err instanceof Error ? err.message : String(err); - - jobLog.error( - { runId, durationMs, err: errorMessage }, - "job execution failed", - ); - - // Record the failure - if (runId) { - try { - await jobStore.completeRun(runId, { - status: "failed", - error: errorMessage, - durationMs, - }); - } catch (completeErr) { - jobLog.error( - { - runId, - err: completeErr instanceof Error ? completeErr.message : String(completeErr), - }, - "failed to record job failure", - ); - } - } - } finally { - // Remove from active set - activeJobs.delete(jobId); - - // 5. Always advance the schedule pointer (even on failure) - try { - await advanceSchedulePointer(job); - } catch (err) { - jobLog.error( - { err: err instanceof Error ? err.message : String(err) }, - "failed to advance schedule pointer", - ); - } - } - } - - // ----------------------------------------------------------------------- - // Core: manual trigger - // ----------------------------------------------------------------------- - - async function triggerJob( - jobId: string, - trigger: "manual" | "retry" = "manual", - ): Promise { - const job = await jobStore.getJobById(jobId); - if (!job) { - throw new Error(`Job not found: ${jobId}`); - } - - if (job.status !== "active") { - throw new Error( - `Job "${job.jobKey}" is not active (status: ${job.status})`, - ); - } - - // Overlap prevention - if (activeJobs.has(jobId)) { - throw new Error( - `Job "${job.jobKey}" is already running — cannot trigger while in progress`, - ); - } - - // Also check DB for running runs (defensive — covers multi-instance) - const existingRuns = await db - .select() - .from(pluginJobRuns) - .where( - and( - eq(pluginJobRuns.jobId, jobId), - eq(pluginJobRuns.status, "running"), - ), - ); - - if (existingRuns.length > 0) { - throw new Error( - `Job "${job.jobKey}" already has a running execution — cannot trigger while in progress`, - ); - } - - // Check worker availability - if (!workerManager.isRunning(job.pluginId)) { - throw new Error( - `Worker for plugin "${job.pluginId}" is not running — cannot trigger job`, - ); - } - - // Create the run and dispatch (non-blocking) - const run = await jobStore.createRun({ - jobId, - pluginId: job.pluginId, - trigger, - }); - - // Dispatch in background — don't block the caller - void dispatchManualRun(job, run.id, trigger); - - return { runId: run.id, jobId }; - } - - /** - * Dispatch a manually triggered job run. - */ - async function dispatchManualRun( - job: typeof pluginJobs.$inferSelect, - runId: string, - trigger: "manual" | "retry", - ): Promise { - const { id: jobId, pluginId, jobKey } = job; - const jobLog = log.child({ jobId, pluginId, jobKey, runId, trigger }); - - activeJobs.add(jobId); - const startedAt = Date.now(); - + let error: string | undefined; try { await jobStore.markRunning(runId); - - await workerManager.call( - pluginId, - "runJob", - { - job: { - jobKey, - runId, - trigger, - scheduledAt: new Date().toISOString(), - }, - }, - jobTimeoutMs, - ); - - const durationMs = Date.now() - startedAt; - await jobStore.completeRun(runId, { - status: "succeeded", - durationMs, - }); - - jobLog.info({ durationMs }, "manual job completed successfully"); + await workerManager.call(job.pluginId, "runJob", { + job: { jobKey: job.jobKey, runId, trigger, + scheduledAt: (trigger === "schedule" ? job.nextRunAt! : new Date()).toISOString() }, + }, jobTimeoutMs); } catch (err) { - const durationMs = Date.now() - startedAt; - const errorMessage = err instanceof Error ? err.message : String(err); - jobLog.error({ durationMs, err: errorMessage }, "manual job failed"); - - try { - await jobStore.completeRun(runId, { - status: "failed", - error: errorMessage, - durationMs, - }); - } catch (completeErr) { - jobLog.error( - { - err: completeErr instanceof Error ? completeErr.message : String(completeErr), - }, - "failed to record manual job failure", - ); - } - } finally { - activeJobs.delete(jobId); + error = err instanceof Error ? err.message : String(err); + log.error({ jobId: job.id, runId, err }, "plugin job failed"); } + const finishedAt = new Date(); + pendingSettlements.set(job.id, { persist: () => db.transaction(async tx => { + await tx.execute(sql`select set_config('statement_timeout', ${String(settlementTimeoutMs)}, true)`); + const store = pluginJobStore(tx); + await store.completeRun(runId, { + status: error === undefined ? "succeeded" : "failed", error, + durationMs: finishedAt.getTime() - startedAt, + }); + if (trigger === "schedule") { + const [current] = await tx.select().from(pluginJobs).where(eq(pluginJobs.id, job.id)).for("update"); + if (current && current.schedule === job.schedule && current.nextRunAt?.getTime() === job.nextRunAt?.getTime()) { + await store.updateRunTimestamps(job.id, finishedAt, nextCron(current.schedule, finishedAt)); + } + } + }) }); + try { await settle(job.id); } + finally { activeJobs.delete(job.id); worker?.wake(); } } - // ----------------------------------------------------------------------- - // Schedule pointer management - // ----------------------------------------------------------------------- + async function dispatch(job: Job, trigger: Trigger, signal?: AbortSignal): Promise { + // Reserve synchronously at the one dispatch boundary. A refresh may have + // yielded since filtering its snapshot; it cannot borrow a manual run's slot. + if (activeJobs.has(job.id) || pendingSettlements.has(job.id)) { + if (trigger !== "schedule") throw new Error(`Job "${job.jobKey}" is already running`); + return null; + } + if (trigger === "schedule" && activeJobs.size >= maxConcurrentJobs) return null; + activeJobs.add(job.id); + try { + const run = await db.transaction(async tx => { + const [current] = await tx.select().from(pluginJobs).where(eq(pluginJobs.id, job.id)).for("update"); + if (!current || current.status !== "active" || signal?.aborted || !canRun() || unregistered.has(job.pluginId) + || !workerManager.isRunning(job.pluginId)) return null; + if (trigger === "schedule" && (current.schedule !== job.schedule + || current.nextRunAt?.getTime() !== job.nextRunAt?.getTime())) return null; + const [outstanding] = await tx.select({ id: pluginJobRuns.id }).from(pluginJobRuns).where(and( + eq(pluginJobRuns.jobId, job.id), inArray(pluginJobRuns.status, ["pending", "queued", "running"]), + )).limit(1); + if (outstanding) return null; + return pluginJobStore(tx).createRun({ jobId: job.id, pluginId: job.pluginId, trigger }); + }); + if (!run) { + activeJobs.delete(job.id); + worker?.wake(); + if (trigger !== "schedule") throw new Error(`Job "${job.jobKey}" is no longer available`); + return null; + } + const completion = track(execute(job, run.id, trigger)); + void completion.catch(err => { log.error({ err, jobId: job.id }, "plugin job settlement deferred"); worker?.wake(); }); + return { runId: run.id, jobId: job.id }; + } catch (err) { activeJobs.delete(job.id); throw err; } + } - /** - * Advance the `lastRunAt` and `nextRunAt` timestamps on a job after a run. - */ - async function advanceSchedulePointer( - job: typeof pluginJobs.$inferSelect, - ): Promise { + async function refresh(signal?: AbortSignal) { + 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) && !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"]))); + const blocked = new Set(outstanding.map(run => run.jobId)); const now = new Date(); - let nextRunAt: Date | null = null; - - if (job.schedule) { - const validationError = validateCron(job.schedule); - if (validationError) { - log.warn( - { jobId: job.id, schedule: job.schedule, error: validationError }, - "invalid cron schedule — cannot compute next run", - ); - } else { - const cron = parseCron(job.schedule); - nextRunAt = nextCronTick(cron, now); - } - } - - await jobStore.updateRunTimestamps(job.id, now, nextRunAt); - } - - /** - * Ensure all active jobs for a plugin have a `nextRunAt` value. - * Called when a plugin is registered with the scheduler. - */ - async function ensureNextRunTimestamps(pluginId: string): Promise { - const jobs = await jobStore.listJobs(pluginId, "active"); - - for (const job of jobs) { - // Skip jobs that already have a valid nextRunAt in the future - if (job.nextRunAt && job.nextRunAt.getTime() > Date.now()) { - continue; - } - - // Skip jobs without a schedule - if (!job.schedule) { - continue; - } - - const validationError = validateCron(job.schedule); - if (validationError) { - log.warn( - { jobId: job.id, jobKey: job.jobKey, schedule: job.schedule, error: validationError }, - "skipping job with invalid cron schedule", - ); - continue; - } - - const cron = parseCron(job.schedule); - const nextRunAt = nextCronTick(cron, new Date()); - - if (nextRunAt) { - await jobStore.updateRunTimestamps( - job.id, - job.lastRunAt ?? new Date(0), - nextRunAt, - ); - log.debug( - { jobId: job.id, jobKey: job.jobKey, nextRunAt: nextRunAt.toISOString() }, - "computed nextRunAt for job", - ); - } - } - } - - // ----------------------------------------------------------------------- - // Plugin registration - // ----------------------------------------------------------------------- - - async function registerPlugin(pluginId: string): Promise { - log.info({ pluginId }, "registering plugin with job scheduler"); - await ensureNextRunTimestamps(pluginId); - } - - async function unregisterPlugin(pluginId: string): Promise { - log.info({ pluginId }, "unregistering plugin from job scheduler"); - - // Cancel any in-flight run records for this plugin that are still - // queued or running. Active jobs in-memory will finish naturally. - try { - const runningRuns = await db - .select() - .from(pluginJobRuns) - .where( - and( - eq(pluginJobRuns.pluginId, pluginId), - or( - eq(pluginJobRuns.status, "running"), - eq(pluginJobRuns.status, "queued"), - ), - ), - ); - - for (const run of runningRuns) { - await jobStore.completeRun(run.id, { - status: "cancelled", - error: "Plugin unregistered", - durationMs: run.startedAt - ? Date.now() - run.startedAt.getTime() - : null, + const dispatches: Promise[] = []; + next = null; + for (const job of runnable) { + if (!canRun() || signal?.aborted) break; + if (blocked.has(job.id) || activeJobs.has(job.id) || pendingSettlements.has(job.id)) continue; + if (!job.nextRunAt) { + const deadline = nextCron(job.schedule, now); + if (!deadline) continue; + await db.transaction(async tx => { + const [current] = await tx.select().from(pluginJobs).where(eq(pluginJobs.id, job.id)).for("update"); + if (current && current.nextRunAt === null && current.schedule === job.schedule && current.status === "active") { + await pluginJobStore(tx).updateRunTimestamps(job.id, current.lastRunAt ?? new Date(0), deadline); + } }); + job.nextRunAt = deadline; + } + const at = job.nextRunAt.getTime(); + if (at <= now.getTime() && activeJobs.size < maxConcurrentJobs) { + dispatches.push(dispatch(job, "schedule", signal)); + } else if (activeJobs.size < maxConcurrentJobs) { + next = next === null ? at : Math.min(next, at); } - } catch (err) { - log.error( - { - pluginId, - err: err instanceof Error ? err.message : String(err), - }, - "error cancelling in-flight runs during unregister", - ); - } - - // Remove any active tracking for jobs owned by this plugin - const jobs = await jobStore.listJobs(pluginId); - for (const job of jobs) { - activeJobs.delete(job.id); } + const results = await Promise.allSettled(dispatches); + const failed = [...settlements, ...results].find(result => result.status === "rejected"); + if (failed?.status === "rejected") throw failed.reason; } - // ----------------------------------------------------------------------- - // Lifecycle: start / stop - // ----------------------------------------------------------------------- - - function start(): void { - if (running) { - log.debug("scheduler already running"); - return; - } - - running = true; - tickTimer = setInterval(() => { - void tick(); - }, tickIntervalMs); - - log.info( - { tickIntervalMs, maxConcurrentJobs }, - "plugin job scheduler started", - ); + function tick(signal?: AbortSignal): Promise { + if (!canRun()) return Promise.resolve(); + if (ticking) return ticking; + tickCount++; lastTickAt = new Date(); + ticking = refresh(signal).finally(() => { ticking = undefined; }); + return ticking; } - function stop(): void { - // Always clear the timer defensively, even if `running` is already false, - // to prevent leaked interval timers. - if (tickTimer !== null) { - clearInterval(tickTimer); - tickTimer = null; - } - - if (!running) return; - running = false; - - log.info( - { activeJobCount: activeJobs.size }, - "plugin job scheduler stopped", - ); - } - - // ----------------------------------------------------------------------- - // Diagnostics - // ----------------------------------------------------------------------- - - function diagnostics(): SchedulerDiagnostics { - return { - running, - activeJobCount: activeJobs.size, - activeJobIds: [...activeJobs], - tickCount, - lastTickAt: lastTickAt?.toISOString() ?? null, - }; - } - - // ----------------------------------------------------------------------- - // Public API - // ----------------------------------------------------------------------- - return { - start, - stop, - registerPlugin, - unregisterPlugin, - triggerJob, + async start(coordinator) { + if (worker) { await worker.ready; return; } + running = true; + worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.pluginSchedules, { + run: tick, next: async () => next, canRun, retryMs: tickIntervalMs, + notificationCoalesceMs: tickIntervalMs, + }); + await worker.ready; + }, + async stop() { + running = false; + for (const unsubscribe of subscriptions.values()) unsubscribe(); + 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); + subscriptions.get(pluginId)?.(); + const handle = workerManager.getWorker(pluginId); + if (handle) { + const wake = () => worker?.wake(); + handle.on("status", wake); + subscriptions.set(pluginId, () => handle.off("status", wake)); + } + worker?.wake(); + }, + async unregisterPlugin(pluginId) { + unregistered.add(pluginId); + subscriptions.get(pluginId)?.(); subscriptions.delete(pluginId); + worker?.wake(); + const runs = await db.select().from(pluginJobRuns).where(and( + eq(pluginJobRuns.pluginId, pluginId), inArray(pluginJobRuns.status, ["pending", "queued", "running"]), + )); + for (const run of runs) { + if (activeJobs.has(run.jobId) || pendingSettlements.has(run.jobId)) continue; + await db.transaction(async tx => { + const [current] = await tx.select().from(pluginJobRuns).where(eq(pluginJobRuns.id, run.id)).for("update"); + // A completion may have committed after the discovery read. Never + // overwrite its result, or the result still retained for persistence. + if (!current || !["running", "queued", "pending"].includes(current.status) + || activeJobs.has(current.jobId) || pendingSettlements.has(current.jobId)) return; + await pluginJobStore(tx).completeRun(current.id, { status: "cancelled", error: "Plugin unregistered", + durationMs: current.startedAt ? Date.now() - current.startedAt.getTime() : null }); + }); + } + }, + triggerJob(jobId, trigger = "manual") { + return track((async () => { + if (!canRun()) throw new Error("Plugin job scheduler is not accepting work"); + const job = await jobStore.getJobById(jobId); + if (!job) throw new Error(`Job not found: ${jobId}`); + if (job.status !== "active") throw new Error(`Job "${job.jobKey}" is not active (status: ${job.status})`); + if (activeJobs.has(jobId) || pendingSettlements.has(jobId)) throw new Error(`Job "${job.jobKey}" is already running`); + const runs = await db.select({ id: pluginJobRuns.id }).from(pluginJobRuns).where(and( + eq(pluginJobRuns.jobId, jobId), inArray(pluginJobRuns.status, ["pending", "queued", "running"]), + )).limit(1); + if (runs.length) throw new Error(`Job "${job.jobKey}" already has a running execution`); + if (unregistered.has(job.pluginId) || !workerManager.isRunning(job.pluginId)) throw new Error(`Worker for plugin "${job.pluginId}" is not running`); + const result = await dispatch(job, trigger); + if (!result) throw new Error(`Job "${job.jobKey}" is no longer available`); + return result; + })()); + }, tick, - diagnostics, + diagnostics: () => ({ running, activeJobCount: activeJobs.size, activeJobIds: [...activeJobs], tickCount, + lastTickAt: lastTickAt?.toISOString() ?? null }), }; } diff --git a/server/src/services/plugin-job-store.ts b/server/src/services/plugin-job-store.ts index bd590a1f882..b70c22e348d 100644 --- a/server/src/services/plugin-job-store.ts +++ b/server/src/services/plugin-job-store.ts @@ -1,3 +1,4 @@ +import { writePluginJobChange, type PluginJobDatabase } from "./plugin-job-work-notifications.js"; /** * Plugin Job Store — persistence layer for scheduled plugin jobs and their * execution history. @@ -100,12 +101,12 @@ export interface CompleteJobRunInput { * }); * ``` */ -export function pluginJobStore(db: Db) { +export function pluginJobStore(db: PluginJobDatabase) { // ----------------------------------------------------------------------- // Internal helpers // ----------------------------------------------------------------------- - async function assertPluginExists(pluginId: string): Promise { + async function assertPluginExists(db: PluginJobDatabase, pluginId: string): Promise { const rows = await db .select({ id: plugins.id }) .from(plugins) @@ -145,63 +146,66 @@ export function pluginJobStore(db: Db) { pluginId: string, declarations: PluginJobDeclaration[], ): Promise { - await assertPluginExists(pluginId); + return writePluginJobChange(db, async db => { + await assertPluginExists(db, pluginId); - // Fetch existing jobs for this plugin - const existingJobs = await db - .select() - .from(pluginJobs) - .where(eq(pluginJobs.pluginId, pluginId)); + // Fetch existing jobs for this plugin + const existingJobs = await db + .select() + .from(pluginJobs) + .where(eq(pluginJobs.pluginId, pluginId)); - const existingByKey = new Map( - existingJobs.map((j) => [j.jobKey, j]), - ); + const existingByKey = new Map( + existingJobs.map((j) => [j.jobKey, j]), + ); - const declaredKeys = new Set(); + const declaredKeys = new Set(); - // Upsert each declared job - for (const decl of declarations) { - declaredKeys.add(decl.jobKey); + // Upsert each declared job + for (const decl of declarations) { + declaredKeys.add(decl.jobKey); - const existing = existingByKey.get(decl.jobKey); - const schedule = decl.schedule ?? ""; + const existing = existingByKey.get(decl.jobKey); + const schedule = decl.schedule ?? ""; - if (existing) { - // Update schedule if it changed; re-activate if it was paused - const updates: Record = { - updatedAt: new Date(), - }; - if (existing.schedule !== schedule) { - updates.schedule = schedule; + if (existing) { + // Update schedule if it changed; re-activate if it was paused + const updates: Record = { + updatedAt: new Date(), + }; + if (existing.schedule !== schedule) { + updates.schedule = schedule; + updates.nextRunAt = null; + } + if (existing.status === "paused") { + updates.status = "active"; + } + + await db + .update(pluginJobs) + .set(updates) + .where(eq(pluginJobs.id, existing.id)); + } else { + // Insert new job + await db.insert(pluginJobs).values({ + pluginId, + jobKey: decl.jobKey, + schedule, + status: "active", + }); } - if (existing.status === "paused") { - updates.status = "active"; + } + + // Pause jobs that are no longer declared in the manifest + for (const existing of existingJobs) { + if (!declaredKeys.has(existing.jobKey) && existing.status !== "paused") { + await db + .update(pluginJobs) + .set({ status: "paused", updatedAt: new Date() }) + .where(eq(pluginJobs.id, existing.id)); } - - await db - .update(pluginJobs) - .set(updates) - .where(eq(pluginJobs.id, existing.id)); - } else { - // Insert new job - await db.insert(pluginJobs).values({ - pluginId, - jobKey: decl.jobKey, - schedule, - status: "active", - }); } - } - - // Pause jobs that are no longer declared in the manifest - for (const existing of existingJobs) { - if (!declaredKeys.has(existing.jobKey) && existing.status !== "paused") { - await db - .update(pluginJobs) - .set({ status: "paused", updatedAt: new Date() }) - .where(eq(pluginJobs.id, existing.id)); - } - } + }); }, /** @@ -290,10 +294,12 @@ export function pluginJobStore(db: Db) { jobId: string, status: JobDefinitionStatus, ): Promise { - await db - .update(pluginJobs) - .set({ status, updatedAt: new Date() }) - .where(eq(pluginJobs.id, jobId)); + return writePluginJobChange(db, async db => { + await db + .update(pluginJobs) + .set({ status, updatedAt: new Date() }) + .where(eq(pluginJobs.id, jobId)); + }); }, /** @@ -311,14 +317,16 @@ export function pluginJobStore(db: Db) { lastRunAt: Date, nextRunAt: Date | null, ): Promise { - await db - .update(pluginJobs) - .set({ - lastRunAt, - nextRunAt, - updatedAt: new Date(), - }) - .where(eq(pluginJobs.id, jobId)); + return writePluginJobChange(db, async db => { + await db + .update(pluginJobs) + .set({ + lastRunAt, + nextRunAt, + updatedAt: new Date(), + }) + .where(eq(pluginJobs.id, jobId)); + }); }, /** @@ -329,9 +337,11 @@ export function pluginJobStore(db: Db) { * @param pluginId - UUID of the owning plugin */ async deleteAllJobs(pluginId: string): Promise { - await db - .delete(pluginJobs) - .where(eq(pluginJobs.pluginId, pluginId)); + return writePluginJobChange(db, async db => { + await db + .delete(pluginJobs) + .where(eq(pluginJobs.pluginId, pluginId)); + }); }, // ===================================================================== @@ -351,17 +361,19 @@ export function pluginJobStore(db: Db) { async createRun( input: CreateJobRunInput, ): Promise { - const rows = await db - .insert(pluginJobRuns) - .values({ - jobId: input.jobId, - pluginId: input.pluginId, - trigger: input.trigger, - status: "queued", - }) - .returning(); + return writePluginJobChange(db, async db => { + const rows = await db + .insert(pluginJobRuns) + .values({ + jobId: input.jobId, + pluginId: input.pluginId, + trigger: input.trigger, + status: "queued", + }) + .returning(); - return rows[0]!; + return rows[0]!; + }); }, /** @@ -370,13 +382,15 @@ export function pluginJobStore(db: Db) { * @param runId - UUID of the run row */ async markRunning(runId: string): Promise { - await db - .update(pluginJobRuns) - .set({ - status: "running" as PluginJobRunStatus, - startedAt: new Date(), - }) - .where(eq(pluginJobRuns.id, runId)); + return writePluginJobChange(db, async db => { + await db + .update(pluginJobRuns) + .set({ + status: "running" as PluginJobRunStatus, + startedAt: new Date(), + }) + .where(eq(pluginJobRuns.id, runId)); + }); }, /** @@ -390,15 +404,17 @@ export function pluginJobStore(db: Db) { runId: string, input: CompleteJobRunInput, ): Promise { - await db - .update(pluginJobRuns) - .set({ - status: input.status, - error: input.error ?? null, - durationMs: input.durationMs ?? null, - finishedAt: new Date(), - }) - .where(eq(pluginJobRuns.id, runId)); + return writePluginJobChange(db, async db => { + await db + .update(pluginJobRuns) + .set({ + status: input.status, + error: input.error ?? null, + durationMs: input.durationMs ?? null, + finishedAt: new Date(), + }) + .where(eq(pluginJobRuns.id, runId)); + }); }, /** diff --git a/server/src/services/plugin-job-work-notifications.ts b/server/src/services/plugin-job-work-notifications.ts new file mode 100644 index 00000000000..3ce7ab287c0 --- /dev/null +++ b/server/src/services/plugin-job-work-notifications.ts @@ -0,0 +1,11 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type PluginJobDatabase = Pick; + +export function writePluginJobChange(db: PluginJobDatabase, write: (tx: PluginJobDatabase) => Promise): Promise { + return db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.pluginSchedules); + return write(tx); + }); +} diff --git a/server/src/services/plugin-loader.ts b/server/src/services/plugin-loader.ts index c07e48ac2e3..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 - jobScheduler.stop(); - - // 2. Stop all worker processes - await workerManager.stopAll(); + await finishServiceShutdown(() => jobScheduler.stop(), () => workerManager.stopAll()); log.info("plugin-loader: all plugins shut down"); }, @@ -2416,9 +2413,9 @@ export function pluginLoader( // 6. Sync job declarations and register with scheduler // ------------------------------------------------------------------ const jobDeclarations = manifest.jobs ?? []; + await jobStore.syncJobDeclarations(pluginId, jobDeclarations); + await jobScheduler.registerPlugin(pluginId); if (jobDeclarations.length > 0) { - await jobStore.syncJobDeclarations(pluginId, jobDeclarations); - await jobScheduler.registerPlugin(pluginId); registered.jobs = jobDeclarations.length; log.info( diff --git a/server/src/services/plugin-registry.ts b/server/src/services/plugin-registry.ts index c47cf4f6f5d..f7574bb293d 100644 --- a/server/src/services/plugin-registry.ts +++ b/server/src/services/plugin-registry.ts @@ -1,3 +1,4 @@ +import { writePluginJobChange } from "./plugin-job-work-notifications.js"; import { asc, eq, isNull, ne, sql, and } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -267,25 +268,27 @@ export function pluginRegistryService(db: Db) { const plugin = await getById(id); if (!plugin) throw notFound("Plugin not found"); - if (removeData) { - // Hard delete – plugin_config cascades via FK onDelete + return writePluginJobChange(db, async db => { + if (removeData) { + // Hard delete – plugin_config cascades via FK onDelete + return db + .delete(plugins) + .where(eq(plugins.id, id)) + .returning() + .then((rows) => rows[0] ?? null); + } + + // Soft delete – mark as uninstalled return db - .delete(plugins) + .update(plugins) + .set({ + status: "uninstalled" as PluginStatus, + updatedAt: new Date(), + }) .where(eq(plugins.id, id)) .returning() .then((rows) => rows[0] ?? null); - } - - // Soft delete – mark as uninstalled - return db - .update(plugins) - .set({ - status: "uninstalled" as PluginStatus, - updatedAt: new Date(), - }) - .where(eq(plugins.id, id)) - .returning() - .then((rows) => rows[0] ?? null); + }); }, // ----- Config --------------------------------------------------------- @@ -670,35 +673,39 @@ export function pluginRegistryService(db: Db) { jobKey: string, input: { schedule: string; status?: PluginJobStatus }, ) => { - const existing = await db - .select() - .from(pluginJobs) - .where(and(eq(pluginJobs.pluginId, pluginId), eq(pluginJobs.jobKey, jobKey))) - .then((rows) => rows[0] ?? null); + return writePluginJobChange(db, async db => { + const existing = await db + .select() + .from(pluginJobs) + .where(and(eq(pluginJobs.pluginId, pluginId), eq(pluginJobs.jobKey, jobKey))) + .for("update") + .then((rows) => rows[0] ?? null); + + if (existing) { + return db + .update(pluginJobs) + .set({ + schedule: input.schedule, + nextRunAt: existing.schedule === input.schedule ? existing.nextRunAt : null, + status: input.status ?? existing.status, + updatedAt: new Date(), + }) + .where(eq(pluginJobs.id, existing.id)) + .returning() + .then((rows) => rows[0]); + } - if (existing) { return db - .update(pluginJobs) - .set({ + .insert(pluginJobs) + .values({ + pluginId, + jobKey, schedule: input.schedule, - status: input.status ?? existing.status, - updatedAt: new Date(), + status: input.status ?? "active", }) - .where(eq(pluginJobs.id, existing.id)) .returning() .then((rows) => rows[0]); - } - - return db - .insert(pluginJobs) - .values({ - pluginId, - jobKey, - schedule: input.schedule, - status: input.status ?? "active", - }) - .returning() - .then((rows) => rows[0]); + }); }, /** @@ -721,17 +728,19 @@ export function pluginRegistryService(db: Db) { trigger: PluginJobRunTrigger, companyId: string | null, ) => { - return db - .insert(pluginJobRuns) - .values({ - pluginId, - jobId, - companyId, - trigger, - status: "pending", - }) - .returning() - .then((rows) => rows[0]); + return writePluginJobChange(db, async db => { + return db + .insert(pluginJobRuns) + .values({ + pluginId, + jobId, + companyId, + trigger, + status: "pending", + }) + .returning() + .then((rows) => rows[0]); + }); }, /** @@ -752,12 +761,14 @@ export function pluginRegistryService(db: Db) { finishedAt?: Date; }, ) => { - return db - .update(pluginJobRuns) - .set(input) - .where(eq(pluginJobRuns.id, runId)) - .returning() - .then((rows) => rows[0] ?? null); + return writePluginJobChange(db, async db => { + return db + .update(pluginJobRuns) + .set(input) + .where(eq(pluginJobRuns.id, runId)) + .returning() + .then((rows) => rows[0] ?? null); + }); }, // ----- Webhooks ------------------------------------------------------- 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"); +}