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(