From da062a397189acbf1b7bd79e4eb762770ef70287 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Thu, 15 Oct 2026 17:44:38 -0700 Subject: [PATCH] Dispatch scheduled work from saved deadlines Routine and issue-monitor work can wait for a heartbeat scan even when the app wakes before a saved deadline. Register independent deadline workers with the existing delivery coordinator and notify them after producer transactions settle. Preserve the interval recovery scan, execution checks, atomic claims, and idle/standby/shutdown gates. Parse each cron expression once per next-run search to avoid repeating that work for every minute of a sparse schedule. Verified focused scheduler, routine, SQL eligibility, and notification tests, workspace typecheck, and full build. Full-suite and real local dispatch acceptance are tracked in the PR before final readiness. Co-Authored-By: Paperclip --- doc/DEVELOPING.md | 12 ++ server/src/__tests__/routines-service.test.ts | 36 +++++ .../__tests__/scheduled-work-postgres.test.ts | 65 +++++++++ server/src/__tests__/scheduled-work.test.ts | 132 ++++++++++++++++++ server/src/index.ts | 42 +++--- .../services/delivery-work-notifications.ts | 2 + server/src/services/issues.ts | 6 + .../services/routines-formatter-cache.test.ts | 14 ++ server/src/services/routines.ts | 20 ++- server/src/services/scheduled-work.ts | 45 ++++++ 10 files changed, 344 insertions(+), 30 deletions(-) create mode 100644 server/src/__tests__/scheduled-work-postgres.test.ts create mode 100644 server/src/__tests__/scheduled-work.test.ts create mode 100644 server/src/services/scheduled-work.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 8a9bfe8d0b5..50593f2c53e 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1448,6 +1448,18 @@ When a workspace service runs Paperclip for browser OAuth QA, configure its `exp ## Heartbeat Service Extractions +Routine and issue-monitor dispatch uses saved future deadlines through +`server/src/services/scheduled-work.ts` and the existing delivery coordinator. +Startup restores the next deadline. Routine edits and monitor changes notify +their worker after the enclosing database transaction settles. Each worker runs +independently and retains the existing atomic claims and execution checks. +The heartbeat interval still requests a recovery scan; it does not determine +the dispatch time of a known future deadline. Failed scans retry at that same +interval. Future timers do not hold idle work open, and standby, idle drain and +shutdown retain the coordinator's admission gates. External database writers +rely on the periodic recovery scan. Sparse cron searches reuse one parsed +expression so repeated parsing cannot block unrelated deadlines. + Keep relevant heartbeat extractions and their focused tests in `server/src/services/heartbeat/`. Follow [Heartbeat execution boundaries](../rules/heartbeat-execution-boundaries.mdc) for state lifetime, admission, recovery ownership, cleanup ordering, and verification. diff --git a/server/src/__tests__/routines-service.test.ts b/server/src/__tests__/routines-service.test.ts index 9d4949e30ab..1f01bc32801 100644 --- a/server/src/__tests__/routines-service.test.ts +++ b/server/src/__tests__/routines-service.test.ts @@ -1,3 +1,5 @@ +import { subscribeDatabaseWork } from "@paperclipai/db"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { createHmac, randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; @@ -214,6 +216,40 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => { return { companyId, agentId, issueSvc, projectId, routine, svc, wakeups }; } + it("notifies scheduled routine work after trigger transactions settle", async () => { + const f = await seedFixture(); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.routineSchedules, event => { events.push(event); }); + try { + const created = await f.svc.createTrigger(f.routine.id, + { kind: "schedule", cronExpression: "0 * * * *", timezone: "UTC" }, {}); + expect(events).toEqual(["intent", "settled"]); + events.length = 0; + await f.svc.updateTrigger(created.trigger.id, { cronExpression: "30 * * * *" }, {}); + expect(events).toEqual(["intent", "settled"]); + events.length = 0; + await f.svc.deleteTrigger(created.trigger.id, {}); + expect(events).toEqual(["intent", "settled"]); + } finally { unsubscribe(); } + }); + + it("notifies monitor work when an issue deadline is created and changed", async () => { + const f = await seedFixture(); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.issueMonitors, event => { events.push(event); }); + const next = new Date(Date.now() + 600_000); + try { + const issue = await f.issueSvc.create(f.companyId, { + title: "Scheduled check", status: "in_progress", assigneeAgentId: f.agentId, + executionPolicy: { mode: "normal", commentRequired: true, stages: [], monitor: { nextCheckAt: next.toISOString() } }, + }); + expect(events).toEqual(["intent", "settled"]); + events.length = 0; + await f.issueSvc.update(issue.id, { monitorNextCheckAt: new Date(next.getTime() + 60_000) }); + expect(events).toEqual(["intent", "settled"]); + } finally { unsubscribe(); } + }); + async function armWorktreeExecution(cutoff: Date, instanceId = "worktree-routines-test") { await db.insert(instanceSettings).values({ singletonKey: "default", diff --git a/server/src/__tests__/scheduled-work-postgres.test.ts b/server/src/__tests__/scheduled-work-postgres.test.ts new file mode 100644 index 00000000000..fb4d9a8d180 --- /dev/null +++ b/server/src/__tests__/scheduled-work-postgres.test.ts @@ -0,0 +1,65 @@ +import { agents, companies, createDb, issues, routines, routineTriggers } from "@paperclipai/db"; +import { sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import type { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, type DeliveryQueue } from "../services/delivery-work-notifications.js"; +import { registerScheduledWork } from "../services/scheduled-work.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("scheduled deadline database queries", () => { + let database: Awaited>; + let db: Db; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-scheduled-deadlines"); + db = createDb(database.connectionString); + }, 30_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + it("selects future eligible deadlines and excludes inactive, mismatched and past work", async () => { + const now = Date.now(); + const at = (minutes: number) => new Date(now + minutes * 60_000); + const [company, pausedCompany] = await db.insert(companies).values([ + { name: "Active schedules", issuePrefix: "SCH" }, + { name: "Paused schedules", issuePrefix: "PAU", status: "paused" }, + ]).returning(); + const [agent] = await db.insert(agents).values({ companyId: company!.id, name: "Scheduled agent" }).returning(); + const [active, paused, otherCompany] = await db.insert(routines).values([ + { companyId: company!.id, title: "Active routine" }, + { companyId: company!.id, title: "Paused routine", status: "paused" }, + { companyId: pausedCompany!.id, title: "Paused company routine" }, + ]).returning(); + const trigger = { companyId: company!.id, routineId: active!.id, kind: "schedule" }; + await db.insert(routineTriggers).values([ + { ...trigger, nextRunAt: at(10) }, + { ...trigger, enabled: false, nextRunAt: at(1) }, + { ...trigger, archived: true, nextRunAt: at(2) }, + { ...trigger, routineId: paused!.id, nextRunAt: at(3) }, + { ...trigger, companyId: pausedCompany!.id, nextRunAt: at(4) }, + { ...trigger, companyId: pausedCompany!.id, routineId: otherCompany!.id, nextRunAt: at(5) }, + { ...trigger, nextRunAt: at(-1) }, + ]); + const monitor = { companyId: company!.id, title: "Monitor", status: "in_progress", assigneeAgentId: agent!.id }; + await db.insert(issues).values([ + { ...monitor, monitorNextCheckAt: at(15) }, + { ...monitor, status: "done", monitorNextCheckAt: at(1) }, + { ...monitor, assigneeUserId: "fixture-user", monitorNextCheckAt: at(2) }, + { ...monitor, assigneeAgentId: null, monitorNextCheckAt: at(3) }, + { ...monitor, companyId: pausedCompany!.id, monitorNextCheckAt: at(4) }, + { ...monitor, monitorNextCheckAt: at(-1) }, + ]); + type Task = Parameters["register"]>[1]; + const tasks = new Map(); + registerScheduledWork({ register: (queue, task) => { + tasks.set(queue, task); + return { ready: Promise.resolve(), wake: () => {} }; + } }, db, { retryMs: 60_000, tickTimers: async () => {}, tickRoutines: async () => {} }); + for (const task of tasks.values()) await task.run(new AbortController().signal); + expect(tasks.get(DELIVERY_QUEUES.routineSchedules)!.nextRunAt!()).toBe(at(10).getTime()); + expect(tasks.get(DELIVERY_QUEUES.issueMonitors)!.nextRunAt!()).toBe(at(15).getTime()); + await db.execute(sql`truncate table companies restart identity cascade`); + for (const task of tasks.values()) await task.run(new AbortController().signal); + for (const task of tasks.values()) expect(task.nextRunAt!()).toBeNull(); + }); +}); diff --git a/server/src/__tests__/scheduled-work.test.ts b/server/src/__tests__/scheduled-work.test.ts new file mode 100644 index 00000000000..4194f9d7d9c --- /dev/null +++ b/server/src/__tests__/scheduled-work.test.ts @@ -0,0 +1,132 @@ +import { PgDialect } from "drizzle-orm/pg-core"; +import type { Db } from "@paperclipai/db"; +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js"; +import { registerScheduledWork } from "../services/scheduled-work.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const coordinators: ReturnType[] = []; +afterEach(async () => { + for (const coordinator of coordinators.splice(0)) await coordinator.stop(); + vi.useRealTimers(); +}); + +function fixture(input: { routine?: number | null; monitor?: number | null; tickRoutines?: () => Promise } = {}) { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2020-01-01T00:00:00.000Z")); + const dialect = new PgDialect(); + const deadlines = { routine: input.routine ?? null, monitor: input.monitor ?? null }; + let canRun = true; + let failRead = false; + const execute = vi.fn(async (query: any) => { + const text = dialect.sqlToQuery(query).sql; + if (text.includes("pg_current_xact_id")) return [{ xid: "42" }]; + if (failRead) throw new Error("database unavailable"); + const at = deadlines[text.includes("routine_triggers") ? "routine" : "monitor"]; + return [{ next_at: at !== null && at > Date.now() ? new Date(at) : null }]; + }); + const owner = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise) { + return callback({ execute, transaction: owner.transaction }); + } }); + const errors = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, onError: errors }); + coordinators.push(coordinator); + const tickTimers = vi.fn(async () => {}); + const tickRoutines = vi.fn(input.tickRoutines ?? (async () => {})); + const workers = registerScheduledWork(coordinator, owner as unknown as Db, + { retryMs: 60_000, tickTimers, tickRoutines }); + return { coordinator, workers, owner, deadlines, execute, tickTimers, tickRoutines, errors, + ready: () => Promise.all([workers.routines.ready, workers.timers.ready]), + canRun: (value: boolean) => { canRun = value; }, failRead: (value: boolean) => { failRead = value; } }; +} + +describe("scheduled routine and monitor deadlines", () => { + it("dispatches at each saved deadline without waiting for a periodic tick", async () => { + const now = Date.UTC(2020, 0, 1); + const f = fixture({ routine: now + 10_000, monitor: now + 20_000 }); + await f.ready(); + expect(f.coordinator.nextWakeAt()).toBe(now + 10_000); + expect(idleWorkSnapshot().active).toBe(0); + await vi.advanceTimersByTimeAsync(9999); + expect(f.tickRoutines).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + expect(f.tickRoutines).toHaveBeenCalledTimes(2); + expect(f.tickTimers).toHaveBeenCalledTimes(1); + expect(f.coordinator.nextWakeAt()).toBe(now + 20_000); + await vi.advanceTimersByTimeAsync(10_000); + expect(f.tickTimers).toHaveBeenCalledTimes(2); + expect(f.coordinator.nextWakeAt()).toBeNull(); + expect(f.execute).toHaveBeenCalledTimes(4); + expect(idleWorkSnapshot().active).toBe(0); + await vi.advanceTimersByTimeAsync(3_600_000); + expect(f.execute).toHaveBeenCalledTimes(4); + }); + + it("rearms an earlier deadline only after the producer transaction settles", async () => { + const now = Date.UTC(2020, 0, 1); + const f = fixture({ routine: now + 600_000 }); + await f.ready(); + let commit!: () => void; + const held = new Promise(resolve => { commit = resolve; }); + const writing = f.owner.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); + f.deadlines.routine = now + 10_000; + await held; + }); + try { + await vi.advanceTimersByTimeAsync(1000); + expect(f.tickRoutines).toHaveBeenCalledTimes(1); + expect(idleWorkSnapshot().active).toBe(1); + } finally { commit(); await writing; } + await vi.advanceTimersByTimeAsync(1); + expect(f.coordinator.nextWakeAt()).toBe(now + 10_000); + expect(idleWorkSnapshot().active).toBe(0); + await vi.advanceTimersByTimeAsync(8999); + expect(f.tickRoutines).toHaveBeenCalledTimes(3); + expect(f.tickTimers).toHaveBeenCalledTimes(1); + }); + + it("keeps monitor dispatch responsive while routine work is slow", async () => { + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const f = fixture({ monitor: Date.UTC(2020, 0, 1) + 10_000, tickRoutines: () => held }); + try { + await f.workers.timers.ready; + await vi.advanceTimersByTimeAsync(10_000); + expect(f.tickTimers).toHaveBeenCalledTimes(2); + expect(f.tickRoutines).toHaveBeenCalledTimes(1); + } finally { release(); await f.workers.routines.ready; } + }); + + it("does not dispatch during idle drain or standby and cancels deadlines on stop", async () => { + const f = fixture({ routine: Date.UTC(2020, 0, 1) + 10_000 }); + await f.ready(); + f.canRun(false); + await vi.advanceTimersByTimeAsync(10_000); + expect(f.tickRoutines).toHaveBeenCalledTimes(1); + expect(idleWorkSnapshot().active).toBe(0); + await f.coordinator.stop(); + f.canRun(true); + await vi.advanceTimersByTimeAsync(3_600_000); + expect(f.tickRoutines).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + + it("retries failed deadline reads at the fallback interval without a tight loop", async () => { + const f = fixture(); + await f.ready(); + f.failRead(true); + f.workers.routines.wake(); + await vi.advanceTimersByTimeAsync(1); + expect(f.errors).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(59_000); + expect(f.tickRoutines).toHaveBeenCalledTimes(2); + f.failRead(false); + await vi.advanceTimersByTimeAsync(1000); + expect(f.tickRoutines).toHaveBeenCalledTimes(3); + expect(f.coordinator.nextWakeAt()).toBeNull(); + expect(idleWorkSnapshot().active).toBe(0); + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index b76a63fb107..bc27e1a509e 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -7,6 +7,7 @@ import { } from "./services/skill-version-retention.js"; import { subscribeAllCompanyLiveEvents } from "./services/live-events.js"; import { DELIVERY_QUEUES } from "./services/delivery-work-notifications.js"; +import { registerScheduledWork } from "./services/scheduled-work.js"; import type { createDeliveryWorkCoordinator } from "./services/delivery-work-coordinator.js"; import { registerTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "./services/terminal-workspace-work.js"; import { isIdleTaskDrainActive, beginIdleTrackedWork, trackIdleWork } from "./services/task-admission.js"; @@ -1718,7 +1719,7 @@ async function startServerWithDatabaseTeardown( await step("retention_sweep", runRetentionSweep); }; - const startHeartbeatScheduler = () => startHeartbeatSchedulerInterval(() => { + const createHeartbeatSchedulerTick = (scheduled: ReturnType) => () => { // Track the outer async callback as well as the work it starts. Shutdown // can then wait through an already-running suppression check before it // captures the authoritative set of running heartbeat rows. @@ -1738,18 +1739,7 @@ async function startServerWithDatabaseTeardown( ); } - if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { - trackHeartbeatSchedulerWork(heartbeat - .tickTimers(new Date()) - .then((result) => { - if (result.enqueued > 0) { - logger.info({ ...result }, "heartbeat timer tick enqueued runs"); - } - }) - .catch((err) => { - logger.error({ err }, "heartbeat timer tick failed"); - })); - } + scheduled.timers.wake(); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleExternalObjectRefreshSweep(new Date()); @@ -1763,16 +1753,7 @@ async function startServerWithDatabaseTeardown( scheduleEnvironmentLeaseCleanupSweep(); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - trackHeartbeatSchedulerWork(routines - .tickScheduledTriggers(new Date()) - .then((result) => { - if (result.triggered > 0) { - logger.info({ ...result }, "routine scheduler tick enqueued runs"); - } - }) - .catch((err) => { - logger.error({ err }, "routine scheduler tick failed"); - })); + scheduled.routines.wake(); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork((async () => { @@ -1896,7 +1877,20 @@ async function startServerWithDatabaseTeardown( })().catch((err) => { logger.error({ err }, "heartbeat scheduler tick failed"); })); - }); + }; + const startHeartbeatScheduler = () => { + const scheduled = registerScheduledWork(deliveryWork, db, { + retryMs: config.heartbeatSchedulerIntervalMs, + tickTimers: () => heartbeat.tickTimers(new Date()).then(result => { + if (result.enqueued > 0) logger.info({ ...result }, "heartbeat timer tick enqueued runs"); + }), + tickRoutines: () => routines.tickScheduledTriggers(new Date()).then(result => { + if (result.triggered > 0) logger.info({ ...result }, "routine scheduler tick enqueued runs"); + }), + }); + trackHeartbeatSchedulerWork(Promise.all([scheduled.timers.ready, scheduled.routines.ready])); + startHeartbeatSchedulerInterval(createHeartbeatSchedulerTick(scheduled)); + }; trackHeartbeatSchedulerWork(runStartupHousekeeping().then(() => { if (!heartbeatSchedulerStopped) startHeartbeatScheduler(); })); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 88387c21709..e03b811e275 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -27,6 +27,8 @@ export const DELIVERY_QUEUES = { pluginLifecycle: "plugin-resource-lifecycle", toolAction: "tool-action-receipts", terminalWorkspaces: "terminal-workspaces", + routineSchedules: "routine-schedules", + issueMonitors: "issue-monitors", } as const; export type DeliveryQueue = typeof DELIVERY_QUEUES[keyof typeof DELIVERY_QUEUES]; diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 584c483612f..aaae404ac66 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -9,6 +9,7 @@ import { type TaskWorkspaceSelection, type TaskWorkspaceIntent } from "@papercli import { executionWorkspaceService } from "./execution-workspaces.js"; import { notifyChatPublicationWork } from "./chat-work-notifications.js"; import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { recordChatHandoff, recordChatCompletion, existingChatCompletionReply, acknowledgeChatCompletionReply } from "./chat-completion-delivery.js"; import { mirrorSlackBoardComment, slackBoardReplyBindings } from "./slack-board-messages.js"; import { assertAgentRunWriteAllowed } from "../agent-run-cancellation.js"; @@ -10383,6 +10384,7 @@ export function issueService(db: Db) { await notifyTerminalWorkspaceWork(tx, values.executionWorkspaceId ? `workspace:${values.executionWorkspaceId}` : values.parentId ? `issue:${values.parentId}` : undefined); } + if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const [issue] = await tx.insert(issues).values(values).returning(); await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId }); if (issue.visibility === "private" && !inheritedPrivateSubtree && issueData.createdByAgentId) { @@ -11205,6 +11207,10 @@ export function issueService(db: Db) { || issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); } + if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt && + (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined))) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + } const updated = await tx .update(issues) .set(patch) diff --git a/server/src/services/routines-formatter-cache.test.ts b/server/src/services/routines-formatter-cache.test.ts index daf9064a3f7..93f674d0948 100644 --- a/server/src/services/routines-formatter-cache.test.ts +++ b/server/src/services/routines-formatter-cache.test.ts @@ -1,11 +1,25 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { nextCronTickInTimeZone } from "./routines.js"; +import * as cron from "./cron.js"; describe("nextCronTickInTimeZone formatter caching", () => { afterEach(() => { vi.restoreAllMocks(); }); + it("parses a sparse schedule once instead of once per searched minute", () => { + const parse = cron.parseCron; + let parses = 0; + vi.spyOn(cron, "parseCron").mockImplementation(expression => { + parses++; + if (parses > 1) throw new Error("Repeated cron parsing blocks unrelated scheduled work"); + return parse(expression); + }); + const next = nextCronTickInTimeZone("0 12 1 1 *", "UTC", new Date("2020-01-01T12:00:00Z")); + expect(next?.toISOString()).toBe("2021-01-01T12:00:00.000Z"); + expect(parses).toBe(1); + }); + it("computes the next occurrence of a sparse monthly cron correctly", () => { const next = nextCronTickInTimeZone( "30 6 1 * *", diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index 98bffd4f184..642c70f939c 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { verifyAppWebhook } from "./app-webhook.js"; import crypto from "node:crypto"; import { verifyFirefliesWebhook } from "./fireflies-webhook.js"; @@ -253,8 +254,7 @@ function getZonedMinuteParts(date: Date, timeZone: string) { }; } -function matchesCronMinute(expression: string, timeZone: string, date: Date) { - const cron = parseCron(expression); +function matchesCronMinute(cron: ReturnType, timeZone: string, date: Date) { const parts = getZonedMinuteParts(date, timeZone); return ( cron.minutes.includes(parts.minute) && @@ -268,16 +268,18 @@ function matchesCronMinute(expression: string, timeZone: string, date: Date) { export function nextCronTickInTimeZone(expression: string, timeZone: string, after: Date) { const trimmed = expression.trim(); assertTimeZone(timeZone); - const error = validateCron(trimmed); - if (error) { - throw unprocessable(error); + let cron: ReturnType; + try { + cron = parseCron(trimmed); + } catch (error) { + throw unprocessable(error instanceof Error ? error.message : String(error)); } const cursor = floorToMinute(after); cursor.setUTCMinutes(cursor.getUTCMinutes() + 1); const limit = 366 * 24 * 60 * 5; for (let i = 0; i < limit; i += 1) { - if (matchesCronMinute(trimmed, timeZone, cursor)) { + if (matchesCronMinute(cron, timeZone, cursor)) { return new Date(cursor.getTime()); } cursor.setUTCMinutes(cursor.getUTCMinutes() + 1); @@ -2230,6 +2232,7 @@ export function routineService( } const createdRoutine = await db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await notifyDeliveryWork(txDb, DELIVERY_QUEUES.routineSchedules); const [created] = await txDb .insert(routines) .values({ @@ -2334,6 +2337,7 @@ export function routineService( } const updatedRoutine = await db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await notifyDeliveryWork(txDb, DELIVERY_QUEUES.routineSchedules); await tx.execute(sql`select id from ${routines} where ${routines.id} = ${id} for update`); const locked = await txDb .select() @@ -2488,6 +2492,7 @@ export function routineService( const { trigger, revision } = await db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await notifyDeliveryWork(txDb, DELIVERY_QUEUES.routineSchedules); await tx.execute(sql`select id from ${routines} where ${routines.id} = ${routine.id} for update`); const [createdTrigger] = await txDb .insert(routineTriggers) @@ -2562,6 +2567,7 @@ export function routineService( const result = await db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await notifyDeliveryWork(txDb, DELIVERY_QUEUES.routineSchedules); await tx.execute(sql`select id from ${routines} where ${routines.id} = ${existing.routineId} for update`); const [updated] = await txDb .update(routineTriggers) @@ -2601,6 +2607,7 @@ export function routineService( if (!existing) return { deleted: false, revision: null }; const result = await db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await notifyDeliveryWork(txDb, DELIVERY_QUEUES.routineSchedules); await tx.execute(sql`select id from ${routines} where ${routines.id} = ${existing.routineId} for update`); await txDb.delete(routineTriggers).where(eq(routineTriggers.id, id)); const routine = await txDb @@ -2719,6 +2726,7 @@ export function routineService( const result = await db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await notifyDeliveryWork(txDb, DELIVERY_QUEUES.routineSchedules); await tx.execute(sql`select id from ${routines} where ${routines.id} = ${existingRoutine.id} for update`); const locked = await txDb .select() diff --git a/server/src/services/scheduled-work.ts b/server/src/services/scheduled-work.ts new file mode 100644 index 00000000000..e41411b2765 --- /dev/null +++ b/server/src/services/scheduled-work.ts @@ -0,0 +1,45 @@ +import type { Db } from "@paperclipai/db"; +import { sql, type SQL } from "drizzle-orm"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, type DeliveryQueue } from "./delivery-work-notifications.js"; + +export function registerScheduledWork( + coordinator: Pick, "register">, + db: Pick, + options: { + retryMs: number; + tickTimers: () => Promise; + tickRoutines: () => Promise; + }, +) { + function register(queue: DeliveryQueue, work: () => Promise, query: (now: Date) => SQL) { + let next: number | null = null; + return coordinator.register(queue, { + retryMs: options.retryMs, + run: async () => { + await work(); + const [row] = await db.execute<{ next_at: Date | string | null }>(query(new Date())); + if (!row || row.next_at === undefined) throw new Error("Scheduled deadline query returned no result"); + next = row.next_at === null ? null + : row.next_at instanceof Date ? row.next_at.getTime() : Date.parse(row.next_at); + if (next !== null && !Number.isFinite(next)) throw new Error("Scheduled deadline is invalid"); + }, + hasPending: async () => false, + nextRunAt: () => next, + }); + } + + const routines = register(DELIVERY_QUEUES.routineSchedules, options.tickRoutines, now => sql` + SELECT min(t.next_run_at) AS next_at + FROM routine_triggers t JOIN routines r ON r.id = t.routine_id AND r.company_id = t.company_id + JOIN companies c ON c.id = r.company_id + WHERE t.kind = 'schedule' AND t.enabled AND NOT t.archived + AND r.status = 'active' AND c.status = 'active' AND t.next_run_at > ${now.toISOString()}::timestamptz`); + const timers = register(DELIVERY_QUEUES.issueMonitors, options.tickTimers, now => sql` + SELECT min(i.monitor_next_check_at) AS next_at + FROM issues i JOIN companies c ON c.id = i.company_id + WHERE c.status = 'active' AND i.status IN ('in_progress', 'in_review') + AND i.assignee_agent_id IS NOT NULL AND i.assignee_user_id IS NULL + AND i.monitor_next_check_at > ${now.toISOString()}::timestamptz`); + return { routines, timers }; +}