From de7289a3facc43301490bb7320c7a062b4f49ea3 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 26 Sep 2026 19:23:56 -0700 Subject: [PATCH] fix(cron): bound tick cost and make every accepted schedule fire Each expression is parsed once into allowed-value sets, capped at 256 characters and 64 clauses. The next-fire search skips non-matching days and hours, so it costs a few steps per day and covers eight years, long enough for Feb 29 across 2100. The ticker asks whether any minute after lastFiredAt (or createdAt) and at or before now matches, searching only the last window, so a Feb 29 schedule fires and an old anchor costs no more. POST rejects an expression that can never fire; the ticker stops such a row with invalid_expression instead of re-scanning it. `5/2` now steps to the field maximum. DOM/DOW OR semantics are unchanged. Closes CL-9447 --- README.md | 32 ++--- e2e/routes.test.ts | 31 +++++ e2e/ticker.test.ts | 67 ++++++++++ src/cron.test.ts | 105 ++++++++++++++++ src/cron.ts | 306 +++++++++++++++++++++++++++------------------ src/routes.ts | 4 +- src/ticker.ts | 48 ++++--- 7 files changed, 438 insertions(+), 155 deletions(-) diff --git a/README.md b/README.md index 601a513..def4c9d 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A Corbits hub module that wakes Interchange agents on five-field UTC cron schedules, mounted as `@intx/hub-api` routes on the hub (Interchange's multi-tenant control plane) and stored in its Postgres. A ticker mails each due schedule to its agent's current run, the running instance of the agent's workflow definition. -Schedules support create, list and delete. There is no edit: delete and recreate. +Schedules support create, list and delete. There is no edit: delete and recreate. An expression must be able to fire (no Feb 30), at most 256 characters and 64 comma clauses. ## Why @corbits/cron? @@ -59,15 +59,15 @@ Paths are relative to where the host mounts the sub-app. Returns `{ start(), stop() }`. -| `opts` | Type | Purpose | -| ------------------- | -------------------------------------- | ---------------------------------------------------------------------------------- | -| `db` | `CronDb` | The hub's drizzle handle. | -| `deliver` | `DeliverCronMail` | Sends one due schedule's mail. `createRunTriggerCronDeliver(deliverer)` builds it. | -| `intervalMs` | `number` (optional) | Poll period. Defaults to 60 000. | -| `onTickError` | `(error) => void` (optional) | A tick failed before delivering, such as a lost connection. The next tick retries. | -| `onDeliveryError` | `(error, schedule) => void` (optional) | One delivery failed. | -| `onScheduleWaiting` | `(schedule) => void` (optional) | A schedule started waiting for its agent's next run. | -| `onScheduleStopped` | `(schedule) => void` (optional) | A schedule stopped because its agent was deleted. | +| `opts` | Type | Purpose | +| ------------------- | -------------------------------------- | ------------------------------------------------------------------------------------------------------- | +| `db` | `CronDb` | The hub's drizzle handle. | +| `deliver` | `DeliverCronMail` | Sends one due schedule's mail. `createRunTriggerCronDeliver(deliverer)` builds it. | +| `intervalMs` | `number` (optional) | Poll period. Defaults to 60 000. | +| `onTickError` | `(error) => void` (optional) | A tick failed before delivering, such as a lost connection. The next tick retries. | +| `onDeliveryError` | `(error, schedule) => void` (optional) | One delivery failed. | +| `onScheduleWaiting` | `(schedule) => void` (optional) | A schedule started waiting for its agent's next run. | +| `onScheduleStopped` | `(schedule) => void` (optional) | A schedule stopped: `reason` is `agent_deleted`, or `invalid_expression` for a row that can never fire. | A deliverer throws an error carrying `RUN_GRANTS_NOT_ROUTABLE` or `RUN_MAIL_NOT_ROUTABLE` (checked with `isUnroutableRunTrigger`) when a run's address is dead. If that run's sidecar is already released, the ticker marks the run failed so the schedule waits for the agent's next run. @@ -77,12 +77,12 @@ From `@corbits/cron/migrations`. Takes the same arguments as `@intx/db`'s `runMi ### Other exports -| Export | Use | -| ---------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------- | -| `isValidCronExpression(expression)` | The check `POST /` runs. | -| `nextCronFireAfter(expression, after)` | The next matching UTC minute after `after`. | -| `isUnroutableRunTrigger(error)`, `RUN_GRANTS_NOT_ROUTABLE`, `RUN_MAIL_NOT_ROUTABLE` | The dead-address error contract a deliverer follows. | -| `CronRoutesDeps`, `CreateCronTickerOpts`, `CronTicker`, `CronDb`, `DeliverCronMail`, `RunTriggerDeliverer`, `UnroutableRunTrigger` | Types for the above. | +| Export | Use | +| ---------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------- | +| `isValidCronExpression(expression)` | Syntax, range and cap check. `POST /` also requires a possible fire. | +| `nextCronFireAfter(expression, after)` | The next matching UTC minute after `after`. | +| `isUnroutableRunTrigger(error)`, `RUN_GRANTS_NOT_ROUTABLE`, `RUN_MAIL_NOT_ROUTABLE` | The dead-address error contract a deliverer follows. | +| `CronRoutesDeps`, `CreateCronTickerOpts`, `CronTicker`, `CronDb`, `DeliverCronMail`, `RunTriggerDeliverer`, `UnroutableRunTrigger` | Types for the above. | ## Using with Interchange diff --git a/e2e/routes.test.ts b/e2e/routes.test.ts index bc65bd8..8dbb9f3 100644 --- a/e2e/routes.test.ts +++ b/e2e/routes.test.ts @@ -78,6 +78,37 @@ describeIfDb("createCronRoutes", () => { } }); + test("an expression that can never fire or is over the caps is rejected", async () => { + const { db, close } = createDB(requireDatabase().config); + try { + const tenantId = `tnt_cron_expr_${randomUUID().slice(0, 8)}`; + await seedTenant(db, tenantId); + await seedDeployment(db, tenantId, "agent-expr-source"); + const app = cronRoutesApp(db, tenantId, allowAll); + + const huge = Array.from({ length: 5000 }, () => "1").join(","); + for (const expression of ["0 0 31 2 *", `${huge} * * * *`]) { + const response = await post(app, { + expression, + definitionName: "agent-expr-source", + subject: "s", + body: "b", + }); + expect(response.status).toBe(400); + expect(await response.json()).toEqual({ error: "invalid_expression" }); + } + const leap = await post(app, { + expression: "0 0 29 2 *", + definitionName: "agent-expr-source", + subject: "s", + body: "b", + }); + expect(leap.status).toBe(201); + } finally { + await close(); + } + }); + test("each route is gated by the host's requireGrant", async () => { const { db, close } = createDB(requireDatabase().config); try { diff --git a/e2e/ticker.test.ts b/e2e/ticker.test.ts index 5e73265..70af1c4 100644 --- a/e2e/ticker.test.ts +++ b/e2e/ticker.test.ts @@ -228,6 +228,73 @@ describeIfDb("createCronTicker", () => { } }); + test("a long-gap schedule fires; one that can never fire stops, fast", async () => { + const { db, close } = createDB(requireDatabase().config); + try { + const tenantId = `tnt_cron_expr_${randomUUID().slice(0, 8)}`; + await seedTenant(db, tenantId); + await seedDeployment(db, tenantId, "agent-expr-source"); + + const suffix = randomUUID().slice(0, 8); + const leapId = `sched_leap_${suffix}`; + const feb31Id = `sched_feb31_${suffix}`; + const hugeId = `sched_huge_${suffix}`; + const huge = Array.from({ length: 5000 }, () => "1").join(","); + const row = (id: string, expression: string, createdAt: Date) => ({ + id, + tenantId, + expression, + definitionName: "agent-expr-source", + subject: id, + body: "b", + createdAt, + }); + // Feb 29 2024 is more than a year after this row was saved. + await db + .insert(cronScheduleTable) + .values([ + row(leapId, "0 0 29 2 *", new Date("2021-03-01T00:00:00Z")), + row(feb31Id, "0 0 31 2 *", new Date(Date.now() - 2 * 60_000)), + row(hugeId, `${huge} 0 31 2 *`, new Date(Date.now() - 2 * 60_000)), + ]); + + const delivered: string[] = []; + const stopped: Array<{ id: string; reason: string }> = []; + const ticker = createCronTicker({ + db, + intervalMs: 20, + deliver: (message) => { + delivered.push(message.subject); + }, + onScheduleStopped: (schedule) => + stopped.push({ id: schedule.id, reason: schedule.reason }), + }); + const started = performance.now(); + ticker.start(); + // Polling, not a fixed sleep: tick latency follows DB load. + for ( + let i = 0; + i < 250 && (delivered.length === 0 || stopped.length < 2); + i++ + ) { + await new Promise((resolve) => setTimeout(resolve, 20)); + } + const elapsed = performance.now() - started; + ticker.stop(); + // Let an in-flight tick settle before close() ends the client. + await new Promise((resolve) => setTimeout(resolve, 300)); + + expect(delivered).toEqual([leapId]); + expect(stopped.sort((a, b) => a.id.localeCompare(b.id))).toEqual([ + { id: feb31Id, reason: "invalid_expression" }, + { id: hugeId, reason: "invalid_expression" }, + ]); + expect(elapsed).toBeLessThan(2_000); + } finally { + await close(); + } + }); + test("two concurrent tickers deliver a due row exactly once", async () => { const { db: dbA, close: closeA } = createDB(requireDatabase().config); const { db: dbB, close: closeB } = createDB(requireDatabase().config); diff --git a/src/cron.test.ts b/src/cron.test.ts index f28228b..ea896aa 100644 --- a/src/cron.test.ts +++ b/src/cron.test.ts @@ -4,7 +4,11 @@ import { describe, expect, test } from "bun:test"; import { cronExpressionCanFire, + cronIsDue, cronMatchesMinute, + MAX_CRON_CLAUSES, + MAX_CRON_EXPRESSION_LENGTH, + parseCronExpression, isValidCronExpression, isValidTimeZone, nextCronFireAfter, @@ -268,3 +272,104 @@ describe("timezone matching and DST", () => { expect(zonedParts(next, "America/Los_Angeles").hour).toBe(9); }); }); + +describe("step without a range end runs to the field maximum", () => { + test("5/2 on minutes is 5,7,…,59", () => { + const at = (minute: number) => new Date(Date.UTC(2026, 0, 1, 0, minute)); + expect(cronMatchesMinute("5/2 * * * *", at(5))).toBe(true); + expect(cronMatchesMinute("5/2 * * * *", at(7))).toBe(true); + expect(cronMatchesMinute("5/2 * * * *", at(59))).toBe(true); + expect(cronMatchesMinute("5/2 * * * *", at(6))).toBe(false); + expect(cronMatchesMinute("5/2 * * * *", at(4))).toBe(false); + }); +}); + +describe("limits", () => { + test("rejects an expression over the length cap", () => { + const long = `${"0,".repeat(MAX_CRON_EXPRESSION_LENGTH)}0 * * * *`; + expect(isValidCronExpression(long)).toBe(false); + }); + + test("rejects an expression over the clause cap", () => { + const clauses = Array.from({ length: MAX_CRON_CLAUSES }, () => "1").join( + ",", + ); + expect(isValidCronExpression(`${clauses} * * * *`)).toBe(false); + }); + + test("never-firing expressions cannot fire; Feb 29 can", () => { + expect(cronExpressionCanFire("0 0 31 2 *")).toBe(false); + expect(cronExpressionCanFire("0 0 30 2 *")).toBe(false); + expect(cronExpressionCanFire("0 0 31 4,6,9,11 *")).toBe(false); + expect(cronExpressionCanFire("0 0 29 2 *")).toBe(true); + expect(cronExpressionCanFire("0 0 31 2 5")).toBe(true); + }); +}); + +describe("cronIsDue", () => { + const parsed = (expression: string) => { + const result = parseCronExpression(expression); + if (result === undefined) throw new Error(`unparseable: ${expression}`); + return result; + }; + + test("a Feb 29 schedule created years earlier fires on Feb 29", () => { + const after = new Date("2025-03-01T00:00:00Z"); + expect( + cronIsDue(parsed("0 0 29 2 *"), after, new Date("2028-02-28T23:59:00Z")), + ).toBe(false); + expect( + cronIsDue(parsed("0 0 29 2 *"), after, new Date("2028-02-29T00:00:30Z")), + ).toBe(true); + }); + + test("Feb 29 across the 2100 non-leap gap", () => { + expect( + nextCronFireAfter("0 0 29 2 *", new Date("2096-03-01T00:00:00Z")), + ).toEqual(new Date("2104-02-29T00:00:00Z")); + }); + + test("keeps Vixie OR when both day fields are restricted", () => { + // 2026-01-16 is a Friday, not the 13th. + expect( + cronIsDue( + parsed("0 0 13 * 5"), + new Date("2026-01-15T00:00:00Z"), + new Date("2026-01-16T00:00:00Z"), + ), + ).toBe(true); + }); + + test("a row last fired decades ago costs a bounded scan", () => { + const start = performance.now(); + expect( + cronIsDue( + parsed("0 0 29 2 *"), + new Date("1970-01-01T00:00:00Z"), + new Date("2027-12-31T00:00:00Z"), + ), + ).toBe(true); + expect(performance.now() - start).toBeLessThan(100); + }); +}); + +describe("cost is bounded", () => { + test("a 5,000-clause expression is rejected without scanning", () => { + const big = Array.from({ length: 5000 }, () => "1").join(","); + const start = performance.now(); + expect(parseCronExpression(`${big} 0 31 2 *`)).toBeUndefined(); + expect(cronExpressionCanFire(`${big} 0 31 2 *`)).toBe(false); + expect(performance.now() - start).toBeLessThan(50); + }); + + test("the costliest accepted expression finds its fire quickly", () => { + const minutes = Array.from({ length: MAX_CRON_CLAUSES - 4 }, (_, i) => + String(59 - i), + ).join(","); + const expression = `${minutes} 23 29 2 *`; + expect(isValidCronExpression(expression)).toBe(true); + const start = performance.now(); + nextCronFireAfter(expression, new Date("2096-03-01T00:00:00Z")); + expect(performance.now() - start).toBeLessThan(100); + }); +}); diff --git a/src/cron.ts b/src/cron.ts index 0cc2148..9551d92 100644 --- a/src/cron.ts +++ b/src/cron.ts @@ -22,14 +22,6 @@ export const CRON_FIELD_RANGES: Readonly< dayOfWeek: [0, 7], }; -const CRON_FIELD_ORDER: readonly CronField[] = [ - "minute", - "hour", - "dayOfMonth", - "month", - "dayOfWeek", -]; - type CronClause = { readonly base: "*" | number; readonly rangeEnd?: number; @@ -68,80 +60,106 @@ function clauseInRange( return clause.rangeEnd >= clause.base; } -/** Does `clause` match `value`? Star-with-step steps from `min`, not zero, - * so on a 1-based day-of-month it yields 1,3,5… not 2,4,6…. */ -function clauseMatches( - clause: CronClause, - value: number, - min: number, -): boolean { - if (clause.base === "*") { - return clause.step === undefined ? true : (value - min) % clause.step === 0; - } - if (clause.rangeEnd === undefined && clause.step === undefined) { - return value === clause.base; - } - const upper = clause.rangeEnd ?? clause.base; - if (value < clause.base || value > upper) return false; - if (clause.step === undefined) return true; - return (value - clause.base) % clause.step === 0; +/** Every value `clause` allows in `[min, max]`. A step without a range end + * runs to the field maximum, so `5/2` on minutes is 5,7,…,59 (Vixie). */ +function clauseValues(clause: CronClause, min: number, max: number): number[] { + const start = clause.base === "*" ? min : clause.base; + const end = + clause.base === "*" || + (clause.rangeEnd === undefined && clause.step !== undefined) + ? max + : (clause.rangeEnd ?? clause.base); + const step = clause.step ?? 1; + const values: number[] = []; + for (let value = start; value <= end; value += step) values.push(value); + return values; } -function everyClause( - field: string, - test: (clause: CronClause) => boolean, -): boolean { - const clauses = field.split(",").map(parseCronClause); - if (clauses.length === 0) return false; - return clauses.every((clause) => clause !== undefined && test(clause)); +/** Bounds parse and match cost: a longer expression is rejected, not + * scanned. */ +export const MAX_CRON_EXPRESSION_LENGTH = 256; +export const MAX_CRON_CLAUSES = 64; + +/** A cron expression parsed once into its allowed values per field. */ +export type ParsedCron = { + readonly minutes: readonly boolean[]; + readonly hours: readonly boolean[]; + readonly daysOfMonth: readonly boolean[]; + readonly months: readonly boolean[]; + /** 0–6; an expression's 7 is folded onto 0 (Sunday). */ + readonly daysOfWeek: readonly boolean[]; + /** Vixie: DOM and DOW OR when both are restricted, otherwise AND. */ + readonly dayFieldsOr: boolean; +}; + +// True when the field is restricted (not a bare wildcard or step-1). Vixie +// OR-semantics for DOM/DOW only apply when both fields are restricted. +function isDayFieldRestricted(field: string): boolean { + return field !== "*" && field !== "*/1"; } -function someClause( +function parseField( field: string, - test: (clause: CronClause) => boolean, -): boolean { - return field.split(",").some((raw) => { + [min, max]: readonly [number, number], +): boolean[] | undefined { + const allowed = Array.from({ length: max + 1 }, () => false); + for (const raw of field.split(",")) { const clause = parseCronClause(raw); - return clause !== undefined && test(clause); - }); + if (clause === undefined || !clauseInRange(clause, [min, max])) { + return undefined; + } + for (const value of clauseValues(clause, min, max)) allowed[value] = true; + } + return allowed; } -/** Loud, eager syntax + range validation (never a fire-time surprise like a - * minute of 60). Whether it can ever actually fire is `cronExpressionCanFire`. */ -export function isValidCronExpression(expression: string): boolean { +/** Parses `expression`, or `undefined` when it is malformed, out of range or + * over the length or clause cap. */ +export function parseCronExpression( + expression: string, +): ParsedCron | undefined { + if (expression.length > MAX_CRON_EXPRESSION_LENGTH) return undefined; const fields = expression.trim().split(/\s+/); - if (fields.length !== 5) return false; - return fields.every((field, index) => { - const cronField = CRON_FIELD_ORDER[index]; - if (cronField === undefined) return false; - return everyClause(field, (clause) => - clauseInRange(clause, CRON_FIELD_RANGES[cronField]), - ); - }); -} - -function fieldMatches(field: string, value: number, min: number): boolean { - return someClause(field, (clause) => clauseMatches(clause, value, min)); -} - -/** Day-of-week match with 0/7 both meaning Sunday. */ -function dayOfWeekMatches(field: string, dayOfWeek: number): boolean { - const [min] = CRON_FIELD_RANGES.dayOfWeek; - if (fieldMatches(field, dayOfWeek, min)) return true; - // Date APIs report Sunday as 0; expressions may say 7. - if (dayOfWeek === 0 && fieldMatches(field, 7, min)) return true; - return false; + if (fields.length !== 5) return undefined; + if (fields.join(",").split(",").length > MAX_CRON_CLAUSES) return undefined; + const [minute, hour, dayOfMonth, month, dayOfWeek] = fields as [ + string, + string, + string, + string, + string, + ]; + const minutes = parseField(minute, CRON_FIELD_RANGES.minute); + const hours = parseField(hour, CRON_FIELD_RANGES.hour); + const daysOfMonth = parseField(dayOfMonth, CRON_FIELD_RANGES.dayOfMonth); + const months = parseField(month, CRON_FIELD_RANGES.month); + const weekdays = parseField(dayOfWeek, CRON_FIELD_RANGES.dayOfWeek); + if ( + minutes === undefined || + hours === undefined || + daysOfMonth === undefined || + months === undefined || + weekdays === undefined + ) { + return undefined; + } + return { + minutes, + hours, + daysOfMonth, + months, + daysOfWeek: weekdays + .slice(0, 7) + .map((on, day) => on || (day === 0 && weekdays[7] === true)), + dayFieldsOr: + isDayFieldRestricted(dayOfMonth) && isDayFieldRestricted(dayOfWeek), + }; } -// True when the field is restricted (not a bare wildcard or step-1). Vixie -// OR-semantics for DOM/DOW only apply when both fields are restricted. -function isDayFieldRestricted(field: string): boolean { - const trimmed = field.trim(); - if (trimmed === "*") return false; - // `*/1` is every day — unrestricted in effect — but any other form - // (including `*/2`, `1-5`, `1,15`) is a restriction. - if (trimmed === "*/1") return false; - return true; +/** Loud, eager syntax + range validation (never a fire-time surprise like a + * minute of 60). Whether it can ever actually fire is `cronExpressionCanFire`. */ +export function isValidCronExpression(expression: string): boolean { + return parseCronExpression(expression) !== undefined; } export type ZonedParts = { @@ -223,6 +241,21 @@ export function isValidTimeZone(timeZone: string): boolean { } } +function dayMatches(cron: ParsedCron, parts: ZonedParts): boolean { + if (cron.months[parts.month] !== true) return false; + const dom = cron.daysOfMonth[parts.day] === true; + const dow = cron.daysOfWeek[parts.dayOfWeek] === true; + return cron.dayFieldsOr ? dom || dow : dom && dow; +} + +function parseOrThrow(expression: string): ParsedCron { + const cron = parseCronExpression(expression); + if (cron === undefined) { + throw new Error(`"${expression}" is not a valid cron expression`); + } + return cron; +} + /** True when `expression` matches the wall-clock minute of `at` in * `timeZone`. DOM and DOW OR when both are restricted; otherwise AND. */ export function cronMatchesMinute( @@ -230,37 +263,13 @@ export function cronMatchesMinute( at: Date, timeZone: string = "UTC", ): boolean { - const fields = expression.trim().split(/\s+/); - const [minute, hour, dayOfMonth, month, dayOfWeek] = fields; - if ( - minute === undefined || - hour === undefined || - dayOfMonth === undefined || - month === undefined || - dayOfWeek === undefined - ) { - throw new Error(`"${expression}" is not a 5-field cron expression`); - } - + const cron = parseOrThrow(expression); const parts = zonedParts(at, timeZone); - const timeAndMonth = - fieldMatches(minute, parts.minute, CRON_FIELD_RANGES.minute[0]) && - fieldMatches(hour, parts.hour, CRON_FIELD_RANGES.hour[0]) && - fieldMatches(month, parts.month, CRON_FIELD_RANGES.month[0]); - if (!timeAndMonth) return false; - - const domOk = fieldMatches( - dayOfMonth, - parts.day, - CRON_FIELD_RANGES.dayOfMonth[0], + return ( + dayMatches(cron, parts) && + cron.hours[parts.hour] === true && + cron.minutes[parts.minute] === true ); - const dowOk = dayOfWeekMatches(dayOfWeek, parts.dayOfWeek); - - if (isDayFieldRestricted(dayOfMonth) && isDayFieldRestricted(dayOfWeek)) { - // Vixie: either day-of-month or day-of-week may match. - return domOk || dowOk; - } - return domOk && dowOk; } /** The UTC minute `at` falls in, as a stable, comparable integer key. */ @@ -268,9 +277,39 @@ function minuteKey(at: Date): number { return Math.floor(at.getTime() / 60_000); } -/** Bounds how far ahead `nextCronFireAfter` searches — one leap year is - * enough for any expression that fires at least annually. */ -export const MAX_LOOKAHEAD_MINUTES = 366 * 24 * 60; +/** Bounds every search. Eight years plus a day covers the longest gap any + * valid expression has: Feb 29 across a skipped century leap year. */ +export const MAX_LOOKAHEAD_MINUTES = (8 * 366 + 1) * 24 * 60; + +/** The first matching minute key in `[from, to]`, or `undefined`. Skips + * whole non-matching hours and days, so a scan costs a few steps per day + * however long the expression. Offsets are whole minutes and every skip + * stops at or before the next local hour or midnight, so DST cannot jump + * past a match. */ +function firstMatchBetween( + cron: ParsedCron, + from: number, + to: number, + timeZone: string, +): number | undefined { + let minute = from; + while (minute <= to) { + const parts = zonedParts(new Date(minute * 60_000), timeZone); + const toNextHour = 60 - parts.minute; + if (!dayMatches(cron, parts)) { + minute += Math.max(toNextHour, (23 - parts.hour) * 60 - parts.minute); + continue; + } + if (cron.hours[parts.hour] !== true) { + minute += toNextHour; + continue; + } + const next = cron.minutes.indexOf(true, parts.minute); + if (next === parts.minute) return minute; + minute += next === -1 ? toNextHour : next - parts.minute; + } + return undefined; +} /** The next minute at or after `after` (exclusive) that `expression` * matches. Always returns a UTC instant, even when matching is zoned. */ @@ -280,31 +319,54 @@ export function nextCronFireAfter( timeZone: string = "UTC", ): Date { const start = minuteKey(after) + 1; - for (let minute = start; minute - start <= MAX_LOOKAHEAD_MINUTES; minute++) { - const candidate = new Date(minute * 60_000); - if (cronMatchesMinute(expression, candidate, timeZone)) return candidate; - } - throw new Error( - `"${expression}" has no fire time within the lookahead window` + - (timeZone === "UTC" ? "" : ` in ${timeZone}`), + const next = firstMatchBetween( + parseOrThrow(expression), + start, + start + MAX_LOOKAHEAD_MINUTES, + timeZone, ); + if (next === undefined) { + throw new Error( + `"${expression}" has no fire time within the lookahead window` + + (timeZone === "UTC" ? "" : ` in ${timeZone}`), + ); + } + return new Date(next * 60_000); } -/** True when `expression` has at least one fire inside the lookahead window - * from `from`. Used at save time to reject an impossible expression. */ +/** True when a UTC minute after `after` and at or before `now` matches. + * Only the last lookahead window before `now` is searched: a valid + * expression fires inside every such window, so an old `after` never + * makes the scan longer or hides a fire. */ +export function cronIsDue(cron: ParsedCron, after: Date, now: Date): boolean { + const to = minuteKey(now); + const from = Math.max(minuteKey(after) + 1, to - MAX_LOOKAHEAD_MINUTES); + return firstMatchBetween(cron, from, to, "UTC") !== undefined; +} + +const DAYS_IN_MONTH = [0, 31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]; + +/** Arithmetic, no scan. Hours and minutes always have a value, and a + * restricted DOW fires every month, so only a DOM-only day list can miss + * every chosen month (Feb 30, Apr 31). Feb 29 counts: leap years come. */ +export function parsedCronCanFire(cron: ParsedCron): boolean { + return cron.months.some((inMonth, month) => { + if (!inMonth) return false; + if (cron.dayFieldsOr) return true; + const dayFits = cron.daysOfMonth.some( + (on, day) => on && day <= (DAYS_IN_MONTH[month] ?? 0), + ); + return dayFits && cron.daysOfWeek.some(Boolean); + }); +} + +/** True when `expression` is valid and some minute ever matches it. Used + * at save time to reject an impossible expression. */ export function cronExpressionCanFire( expression: string, timeZone: string = "UTC", - from: Date = new Date(0), ): boolean { - if (!isValidCronExpression(expression)) return false; if (!isValidTimeZone(timeZone)) return false; - try { - nextCronFireAfter(expression, from, timeZone); - return true; - } catch { - // report-error-ignore: next-fire throws on impossible expressions; - // that is the false signal, not an operational failure. - return false; - } + const cron = parseCronExpression(expression); + return cron !== undefined && parsedCronCanFire(cron); } diff --git a/src/routes.ts b/src/routes.ts index 2abfc79..b1ede82 100644 --- a/src/routes.ts +++ b/src/routes.ts @@ -7,7 +7,7 @@ import { eq, and } from "drizzle-orm"; import { Hono } from "hono"; import { idResource, type RequireGrant, type TenantEnv } from "@intx/hub-api"; -import { isValidCronExpression } from "./cron.js"; +import { cronExpressionCanFire } from "./cron.js"; import { definitionExists } from "./deployment.js"; import { cronScheduleTable } from "./schema.js"; import type { CronDb } from "./ticker.js"; @@ -46,7 +46,7 @@ export function createCronRoutes({ if (parsed instanceof type.errors) { return c.json({ error: "invalid_body", detail: parsed.summary }, 400); } - if (!isValidCronExpression(parsed.expression)) { + if (!cronExpressionCanFire(parsed.expression)) { return c.json({ error: "invalid_expression" }, 400); } // Only a name no agent carries is a dead row from birth: an agent that is diff --git a/src/ticker.ts b/src/ticker.ts index 428e06e..3f30a1a 100644 --- a/src/ticker.ts +++ b/src/ticker.ts @@ -3,7 +3,7 @@ import { eq, isNull } from "drizzle-orm"; import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; -import { nextCronFireAfter } from "./cron.js"; +import { cronIsDue, parseCronExpression, parsedCronCanFire } from "./cron.js"; import { definitionExists, failStaleAnchorRun, @@ -42,8 +42,9 @@ export type CreateCronTickerOpts< /** Told about a tick that failed before delivering, such as a lost DB * connection. The next tick retries. */ onTickError?: (error: unknown) => void; - /** Told once when a schedule stops because the agent it targets was - * deleted. A later redeploy does not resume it. */ + /** Told once when a schedule stops: `agent_deleted` when the agent it + * targets was deleted (a later redeploy does not resume it), or + * `invalid_expression` when its expression can never fire. */ onScheduleStopped?: (schedule: { id: string; tenantId: string; @@ -64,23 +65,25 @@ export type CronTicker = { stop(): void; }; +type Due = "due" | "idle" | "invalid"; + /** A freshly saved schedule is due at its first matching minute after - * creation, not retroactively for every minute since the epoch. */ -function isDue( + * creation, not retroactively for every minute since the epoch. A row that + * can never fire (saved before the create route checked, or over today's + * caps) is "invalid" and stopped rather than re-scanned every tick. */ +function dueness( row: { expression: string; lastFiredAt: Date | null; createdAt: Date }, now: Date, -): boolean { - const after = row.lastFiredAt ?? row.createdAt; - try { - return nextCronFireAfter(row.expression, after) <= now; - } catch { - // report-error-ignore: an expression with no fire in the lookahead - // window is simply never due, not an operational failure. - return false; - } +): Due { + const cron = parseCronExpression(row.expression); + if (cron === undefined || !parsedCronCanFire(cron)) return "invalid"; + return cronIsDue(cron, row.lastFiredAt ?? row.createdAt, now) + ? "due" + : "idle"; } const AGENT_DELETED = "agent_deleted"; +const INVALID_EXPRESSION = "invalid_expression"; const DEFAULT_INTERVAL_MS = 60_000; async function tick>( @@ -114,7 +117,22 @@ async function tick>( .where(isNull(cronScheduleTable.stoppedAt)) .for("update", { skipLocked: true }); - for (const row of candidates.filter((row) => isDue(row, now))) { + for (const row of candidates) { + const due = dueness(row, now); + if (due === "idle") continue; + if (due === "invalid") { + await tx + .update(cronScheduleTable) + .set({ stoppedAt: now, stoppedReason: INVALID_EXPRESSION }) + .where(eq(cronScheduleTable.id, row.id)); + onScheduleStopped({ + id: row.id, + tenantId: row.tenantId, + definitionName: row.definitionName, + reason: INVALID_EXPRESSION, + }); + continue; + } // The run behind a target dies on every restart and redeploy, so the // address is resolved now rather than stored. const deployment = await resolveLiveDeployment(