mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 09:37:19 +01:00
fix(server): schedule plugin jobs from saved deadlines
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
553e69ffea
commit
8b4f551dc2
15 files changed
+698
-870
No files matched your search
@@ -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: {
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
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, { 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<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 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<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("deleting the last job cancels its future deadline", async () => {
|
||||
await job(); await start(); await settled(); await pluginJobStore(db).deleteAllJobs(pluginId); await quiet();
|
||||
});
|
||||
});
|
||||
+2
-2
@@ -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();
|
||||
|
||||
@@ -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(); } };
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
@@ -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,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<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). */
|
||||
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 = 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<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 => {
|
||||
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> {
|
||||
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) {
|
||||
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<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)) 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<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,
|
||||
});
|
||||
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 }),
|
||||
};
|
||||
}
|
||||
@@ -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);
|
||||
});
|
||||
}
|
||||
@@ -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(
|
||||
|
||||
@@ -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 -------------------------------------------------------
|
||||
|
||||
Reference in new issue
Block a user