mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 20:21:50 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
39e811c9c5
commit
96727b3cfe
17 files changed
+1313
-863
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -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: {
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>, pluginId: string;
|
||||
let scheduler: ReturnType<typeof createPluginJobScheduler> | undefined;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let available = true, draining = false;
|
||||
let events: EventEmitter;
|
||||
const rpc = vi.fn<(...args: unknown[]) => Promise<unknown>>();
|
||||
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<void>(resolve => { release = resolve; });
|
||||
const acquired = new Promise<void>(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<void>(resolve => { resumeRead = resolve; });
|
||||
const readEntered = new Promise<void>(resolve => { enteredRead = resolve; });
|
||||
const callGate = new Promise<void>(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<void>(resolve => { release = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { locked = resolve; });
|
||||
const advancing = db.transaction(async tx => {
|
||||
await pluginJobStore(tx).updateRunTimestamps(row.id, new Date(), advanced);
|
||||
locked(); await barrier;
|
||||
});
|
||||
await acquired;
|
||||
const updating = pluginRegistryService(db).upsertJob(pluginId, row.jobKey, { schedule: row.schedule });
|
||||
try {
|
||||
await vi.waitFor(async () => {
|
||||
const [waiting] = await db.execute<{ count: number }>(sql`select count(*)::int as count from pg_stat_activity
|
||||
where datname = current_database() and state = 'active' and wait_event_type = 'Lock'`);
|
||||
expect(waiting.count).toBeGreaterThan(0);
|
||||
});
|
||||
} finally { release(); await advancing; await updating; }
|
||||
expect((await pluginJobStore(db).getJobById(row.id))!.nextRunAt).toEqual(advanced);
|
||||
});
|
||||
it("records provider failure and advances the cron obligation", async () => {
|
||||
rpc.mockRejectedValueOnce(new Error("plugin rejected job")); await due(); await start(); await completed();
|
||||
expect((await runs())[0]).toMatchObject({ status: "failed", error: "plugin rejected job" });
|
||||
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<void>(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<void>(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<void>;
|
||||
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<void>(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<void>(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<void>(resolve => { unlock = resolve; });
|
||||
const admitted = new Promise<void>(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<void> | 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();
|
||||
});
|
||||
});
|
||||
+4
-6
@@ -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<void> | null = null;
|
||||
const shutdownAppServices = (): Promise<void> => {
|
||||
if (appServicesShutdown) return appServicesShutdown;
|
||||
appServicesShutdown = (async () => {
|
||||
// The scheduler tick queries the database. Stop it here, inside the
|
||||
// awaited teardown, so no tick runs after the caller ends the pool.
|
||||
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;
|
||||
|
||||
@@ -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(); } };
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>;
|
||||
|
||||
/** 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<void>;
|
||||
stop(): Promise<void>;
|
||||
registerPlugin(pluginId: string): Promise<void>;
|
||||
|
||||
/**
|
||||
* 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<void>;
|
||||
|
||||
/**
|
||||
* 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<TriggerJobResult>;
|
||||
|
||||
/**
|
||||
* Run a single scheduler tick immediately (for testing).
|
||||
*
|
||||
* @internal
|
||||
*/
|
||||
tick(): Promise<void>;
|
||||
|
||||
/**
|
||||
* Get diagnostic information about the scheduler state.
|
||||
*/
|
||||
tick(signal?: AbortSignal): Promise<void>;
|
||||
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<typeof setInterval> | null = null;
|
||||
|
||||
/** Whether the scheduler is running. */
|
||||
let running = false;
|
||||
|
||||
/** Set of job IDs currently being executed (for overlap prevention). */
|
||||
const activeJobs = new Set<string>();
|
||||
|
||||
/** Total number of ticks since start. */
|
||||
const pendingSettlements = new Map<string, { persist: () => Promise<void>; running?: Promise<void> }>();
|
||||
const inFlight = new Set<Promise<unknown>>();
|
||||
const subscriptions = new Map<string, () => void>();
|
||||
const unregistered = new Set<string>();
|
||||
let worker: ReturnType<typeof registerDeadlineWork> | 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<void> | 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<void> {
|
||||
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<void>[] = [];
|
||||
|
||||
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<T>(promise: Promise<T>): Promise<T> {
|
||||
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<void> {
|
||||
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<TriggerJobResult> {
|
||||
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<void> {
|
||||
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<TriggerJobResult | null> {
|
||||
// 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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
log.info({ pluginId }, "registering plugin with job scheduler");
|
||||
await ensureNextRunTimestamps(pluginId);
|
||||
}
|
||||
|
||||
async function unregisterPlugin(pluginId: string): Promise<void> {
|
||||
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<unknown>[] = [];
|
||||
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<void> {
|
||||
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 }),
|
||||
};
|
||||
}
|
||||
@@ -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<void> {
|
||||
async function assertPluginExists(db: PluginJobDatabase, pluginId: string): Promise<void> {
|
||||
const rows = await db
|
||||
.select({ id: plugins.id })
|
||||
.from(plugins)
|
||||
@@ -145,63 +146,66 @@ export function pluginJobStore(db: Db) {
|
||||
pluginId: string,
|
||||
declarations: PluginJobDeclaration[],
|
||||
): Promise<void> {
|
||||
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<string>();
|
||||
const declaredKeys = new Set<string>();
|
||||
|
||||
// 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<string, unknown> = {
|
||||
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<string, unknown> = {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<typeof pluginJobRuns.$inferSelect> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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));
|
||||
});
|
||||
},
|
||||
|
||||
/**
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export type PluginJobDatabase = Pick<Db, "transaction" | "select" | "update" | "insert" | "delete">;
|
||||
|
||||
export function writePluginJobChange<T>(db: PluginJobDatabase, write: (tx: PluginJobDatabase) => Promise<T>): Promise<T> {
|
||||
return db.transaction(async tx => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.pluginSchedules);
|
||||
return write(tx);
|
||||
});
|
||||
}
|
||||
@@ -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(
|
||||
|
||||
@@ -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 -------------------------------------------------------
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -272,3 +272,18 @@ export async function coordinateHeartbeatSchedulerShutdown<
|
||||
waitedForSchedulerIdle: true,
|
||||
};
|
||||
}
|
||||
|
||||
export async function finishServiceShutdown(
|
||||
stop: () => Promise<void>,
|
||||
cleanup: () => Promise<void>,
|
||||
): Promise<void> {
|
||||
const failures: unknown[] = [];
|
||||
let retryStop = false;
|
||||
try { await stop(); } catch { retryStop = true; }
|
||||
try { await cleanup(); } catch (error) { failures.push(error); }
|
||||
if (retryStop) {
|
||||
try { await stop(); } catch (error) { failures.unshift(error); }
|
||||
}
|
||||
if (failures.length === 1) throw failures[0];
|
||||
if (failures.length > 1) throw new AggregateError(failures, "Service shutdown failed");
|
||||
}
|
||||
Reference in new issue
Block a user