diff --git a/doc/idle-plugin-jobs.md b/doc/idle-plugin-jobs.md new file mode 100644 index 00000000000..e1cc4c1f2ad --- /dev/null +++ b/doc/idle-plugin-jobs.md @@ -0,0 +1,75 @@ +# 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. +- 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 wakes waiting work without an interval. 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. +- Stop rejects new admissions, detaches worker subscriptions, and waits for + admitted work. Existing calls retain their RPC timeout. The application then + stops the shared coordinator before ending the database pool. + +## 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. diff --git a/doc/plans/2026-10-24-plugin-job-sleep-audit.md b/doc/plans/2026-10-24-plugin-job-sleep-audit.md index b09edf2faf1..5c745ecf9c0 100644 --- a/doc/plans/2026-10-24-plugin-job-sleep-audit.md +++ b/doc/plans/2026-10-24-plugin-job-sleep-audit.md @@ -70,3 +70,21 @@ 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. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index ca388a5f3b9..11bd37b363d 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -931,3 +931,24 @@ after its single justified retry. The duplicate cancelled workflows for #17490 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. diff --git a/server/src/__tests__/cloud-warm-standby.test.ts b/server/src/__tests__/cloud-warm-standby.test.ts index e5d8bbfdcd5..4a73563341b 100644 --- a/server/src/__tests__/cloud-warm-standby.test.ts +++ b/server/src/__tests__/cloud-warm-standby.test.ts @@ -140,16 +140,18 @@ describe("unclaimed Cloud background work", () => { tickIntervalMs: 1000, isBackgroundWorkEnabled: () => enabled, }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, onError: () => {} }); try { email.start(); - scheduler.start(); + await scheduler.start(coordinator); await vi.advanceTimersByTimeAsync(10 * 60_000); expect(select).not.toHaveBeenCalled(); enabled = true; await vi.advanceTimersByTimeAsync(1000); expect(select.mock.calls.length).toBeGreaterThanOrEqual(2); } finally { - scheduler.stop(); + await scheduler.stop(); + await coordinator.stop(); await email.shutdown(); } }); diff --git a/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts b/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts index 766a9e4fc54..533df782fc0 100644 --- a/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts +++ b/server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts @@ -439,7 +439,7 @@ describe("listReadyPluginEnvironmentDrivers worker recovery", () => { clearPlugin: vi.fn(), subscriptionCount: vi.fn(() => 0), }; - const jobScheduler = { unregisterPlugin: vi.fn().mockResolvedValue(undefined) }; + const jobScheduler = { registerPlugin: vi.fn().mockResolvedValue(undefined), unregisterPlugin: vi.fn().mockResolvedValue(undefined) }; const toolDispatcher = { unregisterPluginTools: vi.fn(), registerPluginTools: vi.fn() }; const loader = pluginLoader( @@ -454,7 +454,7 @@ describe("listReadyPluginEnvironmentDrivers worker recovery", () => { workerManager, eventBus, jobScheduler, - jobStore: {}, + jobStore: { syncJobDeclarations: vi.fn().mockResolvedValue(undefined) }, toolDispatcher, buildHostHandlers: vi.fn(() => ({})), instanceInfo: { diff --git a/server/src/__tests__/plugin-scheduled-work.test.ts b/server/src/__tests__/plugin-scheduled-work.test.ts index 1bd35c8f04c..b50937a2c7a 100644 --- a/server/src/__tests__/plugin-scheduled-work.test.ts +++ b/server/src/__tests__/plugin-scheduled-work.test.ts @@ -1,33 +1,213 @@ -import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { sql } from "drizzle-orm"; -import { createDb } from "@paperclipai/db"; +import { EventEmitter } from "node:events"; +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { createDb, plugins, pluginJobs, pluginJobRuns } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { createPluginJobScheduler } from "../services/plugin-job-scheduler.js"; import { pluginJobStore } from "../services/plugin-job-store.js"; +import { pluginRegistryService } from "../services/plugin-registry.js"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PLUGIN_SCHEDULE_TEST_DATABASE_URL"); const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("plugin job scheduling without idle polling", () => { let temporary: Awaited> | undefined; - let db: ReturnType; + let db: ReturnType, pluginId: string; let scheduler: ReturnType | undefined; + let coordinator: ReturnType | undefined; + let available = true, draining = false; + let events: EventEmitter; + const rpc = vi.fn<(...args: unknown[]) => Promise>(); + const errors: unknown[] = []; beforeAll(async () => { if (!external) temporary = await startEmbeddedPostgresTestDatabase("plugin-schedule-"); - db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); - await db.execute(sql`select 1`); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }, 30_000); - afterEach(() => { scheduler?.stop(); vi.useRealTimers(); vi.restoreAllMocks(); }); - afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); - - it("does not query repeatedly after an empty startup reconciliation", async () => { - const workerManager = { isRunning: () => false, call: vi.fn() } as unknown as PluginWorkerManager; - scheduler = createPluginJobScheduler({ db, jobStore: pluginJobStore(db), workerManager }); + 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 }); - scheduler.start(); await scheduler.tick(); - const reads = vi.spyOn(db, "select"); + }); + 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) { + const workerManager = { isRunning: () => available, call: rpc, getWorker: () => events } as unknown as PluginWorkerManager; + scheduler = createPluginJobScheduler({ db, jobStore: pluginJobStore(db), workerManager, + isBackgroundWorkEnabled: () => !draining, maxConcurrentJobs }); + 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 settled(); expect(coordinator!.nextWakeAt()).toBeNull(); + 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(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("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(() => expect(errors.length).toBeGreaterThan(0)); 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("records provider failure and advances the cron obligation", async () => { + rpc.mockRejectedValueOnce(new Error("plugin rejected job")); await due(); await start(); await completed(); + expect((await runs())[0]).toMatchObject({ status: "failed", error: "plugin rejected job" }); + await settled(); expect(coordinator!.nextWakeAt()! > Date.now()).toBe(true); + }); + it("does not replay an unresolved run left by another process", async () => { + const row = await due(); await pluginJobStore(db).createRun({ jobId: row.id, pluginId, trigger: "schedule" }); + await start(); await quiet(); expect(rpc).not.toHaveBeenCalled(); + const [run] = await runs(); await pluginJobStore(db).completeRun(run.id, { status: "cancelled" }); await completed(); + }); + it("fills released capacity from completion without a poll", async () => { + await due("one"); await due("two"); let release!: () => void; + rpc.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + const starting = start(1); await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(1)); + release(); await starting; await completed(2); expect(rpc).toHaveBeenCalledTimes(2); + }); + it("tracks manual work and waits for admitted calls during shutdown", async () => { + const row = await job("manual", ""); await start(); await quiet(); let release!: () => void; + rpc.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + await scheduler!.triggerJob(row.id); await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(1)); + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + let stopped = false; const stopping = scheduler!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false); + release(); await stopping; expect((await runs())[0].status).toBe("succeeded"); + }); + it("does no SQL while waiting for a future configured deadline", async () => { + const row = await job(); const at = new Date(Date.now() + 600_000); + await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), at); + await start(); await settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + 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 settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + await vi.advanceTimersByTimeAsync(at.getTime() - Date.now()); await completed(); expect(rpc).toHaveBeenCalledTimes(1); + }); + it("rechecks a paused definition after waiting for its lock", async () => { + const row = await due(); let startup!: Promise; + await db.transaction(async tx => { + await pluginJobStore(tx).updateJobStatus(row.id, "paused"); + startup = start(); await vi.advanceTimersByTimeAsync(1000); expect(rpc).not.toHaveBeenCalled(); + }); + await startup; await quiet(); expect(rpc).not.toHaveBeenCalled(); + }); + it("prevents overlapping concurrent manual triggers", async () => { + const row = await job("manual", ""); await start(); await quiet(); let release!: () => void; + rpc.mockImplementation(() => new Promise(resolve => { release = resolve; })); + const attempts = await Promise.allSettled([scheduler!.triggerJob(row.id), scheduler!.triggerJob(row.id)]); + await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(1)); + release(); await completed(); expect(attempts.filter(result => result.status === "fulfilled")).toHaveLength(1); + }); + it("deleting the last job cancels its future deadline", async () => { + await job(); await start(); await settled(); await pluginJobStore(db).deleteAllJobs(pluginId); await quiet(); }); }); diff --git a/server/src/app.ts b/server/src/app.ts index 76b376cde9a..d3dfddd0a9f 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1267,7 +1267,6 @@ export async function createApp( app.use(errorHandler); jobCoordinator.start(); - scheduler.start(); const deliveryWork = createDeliveryWorkCoordinator({ owner: db, canRun: () => !isWarmStandby() && !isIdleTaskDrainActive(), @@ -1275,6 +1274,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(), @@ -1458,7 +1458,7 @@ export async function createApp( 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(); + await scheduler.stop(); await agentLifecycle.stop(); await publicMcpEvents?.stop(); await dotMcpEvents?.stop(); diff --git a/server/src/services/deadline-work.ts b/server/src/services/deadline-work.ts index bb3c74a4f0b..c2558d75e02 100644 --- a/server/src/services/deadline-work.ts +++ b/server/src/services/deadline-work.ts @@ -14,7 +14,7 @@ export function registerDeadlineWork( let requested = 1, reconciled = 0; let next: number | null = null; let earliest = 0; - return coordinator.register(queue, { + const worker = coordinator.register(queue, { retryMs: task.retryMs, pausePassiveMaintenance: true, onSettled: () => { requested++; }, @@ -32,4 +32,5 @@ export function registerDeadlineWork( hasPending: async () => requested !== reconciled, nextRunAt: () => next, }); + return { ...worker, wake: () => { requested++; worker.wake(); } }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 641812c6213..6c03e6db2d4 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -59,6 +59,7 @@ export const DELIVERY_QUEUES = { gitRestorePoints: "git-restore-points", nativeWorkspaceSyncRetention: "native-workspace-sync-retention", question: "question-responses", + pluginSchedules: "plugin-schedules", pluginLifecycle: "plugin-resource-lifecycle", toolAction: "tool-action-receipts", terminalWorkspaces: "terminal-workspaces", diff --git a/server/src/services/plugin-job-coordinator.ts b/server/src/services/plugin-job-coordinator.ts index bb0df0836e7..47ef9fab690 100644 --- a/server/src/services/plugin-job-coordinator.ts +++ b/server/src/services/plugin-job-coordinator.ts @@ -130,8 +130,8 @@ export function createPluginJobCoordinator( { pluginId, pluginKey, jobCount: jobDeclarations.length }, "syncing job declarations from manifest", ); - await jobStore.syncJobDeclarations(pluginId, jobDeclarations); } + await jobStore.syncJobDeclarations(pluginId, jobDeclarations); // Register with the scheduler (computes nextRunAt for active jobs) await scheduler.registerPlugin(pluginId); diff --git a/server/src/services/plugin-job-scheduler.ts b/server/src/services/plugin-job-scheduler.ts index a7044f2e8dd..b630aa7e6a3 100644 --- a/server/src/services/plugin-job-scheduler.ts +++ b/server/src/services/plugin-job-scheduler.ts @@ -1,755 +1,248 @@ -/** - * 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 } from "drizzle-orm"; +import { pluginJobs, pluginJobRuns, type Db } from "@paperclipai/db"; +import { pluginJobStore, type PluginJobStore } from "./plugin-job-store.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { parseCron, nextCronTick, validateCron } from "./cron.js"; +import { trackIdleWork } from "./task-admission.js"; import { logger } from "../middleware/logger.js"; -// --------------------------------------------------------------------------- -// Constants -// --------------------------------------------------------------------------- +type Job = typeof pluginJobs.$inferSelect; +type Trigger = "schedule" | "manual" | "retry"; +type Coordinator = ReturnType; -/** Default interval between scheduler ticks (30 seconds). */ -const DEFAULT_TICK_INTERVAL_MS = 30_000; - -/** Default timeout for a runJob RPC call (5 minutes). */ -const DEFAULT_JOB_TIMEOUT_MS = 5 * 60 * 1_000; - -/** Maximum number of concurrent job executions across all plugins. */ -const DEFAULT_MAX_CONCURRENT_JOBS = 10; - -// --------------------------------------------------------------------------- -// Types -// --------------------------------------------------------------------------- - -/** - * Options for creating a PluginJobScheduler. - */ export interface PluginJobSchedulerOptions { - /** Suppress periodic database work while an unclaimed Cloud app stands by. */ isBackgroundWorkEnabled?: () => boolean; - /** Drizzle database instance. */ db: Db; - /** Persistence layer for jobs and runs. */ jobStore: PluginJobStore; - /** Worker process manager for RPC calls. */ workerManager: PluginWorkerManager; - /** Interval between scheduler ticks in ms (default: 30s). */ tickIntervalMs?: number; - /** Timeout for individual job RPC calls in ms (default: 5min). */ jobTimeoutMs?: number; - /** Maximum number of concurrent job executions (default: 10). */ maxConcurrentJobs?: number; } - -/** - * Result of a manual job trigger. - */ -export interface TriggerJobResult { - /** The created run ID. */ - runId: string; - /** The job ID that was triggered. */ - jobId: string; -} - -/** - * Diagnostic information about the scheduler. - */ +export interface TriggerJobResult { runId: string; jobId: string } export interface SchedulerDiagnostics { - /** Whether the tick loop is running. */ running: boolean; - /** Number of jobs currently executing. */ activeJobCount: number; - /** Set of job IDs currently in-flight. */ activeJobIds: string[]; - /** Total number of ticks executed since start. */ tickCount: number; - /** Timestamp of the last tick (ISO 8601). */ lastTickAt: string | null; } - -// --------------------------------------------------------------------------- -// Scheduler -// --------------------------------------------------------------------------- - -/** - * The public interface of the job scheduler. - */ export interface PluginJobScheduler { - /** - * Start the scheduler tick loop. - * - * Safe to call multiple times — subsequent calls are no-ops. - */ - start(): void; - - /** - * Stop the scheduler tick loop. - * - * In-flight job runs are NOT cancelled — they are allowed to finish - * naturally. The tick loop simply stops firing. - */ - stop(): void; - - /** - * Register a plugin with the scheduler. - * - * Computes `nextRunAt` for all active jobs that are missing it. This is - * typically called after a plugin's worker process starts and - * `syncJobDeclarations()` has been called. - * - * @param pluginId - UUID of the plugin - */ + start(coordinator: Coordinator): Promise; + stop(): Promise; registerPlugin(pluginId: string): Promise; - - /** - * Unregister a plugin from the scheduler. - * - * Cancels any in-flight runs for the plugin and removes tracking state. - * - * @param pluginId - UUID of the plugin - */ unregisterPlugin(pluginId: string): Promise; - - /** - * Manually trigger a specific job (outside of the cron schedule). - * - * Creates a run with `trigger: "manual"` and dispatches immediately, - * respecting the overlap prevention check. - * - * @param jobId - UUID of the job to trigger - * @param trigger - What triggered this run (default: "manual") - * @returns The created run info - * @throws {Error} if the job is not found, not active, or already running - */ triggerJob(jobId: string, trigger?: "manual" | "retry"): Promise; - - /** - * Run a single scheduler tick immediately (for testing). - * - * @internal - */ - tick(): Promise; - - /** - * Get diagnostic information about the scheduler state. - */ + tick(signal?: AbortSignal): Promise; diagnostics(): SchedulerDiagnostics; } -// --------------------------------------------------------------------------- -// Implementation -// --------------------------------------------------------------------------- - -/** - * Create a new PluginJobScheduler. - * - * @example - * ```ts - * const scheduler = createPluginJobScheduler({ - * db, - * jobStore, - * workerManager, - * }); - * - * // Start the tick loop - * scheduler.start(); - * - * // When a plugin comes online, register it - * await scheduler.registerPlugin(pluginId); - * - * // Manually trigger a job - * const { runId } = await scheduler.triggerJob(jobId); - * - * // On server shutdown - * scheduler.stop(); - * ``` - */ -export function createPluginJobScheduler( - options: PluginJobSchedulerOptions, -): PluginJobScheduler { - const { - db, - jobStore, - workerManager, - tickIntervalMs = DEFAULT_TICK_INTERVAL_MS, - jobTimeoutMs = DEFAULT_JOB_TIMEOUT_MS, - maxConcurrentJobs = DEFAULT_MAX_CONCURRENT_JOBS, - } = options; +function nextCron(schedule: string, now: Date): Date | null { + return schedule && !validateCron(schedule) ? nextCronTick(parseCron(schedule), now) : null; +} +export function createPluginJobScheduler(options: PluginJobSchedulerOptions): PluginJobScheduler { + const { db, jobStore, workerManager, tickIntervalMs = 1_000, + jobTimeoutMs = 5 * 60_000, maxConcurrentJobs = 10 } = options; const log = logger.child({ service: "plugin-job-scheduler" }); - - // ----------------------------------------------------------------------- - // State - // ----------------------------------------------------------------------- - - /** Timer handle for the tick loop. */ - let tickTimer: ReturnType | null = null; - - /** Whether the scheduler is running. */ - let running = false; - - /** Set of job IDs currently being executed (for overlap prevention). */ const activeJobs = new Set(); - - /** Total number of ticks since start. */ + const pendingSettlements = new Map Promise; running?: Promise }>(); + const inFlight = new Set>(); + const subscriptions = new Map void>(); + const unregistered = new Set(); + let worker: ReturnType | undefined; + let running = false; + let next: number | null = null; let tickCount = 0; - - /** Timestamp of the last tick. */ let lastTickAt: Date | null = null; + let ticking: Promise | undefined; + const canRun = () => running && options.isBackgroundWorkEnabled?.() !== false; - /** Guard against concurrent tick execution. */ - let tickInProgress = false; - - // ----------------------------------------------------------------------- - // Core: tick - // ----------------------------------------------------------------------- - - /** - * A single scheduler tick. Queries for due jobs and dispatches them. - */ - async function tick(): Promise { - if (options.isBackgroundWorkEnabled?.() === false) return; - // Prevent overlapping ticks (in case a tick takes longer than the interval) - if (tickInProgress) { - log.debug("skipping tick — previous tick still in progress"); - return; - } - - tickInProgress = true; - tickCount++; - lastTickAt = new Date(); - - try { - const now = new Date(); - - // Query for jobs whose nextRunAt has passed and are active. - // We include jobs with null nextRunAt since they may have just been - // registered and need their first run calculated. - const dueJobs = await db - .select() - .from(pluginJobs) - .where( - and( - eq(pluginJobs.status, "active"), - lte(pluginJobs.nextRunAt, now), - ), - ); - - if (dueJobs.length === 0) { - return; - } - - log.debug({ count: dueJobs.length }, "found due jobs"); - - // Dispatch each due job (respecting concurrency limits) - const dispatches: Promise[] = []; - - for (const job of dueJobs) { - // Concurrency limit - if (activeJobs.size >= maxConcurrentJobs) { - log.warn( - { maxConcurrentJobs, activeJobCount: activeJobs.size }, - "max concurrent jobs reached, deferring remaining jobs", - ); - break; - } - - // Overlap prevention: skip if this job is already running - if (activeJobs.has(job.id)) { - log.debug( - { jobId: job.id, jobKey: job.jobKey, pluginId: job.pluginId }, - "skipping job — already running (overlap prevention)", - ); - continue; - } - - // Check if the worker is available - if (!workerManager.isRunning(job.pluginId)) { - log.debug( - { jobId: job.id, pluginId: job.pluginId }, - "skipping job — worker not running", - ); - continue; - } - - // Validate cron expression before dispatching - if (!job.schedule) { - log.warn( - { jobId: job.id, jobKey: job.jobKey }, - "skipping job — no schedule defined", - ); - continue; - } - - dispatches.push(dispatchJob(job)); - } - - if (dispatches.length > 0) { - await Promise.allSettled(dispatches); - } - } catch (err) { - log.error( - { err: err instanceof Error ? err.message : String(err) }, - "scheduler tick error", - ); - } finally { - tickInProgress = false; - } + function track(promise: Promise): Promise { + const tracked = trackIdleWork(promise); + inFlight.add(tracked); + void tracked.then(() => inFlight.delete(tracked), () => inFlight.delete(tracked)); + return tracked; } - // ----------------------------------------------------------------------- - // Core: dispatch a single job - // ----------------------------------------------------------------------- + async function settle(jobId: string) { + const entry = pendingSettlements.get(jobId); + if (!entry) return; + entry.running ??= entry.persist().finally(() => { entry.running = undefined; }); + await entry.running; + if (pendingSettlements.get(jobId) === entry) pendingSettlements.delete(jobId); + } - /** - * Dispatch a single job run — create the run record, call the worker, - * record the result, and advance the schedule pointer. - */ - async function dispatchJob( - job: typeof pluginJobs.$inferSelect, - ): Promise { - const { id: jobId, pluginId, jobKey, schedule } = job; - const jobLog = log.child({ jobId, pluginId, jobKey }); - - // Mark as active (overlap prevention) - activeJobs.add(jobId); - - let runId: string | undefined; + async function execute(job: Job, runId: string, trigger: Trigger) { const startedAt = Date.now(); - - try { - // 1. Create run record - const run = await jobStore.createRun({ - jobId, - pluginId, - trigger: "schedule", - }); - runId = run.id; - - jobLog.info({ runId }, "dispatching scheduled job"); - - // 2. Mark run as running - await jobStore.markRunning(runId); - - // 3. Call worker via RPC - await workerManager.call( - pluginId, - "runJob", - { - job: { - jobKey, - runId, - trigger: "schedule" as const, - scheduledAt: (job.nextRunAt ?? new Date()).toISOString(), - }, - }, - jobTimeoutMs, - ); - - // 4. Mark run as succeeded - const durationMs = Date.now() - startedAt; - await jobStore.completeRun(runId, { - status: "succeeded", - durationMs, - }); - - jobLog.info({ runId, durationMs }, "job completed successfully"); - } catch (err) { - const durationMs = Date.now() - startedAt; - const errorMessage = err instanceof Error ? err.message : String(err); - - jobLog.error( - { runId, durationMs, err: errorMessage }, - "job execution failed", - ); - - // Record the failure - if (runId) { - try { - await jobStore.completeRun(runId, { - status: "failed", - error: errorMessage, - durationMs, - }); - } catch (completeErr) { - jobLog.error( - { - runId, - err: completeErr instanceof Error ? completeErr.message : String(completeErr), - }, - "failed to record job failure", - ); - } - } - } finally { - // Remove from active set - activeJobs.delete(jobId); - - // 5. Always advance the schedule pointer (even on failure) - try { - await advanceSchedulePointer(job); - } catch (err) { - jobLog.error( - { err: err instanceof Error ? err.message : String(err) }, - "failed to advance schedule pointer", - ); - } - } - } - - // ----------------------------------------------------------------------- - // Core: manual trigger - // ----------------------------------------------------------------------- - - async function triggerJob( - jobId: string, - trigger: "manual" | "retry" = "manual", - ): Promise { - const job = await jobStore.getJobById(jobId); - if (!job) { - throw new Error(`Job not found: ${jobId}`); - } - - if (job.status !== "active") { - throw new Error( - `Job "${job.jobKey}" is not active (status: ${job.status})`, - ); - } - - // Overlap prevention - if (activeJobs.has(jobId)) { - throw new Error( - `Job "${job.jobKey}" is already running — cannot trigger while in progress`, - ); - } - - // Also check DB for running runs (defensive — covers multi-instance) - const existingRuns = await db - .select() - .from(pluginJobRuns) - .where( - and( - eq(pluginJobRuns.jobId, jobId), - eq(pluginJobRuns.status, "running"), - ), - ); - - if (existingRuns.length > 0) { - throw new Error( - `Job "${job.jobKey}" already has a running execution — cannot trigger while in progress`, - ); - } - - // Check worker availability - if (!workerManager.isRunning(job.pluginId)) { - throw new Error( - `Worker for plugin "${job.pluginId}" is not running — cannot trigger job`, - ); - } - - // Create the run and dispatch (non-blocking) - const run = await jobStore.createRun({ - jobId, - pluginId: job.pluginId, - trigger, - }); - - // Dispatch in background — don't block the caller - void dispatchManualRun(job, run.id, trigger); - - return { runId: run.id, jobId }; - } - - /** - * Dispatch a manually triggered job run. - */ - async function dispatchManualRun( - job: typeof pluginJobs.$inferSelect, - runId: string, - trigger: "manual" | "retry", - ): Promise { - const { id: jobId, pluginId, jobKey } = job; - const jobLog = log.child({ jobId, pluginId, jobKey, runId, trigger }); - - activeJobs.add(jobId); - const startedAt = Date.now(); - + let error: string | undefined; try { await jobStore.markRunning(runId); - - await workerManager.call( - pluginId, - "runJob", - { - job: { - jobKey, - runId, - trigger, - scheduledAt: new Date().toISOString(), - }, - }, - jobTimeoutMs, - ); - - const durationMs = Date.now() - startedAt; - await jobStore.completeRun(runId, { - status: "succeeded", - durationMs, - }); - - jobLog.info({ durationMs }, "manual job completed successfully"); + await workerManager.call(job.pluginId, "runJob", { + job: { jobKey: job.jobKey, runId, trigger, + scheduledAt: (trigger === "schedule" ? job.nextRunAt! : new Date()).toISOString() }, + }, jobTimeoutMs); } catch (err) { - const durationMs = Date.now() - startedAt; - const errorMessage = err instanceof Error ? err.message : String(err); - jobLog.error({ durationMs, err: errorMessage }, "manual job failed"); - - try { - await jobStore.completeRun(runId, { - status: "failed", - error: errorMessage, - durationMs, - }); - } catch (completeErr) { - jobLog.error( - { - err: completeErr instanceof Error ? completeErr.message : String(completeErr), - }, - "failed to record manual job failure", - ); - } - } finally { - activeJobs.delete(jobId); + error = err instanceof Error ? err.message : String(err); + log.error({ jobId: job.id, runId, err }, "plugin job failed"); } + const finishedAt = new Date(); + pendingSettlements.set(job.id, { persist: () => db.transaction(async tx => { + 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 { + activeJobs.add(job.id); + try { + const run = await db.transaction(async tx => { + const [current] = await tx.select().from(pluginJobs).where(eq(pluginJobs.id, job.id)).for("update"); + if (!current || current.status !== "active" || signal?.aborted || !canRun() || unregistered.has(job.pluginId) + || !workerManager.isRunning(job.pluginId)) return null; + if (trigger === "schedule" && (current.schedule !== job.schedule + || current.nextRunAt?.getTime() !== job.nextRunAt?.getTime())) return null; + const [outstanding] = await tx.select({ id: pluginJobRuns.id }).from(pluginJobRuns).where(and( + eq(pluginJobRuns.jobId, job.id), inArray(pluginJobRuns.status, ["pending", "queued", "running"]), + )).limit(1); + if (outstanding) return null; + return pluginJobStore(tx).createRun({ jobId: job.id, pluginId: job.pluginId, trigger }); + }); + if (!run) { + activeJobs.delete(job.id); + worker?.wake(); + if (trigger !== "schedule") throw new Error(`Job "${job.jobKey}" is no longer available`); + return null; + } + const completion = track(execute(job, run.id, trigger)); + void completion.catch(err => { log.error({ err, jobId: job.id }, "plugin job settlement deferred"); worker?.wake(); }); + return { runId: run.id, jobId: job.id }; + } catch (err) { activeJobs.delete(job.id); throw err; } + } - /** - * Advance the `lastRunAt` and `nextRunAt` timestamps on a job after a run. - */ - async function advanceSchedulePointer( - job: typeof pluginJobs.$inferSelect, - ): Promise { + async function refresh(signal?: AbortSignal) { + for (const jobId of pendingSettlements.keys()) await settle(jobId); + const jobs = await db.select().from(pluginJobs).where(eq(pluginJobs.status, "active")); + const runnable = jobs.filter(job => !unregistered.has(job.pluginId) && workerManager.isRunning(job.pluginId) + && !activeJobs.has(job.id) && job.schedule && !validateCron(job.schedule)); + const outstanding = runnable.length === 0 ? [] : await db.select({ jobId: pluginJobRuns.jobId }) + .from(pluginJobRuns).where(and(inArray(pluginJobRuns.jobId, runnable.map(job => job.id)), + inArray(pluginJobRuns.status, ["pending", "queued", "running"]))); + const blocked = new Set(outstanding.map(run => run.jobId)); const now = new Date(); - let nextRunAt: Date | null = null; - - if (job.schedule) { - const validationError = validateCron(job.schedule); - if (validationError) { - log.warn( - { jobId: job.id, schedule: job.schedule, error: validationError }, - "invalid cron schedule — cannot compute next run", - ); - } else { - const cron = parseCron(job.schedule); - nextRunAt = nextCronTick(cron, now); - } - } - - await jobStore.updateRunTimestamps(job.id, now, nextRunAt); - } - - /** - * Ensure all active jobs for a plugin have a `nextRunAt` value. - * Called when a plugin is registered with the scheduler. - */ - async function ensureNextRunTimestamps(pluginId: string): Promise { - const jobs = await jobStore.listJobs(pluginId, "active"); - - for (const job of jobs) { - // Skip jobs that already have a valid nextRunAt in the future - if (job.nextRunAt && job.nextRunAt.getTime() > Date.now()) { - continue; - } - - // Skip jobs without a schedule - if (!job.schedule) { - continue; - } - - const validationError = validateCron(job.schedule); - if (validationError) { - log.warn( - { jobId: job.id, jobKey: job.jobKey, schedule: job.schedule, error: validationError }, - "skipping job with invalid cron schedule", - ); - continue; - } - - const cron = parseCron(job.schedule); - const nextRunAt = nextCronTick(cron, new Date()); - - if (nextRunAt) { - await jobStore.updateRunTimestamps( - job.id, - job.lastRunAt ?? new Date(0), - nextRunAt, - ); - log.debug( - { jobId: job.id, jobKey: job.jobKey, nextRunAt: nextRunAt.toISOString() }, - "computed nextRunAt for job", - ); - } - } - } - - // ----------------------------------------------------------------------- - // Plugin registration - // ----------------------------------------------------------------------- - - async function registerPlugin(pluginId: string): Promise { - log.info({ pluginId }, "registering plugin with job scheduler"); - await ensureNextRunTimestamps(pluginId); - } - - async function unregisterPlugin(pluginId: string): Promise { - log.info({ pluginId }, "unregistering plugin from job scheduler"); - - // Cancel any in-flight run records for this plugin that are still - // queued or running. Active jobs in-memory will finish naturally. - try { - const runningRuns = await db - .select() - .from(pluginJobRuns) - .where( - and( - eq(pluginJobRuns.pluginId, pluginId), - or( - eq(pluginJobRuns.status, "running"), - eq(pluginJobRuns.status, "queued"), - ), - ), - ); - - for (const run of runningRuns) { - await jobStore.completeRun(run.id, { - status: "cancelled", - error: "Plugin unregistered", - durationMs: run.startedAt - ? Date.now() - run.startedAt.getTime() - : null, + const dispatches: Promise[] = []; + next = null; + for (const job of runnable) { + if (!canRun() || signal?.aborted) break; + if (blocked.has(job.id)) 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 = results.find(result => result.status === "rejected"); + if (failed?.status === "rejected") throw failed.reason; } - // ----------------------------------------------------------------------- - // Lifecycle: start / stop - // ----------------------------------------------------------------------- - - function start(): void { - if (running) { - log.debug("scheduler already running"); - return; - } - - running = true; - tickTimer = setInterval(() => { - void tick(); - }, tickIntervalMs); - - log.info( - { tickIntervalMs, maxConcurrentJobs }, - "plugin job scheduler started", - ); + function tick(signal?: AbortSignal): Promise { + if (!canRun()) return Promise.resolve(); + if (ticking) return ticking; + tickCount++; lastTickAt = new Date(); + ticking = refresh(signal).finally(() => { ticking = undefined; }); + return ticking; } - function stop(): void { - // Always clear the timer defensively, even if `running` is already false, - // to prevent leaked interval timers. - if (tickTimer !== null) { - clearInterval(tickTimer); - tickTimer = null; - } - - if (!running) return; - running = false; - - log.info( - { activeJobCount: activeJobs.size }, - "plugin job scheduler stopped", - ); - } - - // ----------------------------------------------------------------------- - // Diagnostics - // ----------------------------------------------------------------------- - - function diagnostics(): SchedulerDiagnostics { - return { - running, - activeJobCount: activeJobs.size, - activeJobIds: [...activeJobs], - tickCount, - lastTickAt: lastTickAt?.toISOString() ?? null, - }; - } - - // ----------------------------------------------------------------------- - // Public API - // ----------------------------------------------------------------------- - return { - start, - stop, - registerPlugin, - unregisterPlugin, - triggerJob, + async start(coordinator) { + if (worker) { await worker.ready; return; } + running = true; + worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.pluginSchedules, { + run: tick, next: async () => next, canRun, retryMs: tickIntervalMs, + }); + 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]); + }, + 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) && ["running", "queued", "pending"].includes(run.status)) { + await jobStore.completeRun(run.id, { status: "cancelled", error: "Plugin unregistered", + durationMs: run.startedAt ? Date.now() - run.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`); + activeJobs.add(jobId); + try { + 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; + } catch (err) { activeJobs.delete(jobId); throw err; } + })()); + }, tick, - diagnostics, + diagnostics: () => ({ running, activeJobCount: activeJobs.size, activeJobIds: [...activeJobs], tickCount, + lastTickAt: lastTickAt?.toISOString() ?? null }), }; } diff --git a/server/src/services/plugin-job-store.ts b/server/src/services/plugin-job-store.ts index f5fdbce1a42..07775a862fa 100644 --- a/server/src/services/plugin-job-store.ts +++ b/server/src/services/plugin-job-store.ts @@ -1,3 +1,4 @@ +import { writePluginJobChange, type PluginJobDatabase } from "./plugin-job-work-notifications.js"; /** * Plugin Job Store — persistence layer for scheduled plugin jobs and their * execution history. @@ -100,12 +101,12 @@ export interface CompleteJobRunInput { * }); * ``` */ -export function pluginJobStore(db: Db) { +export function pluginJobStore(db: PluginJobDatabase) { // ----------------------------------------------------------------------- // Internal helpers // ----------------------------------------------------------------------- - async function assertPluginExists(pluginId: string): Promise { + async function assertPluginExists(db: PluginJobDatabase, pluginId: string): Promise { const rows = await db .select({ id: plugins.id }) .from(plugins) @@ -145,63 +146,66 @@ export function pluginJobStore(db: Db) { pluginId: string, declarations: PluginJobDeclaration[], ): Promise { - await assertPluginExists(pluginId); + return writePluginJobChange(db, async db => { + await assertPluginExists(db, pluginId); - // Fetch existing jobs for this plugin - const existingJobs = await db - .select() - .from(pluginJobs) - .where(eq(pluginJobs.pluginId, pluginId)); + // Fetch existing jobs for this plugin + const existingJobs = await db + .select() + .from(pluginJobs) + .where(eq(pluginJobs.pluginId, pluginId)); - const existingByKey = new Map( - existingJobs.map((j) => [j.jobKey, j]), - ); + const existingByKey = new Map( + existingJobs.map((j) => [j.jobKey, j]), + ); - const declaredKeys = new Set(); + const declaredKeys = new Set(); - // Upsert each declared job - for (const decl of declarations) { - declaredKeys.add(decl.jobKey); + // Upsert each declared job + for (const decl of declarations) { + declaredKeys.add(decl.jobKey); - const existing = existingByKey.get(decl.jobKey); - const schedule = decl.schedule ?? ""; + const existing = existingByKey.get(decl.jobKey); + const schedule = decl.schedule ?? ""; - if (existing) { - // Update schedule if it changed; re-activate if it was paused - const updates: Record = { - updatedAt: new Date(), - }; - if (existing.schedule !== schedule) { - updates.schedule = schedule; + if (existing) { + // Update schedule if it changed; re-activate if it was paused + const updates: Record = { + updatedAt: new Date(), + }; + if (existing.schedule !== schedule) { + updates.schedule = schedule; + updates.nextRunAt = null; + } + if (existing.status === "paused") { + updates.status = "active"; + } + + await db + .update(pluginJobs) + .set(updates) + .where(eq(pluginJobs.id, existing.id)); + } else { + // Insert new job + await db.insert(pluginJobs).values({ + pluginId, + jobKey: decl.jobKey, + schedule, + status: "active", + }); } - if (existing.status === "paused") { - updates.status = "active"; + } + + // Pause jobs that are no longer declared in the manifest + for (const existing of existingJobs) { + if (!declaredKeys.has(existing.jobKey) && existing.status !== "paused") { + await db + .update(pluginJobs) + .set({ status: "paused", updatedAt: new Date() }) + .where(eq(pluginJobs.id, existing.id)); } - - await db - .update(pluginJobs) - .set(updates) - .where(eq(pluginJobs.id, existing.id)); - } else { - // Insert new job - await db.insert(pluginJobs).values({ - pluginId, - jobKey: decl.jobKey, - schedule, - status: "active", - }); } - } - - // Pause jobs that are no longer declared in the manifest - for (const existing of existingJobs) { - if (!declaredKeys.has(existing.jobKey) && existing.status !== "paused") { - await db - .update(pluginJobs) - .set({ status: "paused", updatedAt: new Date() }) - .where(eq(pluginJobs.id, existing.id)); - } - } + }); }, /** @@ -290,10 +294,12 @@ export function pluginJobStore(db: Db) { jobId: string, status: JobDefinitionStatus, ): Promise { - await db - .update(pluginJobs) - .set({ status, updatedAt: new Date() }) - .where(eq(pluginJobs.id, jobId)); + return writePluginJobChange(db, async db => { + await db + .update(pluginJobs) + .set({ status, updatedAt: new Date() }) + .where(eq(pluginJobs.id, jobId)); + }); }, /** @@ -311,14 +317,16 @@ export function pluginJobStore(db: Db) { lastRunAt: Date, nextRunAt: Date | null, ): Promise { - await db - .update(pluginJobs) - .set({ - lastRunAt, - nextRunAt, - updatedAt: new Date(), - }) - .where(eq(pluginJobs.id, jobId)); + return writePluginJobChange(db, async db => { + await db + .update(pluginJobs) + .set({ + lastRunAt, + nextRunAt, + updatedAt: new Date(), + }) + .where(eq(pluginJobs.id, jobId)); + }); }, /** @@ -329,9 +337,11 @@ export function pluginJobStore(db: Db) { * @param pluginId - UUID of the owning plugin */ async deleteAllJobs(pluginId: string): Promise { - await db - .delete(pluginJobs) - .where(eq(pluginJobs.pluginId, pluginId)); + return writePluginJobChange(db, async db => { + await db + .delete(pluginJobs) + .where(eq(pluginJobs.pluginId, pluginId)); + }); }, // ===================================================================== @@ -351,17 +361,19 @@ export function pluginJobStore(db: Db) { async createRun( input: CreateJobRunInput, ): Promise { - const rows = await db - .insert(pluginJobRuns) - .values({ - jobId: input.jobId, - pluginId: input.pluginId, - trigger: input.trigger, - status: "queued", - }) - .returning(); + return writePluginJobChange(db, async db => { + const rows = await db + .insert(pluginJobRuns) + .values({ + jobId: input.jobId, + pluginId: input.pluginId, + trigger: input.trigger, + status: "queued", + }) + .returning(); - return rows[0]!; + return rows[0]!; + }); }, /** @@ -370,13 +382,15 @@ export function pluginJobStore(db: Db) { * @param runId - UUID of the run row */ async markRunning(runId: string): Promise { - await db - .update(pluginJobRuns) - .set({ - status: "running" as PluginJobRunStatus, - startedAt: new Date(), - }) - .where(eq(pluginJobRuns.id, runId)); + return writePluginJobChange(db, async db => { + await db + .update(pluginJobRuns) + .set({ + status: "running" as PluginJobRunStatus, + startedAt: new Date(), + }) + .where(eq(pluginJobRuns.id, runId)); + }); }, /** @@ -390,15 +404,17 @@ export function pluginJobStore(db: Db) { runId: string, input: CompleteJobRunInput, ): Promise { - await db - .update(pluginJobRuns) - .set({ - status: input.status, - error: input.error ?? null, - durationMs: input.durationMs ?? null, - finishedAt: new Date(), - }) - .where(eq(pluginJobRuns.id, runId)); + return writePluginJobChange(db, async db => { + await db + .update(pluginJobRuns) + .set({ + status: input.status, + error: input.error ?? null, + durationMs: input.durationMs ?? null, + finishedAt: new Date(), + }) + .where(eq(pluginJobRuns.id, runId)); + }); }, /** diff --git a/server/src/services/plugin-job-work-notifications.ts b/server/src/services/plugin-job-work-notifications.ts new file mode 100644 index 00000000000..3ce7ab287c0 --- /dev/null +++ b/server/src/services/plugin-job-work-notifications.ts @@ -0,0 +1,11 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type PluginJobDatabase = Pick; + +export function writePluginJobChange(db: PluginJobDatabase, write: (tx: PluginJobDatabase) => Promise): Promise { + return db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.pluginSchedules); + return write(tx); + }); +} diff --git a/server/src/services/plugin-loader.ts b/server/src/services/plugin-loader.ts index c07e48ac2e3..283cbc6c401 100644 --- a/server/src/services/plugin-loader.ts +++ b/server/src/services/plugin-loader.ts @@ -2144,7 +2144,7 @@ export function pluginLoader( const { workerManager, jobScheduler } = runtimeServices; // 1. Stop the job scheduler tick loop - jobScheduler.stop(); + await jobScheduler.stop(); // 2. Stop all worker processes await workerManager.stopAll(); @@ -2416,9 +2416,9 @@ export function pluginLoader( // 6. Sync job declarations and register with scheduler // ------------------------------------------------------------------ const jobDeclarations = manifest.jobs ?? []; + await jobStore.syncJobDeclarations(pluginId, jobDeclarations); + await jobScheduler.registerPlugin(pluginId); if (jobDeclarations.length > 0) { - await jobStore.syncJobDeclarations(pluginId, jobDeclarations); - await jobScheduler.registerPlugin(pluginId); registered.jobs = jobDeclarations.length; log.info( diff --git a/server/src/services/plugin-registry.ts b/server/src/services/plugin-registry.ts index c47cf4f6f5d..f03c9006436 100644 --- a/server/src/services/plugin-registry.ts +++ b/server/src/services/plugin-registry.ts @@ -1,3 +1,4 @@ +import { writePluginJobChange } from "./plugin-job-work-notifications.js"; import { asc, eq, isNull, ne, sql, and } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -267,25 +268,27 @@ export function pluginRegistryService(db: Db) { const plugin = await getById(id); if (!plugin) throw notFound("Plugin not found"); - if (removeData) { - // Hard delete – plugin_config cascades via FK onDelete + return writePluginJobChange(db, async db => { + if (removeData) { + // Hard delete – plugin_config cascades via FK onDelete + return db + .delete(plugins) + .where(eq(plugins.id, id)) + .returning() + .then((rows) => rows[0] ?? null); + } + + // Soft delete – mark as uninstalled return db - .delete(plugins) + .update(plugins) + .set({ + status: "uninstalled" as PluginStatus, + updatedAt: new Date(), + }) .where(eq(plugins.id, id)) .returning() .then((rows) => rows[0] ?? null); - } - - // Soft delete – mark as uninstalled - return db - .update(plugins) - .set({ - status: "uninstalled" as PluginStatus, - updatedAt: new Date(), - }) - .where(eq(plugins.id, id)) - .returning() - .then((rows) => rows[0] ?? null); + }); }, // ----- Config --------------------------------------------------------- @@ -670,35 +673,38 @@ 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))) + .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 +727,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 +760,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 -------------------------------------------------------