diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index db9f6accc83..6ab0d3e9e37 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1460,7 +1460,8 @@ 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. During idle drain, expired -deadlines retry at the recovery interval while dispatch is paused. External database writers +deadlines retry at the recovery interval while dispatch is paused; future +deadlines keep their original timer so a canceled drain does not delay them. External database writers rely on the periodic recovery scan. Sparse cron searches reuse one parsed expression and skip impossible UTC dates before localized formatting. diff --git a/server/src/__tests__/delivery-work-coordinator.test.ts b/server/src/__tests__/delivery-work-coordinator.test.ts index 70cb1b0a8c2..b4406a7cd5d 100644 --- a/server/src/__tests__/delivery-work-coordinator.test.ts +++ b/server/src/__tests__/delivery-work-coordinator.test.ts @@ -183,6 +183,24 @@ describe("delivery work coordinator", () => { expect(s.coordinator.nextWakeAt()).toBeNull(); }); + it("preserves a future deadline while reconciling a write during idle drain", async () => { + const s = setup(), t = task(); + const due = Date.now() + 1000; + let next: number | null = due; + await s.coordinator.register(DELIVERY_QUEUES.routineSchedules, { ...t, nextRunAt: () => next }).ready; + s.canRun.mockReturnValue(false); + s.canReconcile.mockReturnValue(true); + await s.enqueue(DELIVERY_QUEUES.routineSchedules); + await vi.advanceTimersByTimeAsync(1); + expect(s.coordinator.nextWakeAt()).toBe(due); + expect(t.run).toHaveBeenCalledTimes(1); + next = null; + s.canRun.mockReturnValue(true); + await vi.advanceTimersByTimeAsync(999); + expect(t.run).toHaveBeenCalledTimes(2); + expect(idleWorkSnapshot().active).toBe(0); + }); + it("does not query during standby and resumes when admission opens", async () => { const s = setup(), t = task(); s.canRun.mockReturnValue(false); await s.coordinator.register(DELIVERY_QUEUES.toolAction, t).ready; diff --git a/server/src/index.ts b/server/src/index.ts index bc27e1a509e..264adcd5589 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1881,9 +1881,11 @@ async function startServerWithDatabaseTeardown( const startHeartbeatScheduler = () => { const scheduled = registerScheduledWork(deliveryWork, db, { retryMs: config.heartbeatSchedulerIntervalMs, - tickTimers: () => heartbeat.tickTimers(new Date()).then(result => { + tickTimers: async () => { + if ((await heartbeat.resolveSchedulingSuppression()).suppressed) return; + const result = await heartbeat.tickTimers(new Date()); 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"); }), diff --git a/server/src/services/delivery-work-coordinator.ts b/server/src/services/delivery-work-coordinator.ts index 1f36e7bca81..c8c9a468bf3 100644 --- a/server/src/services/delivery-work-coordinator.ts +++ b/server/src/services/delivery-work-coordinator.ts @@ -56,7 +56,10 @@ export function createDeliveryWorkCoordinator(input: { if (pending || databaseWorkPending(input.owner, queue)) schedule(task.retryMs); else { finishIntent?.(); finishIntent = undefined; } const next = task.nextRunAt?.(); - if (next != null) schedule(Math.max(input.canRun() ? 0 : task.retryMs, next - Date.now())); + if (next != null) { + const delay = Math.max(0, next - Date.now()); + schedule(delay === 0 && !input.canRun() ? task.retryMs : delay); + } }).catch(error => { schedule(task.retryMs); // Logging must never turn a recoverable sweep failure into an