konnosDashboard
schedule-dispatcher.ts174 lines · main
1import { randomUUID } from 'node:crypto';
2
3import { env } from '../env.js';
4import { log } from '../lib/logger.js';
5import { audit } from '../services/audit.js';
6import { invoke } from '../services/invoke.js';
7import {
8 claimDueSchedules,
9 nextRunAfter,
10 parseCron,
11 recordScheduleResult,
12} from '../services/schedules.js';
13
14/**
15 * Schedule dispatcher. Every TICK_MS the worker:
16 *
17 * 1. claims up to BATCH_SIZE rows whose next_run_at <= now() and that
18 * are enabled + not soft-deleted (SELECT, no row lock — claim is
19 * optimistic).
20 * 2. for each row, computes the next next_run_at locally from the cron
21 * expression and fires invoke() against the deployed function.
22 * 3. records the run outcome with an UPDATE guarded by the pre-claim
23 * next_run_at. Two concurrent dispatchers can't double-fire the
24 * same row — the second loses the race and skips silently.
25 *
26 * Failure handling: invoke errors are caught, recorded as `error` with a
27 * truncated message, and the schedule still advances. Catastrophic worker
28 * crashes simply mean a delayed run on next dispatcher tick — schedules
29 * never get stuck (next_run_at advances on every claim attempt).
30 */
31
32const TICK_MS = 60_000;
33const BATCH_SIZE = 100;
34const MAX_ERROR_LEN = 500;
35
36let timer: ReturnType<typeof setInterval> | null = null;
37let inflight = false;
38
39async function tick(): Promise<void> {
40 if (inflight) {
41 log.warn('schedule_dispatcher_tick_skipped_inflight');
42 return;
43 }
44 inflight = true;
45 const now = new Date();
46 try {
47 const due = await claimDueSchedules(now, BATCH_SIZE);
48 if (due.length === 0) return;
49 log.info('schedule_dispatcher_tick', { dueCount: due.length });
50
51 // Run all due rows in parallel. invoke() is already async + bounded
52 // by the runtime's own pool; we don't add another layer here.
53 await Promise.all(due.map((row) => fireOne(row, now)));
54 } catch (err) {
55 log.error('schedule_dispatcher_tick_failed', {
56 message: err instanceof Error ? err.message : String(err),
57 });
58 } finally {
59 inflight = false;
60 }
61}
62
63async function fireOne(
64 row: Awaited<ReturnType<typeof claimDueSchedules>>[number],
65 now: Date,
66): Promise<void> {
67 const claimedNextRunAt = row.nextRunAt;
68 let newNextRunAt: Date;
69 try {
70 newNextRunAt = nextRunAfter(parseCron(row.cronExpression), now);
71 } catch (err) {
72 // The expression must have been valid when written (service validates
73 // on write), so this is recoverable only by disabling the row. Push
74 // next_run_at forward 1h to back-off and record the parse failure.
75 const errorMessage = err instanceof Error ? err.message : 'cron parse failed';
76 await recordScheduleResult({
77 scheduleId: row.id,
78 claimedNextRunAt,
79 newNextRunAt: new Date(now.getTime() + 60 * 60_000),
80 ranAt: now,
81 status: 'skipped',
82 errorMessage: errorMessage.slice(0, MAX_ERROR_LEN),
83 });
84 return;
85 }
86
87 const requestId = `sched_${randomUUID()}`;
88 let status: 'ok' | 'error' = 'ok';
89 let errorMessage: string | undefined;
90
91 try {
92 const result = await invoke({
93 projectId: row.projectId,
94 functionName: row.functionName,
95 args: row.args,
96 requestId,
97 // Schedule-triggered invocations have no human session — auth is
98 // null and the runtime treats this as a system call. The function
99 // can identify itself as scheduled via ctx.invoker.kind === 'schedule'
100 // once the runtime surfaces that (phase 3 work).
101 auth: null,
102 });
103 if (!result.ok) {
104 status = 'error';
105 errorMessage = `${result.code}: ${result.message}`.slice(0, MAX_ERROR_LEN);
106 }
107 } catch (err) {
108 status = 'error';
109 errorMessage = (err instanceof Error ? err.message : String(err)).slice(0, MAX_ERROR_LEN);
110 }
111
112 const applied = await recordScheduleResult({
113 scheduleId: row.id,
114 claimedNextRunAt,
115 newNextRunAt,
116 ranAt: now,
117 status,
118 errorMessage,
119 });
120
121 if (!applied) {
122 // Lost the race to another dispatcher (or the row was updated by a
123 // user mid-flight). The other instance owns the outcome record; we
124 // still attempted the invoke, but that's idempotent at the customer's
125 // function layer — they own correctness for their own logic.
126 log.info('schedule_dispatcher_record_lost_race', { scheduleId: row.id });
127 return;
128 }
129
130 // Audit log only on outcome — not on the "scheduled" intent (that's
131 // implicit by the schedule row itself, which is in audit via
132 // schedule.create). Keeps the audit table from doubling in size per
133 // every minute a `* * * * *` schedule exists.
134 await audit({
135 actorId: null,
136 projectId: row.projectId,
137 action: status === 'ok' ? 'schedule.run.ok' : 'schedule.run.error',
138 ipHash: null,
139 userAgent: null,
140 metadata: {
141 scheduleId: row.id,
142 requestId,
143 functionName: row.functionName,
144 ...(errorMessage ? { error: errorMessage } : {}),
145 },
146 });
147}
148
149export function startScheduleDispatcher(): void {
150 if (timer) return;
151 if (!env.BRIVEN_DATABASE_URL) {
152 log.warn('schedule_dispatcher_skipped_no_db');
153 return;
154 }
155 // Delay 45s after boot so migrations finish and the usage aggregator
156 // isn't competing for the first DB connection burst.
157 setTimeout(() => {
158 void tick();
159 timer = setInterval(() => {
160 void tick();
161 }, TICK_MS);
162 }, 45_000).unref?.();
163 log.info('schedule_dispatcher_armed', { tickMs: TICK_MS, batch: BATCH_SIZE });
164}
165
166export function stopScheduleDispatcher(): void {
167 if (timer) {
168 clearInterval(timer);
169 timer = null;
170 }
171}
172
173// Exported for tests + the admin dashboard "run now" path (Phase 3).
174export const _internals = { tick, fireOne };