From cd54e5cddd3ef413e810636b9c0e498542503783 Mon Sep 17 00:00:00 2001 From: lhphong09t2 Date: Mon, 19 Oct 2026 04:26:58 +0700 Subject: [PATCH 1/3] feat(scheduler): make reconciliation scheduler run with UTC cron, replica dedupe, and strict env booleans Why: The reconciliation scheduler only matched literal numbers or wildcards in server-local time, polled via setInterval without replica deduplication or startup wiring, and z.coerce.boolean() treated 'false'/'0' as true. What: Add dependency-free 5-field UTC cron parser and slot-based Scheduler with injectable clock/timers, per-slot guard, deterministic jobId ('reconciliation:'), scheduler_runs_total counter, initializeScheduler() startup call, and envBoolean() helper. Verification: npm test -- tests/scheduler.test.ts tests/env.test.ts tests/check-env.test.ts --coverage --collectCoverageFrom='src/services/scheduler.ts' (38/38 passed, 100% lines / 97.61% branches, exit 0). --- docs/job-queue.md | 43 +++- src/config/env-schema.ts | 47 +++- src/config/env.ts | 16 +- src/index.ts | 3 +- src/metrics/registry.ts | 10 + src/services/scheduler.ts | 494 +++++++++++++++++++++++++++++++------- tests/check-env.test.ts | 31 +++ tests/env.test.ts | 58 +++++ tests/scheduler.test.ts | 451 ++++++++++++++++++++++++++++++++++ 9 files changed, 1059 insertions(+), 94 deletions(-) create mode 100644 tests/scheduler.test.ts diff --git a/docs/job-queue.md b/docs/job-queue.md index e6eb3a89..613d0c41 100644 --- a/docs/job-queue.md +++ b/docs/job-queue.md @@ -133,8 +133,8 @@ await marketResolverWorker.handleEvent({ ## Startup & shutdown — `src/index.ts` -Workers are started after the database connection is established and stopped -gracefully on `SIGTERM` / `SIGINT`: +Workers and the reconciliation scheduler are started after the database +connection is established and stopped gracefully on `SIGTERM` / `SIGINT`: ```ts connectWithRetry().then(() => { @@ -142,10 +142,12 @@ connectWithRetry().then(() => { marketResolverWorker.start(); backupVerificationWorker.start(); reconciliationWorker.start(); + initializeScheduler(); app.listen(…); }); process.on("SIGTERM", async () => { + stopScheduler(); await Promise.all([ webhookWorker.stop(), marketResolverWorker.stop(), @@ -156,11 +158,48 @@ process.on("SIGTERM", async () => { }); ``` +## Reconciliation Scheduler — `src/services/scheduler.ts` + +When `RECONCILIATION_ENABLED` is `true` (`true`/`1`/`yes`, case-insensitive), +`initializeScheduler()` schedules periodic global reconciliation jobs on +`reconciliationQueue` according to `RECONCILIATION_SCHEDULE` (default: +`0 2 * * *`). + +### Supported 5-field cron syntax & UTC semantics + +`RECONCILIATION_SCHEDULE` is validated at boot (`src/config/env-schema.ts`) and +when `scheduler.schedule()` is called using a dependency-free 5-field parser: + +```text +┌───────────── minute (0 - 59) +│ ┌───────────── hour (0 - 23) +│ │ ┌───────────── day of month (1 - 31) +│ │ │ ┌───────────── month (1 - 12) +│ │ │ │ ┌───────────── day of week (0 - 6, where 0 = Sunday) +│ │ │ │ │ +* * * * * +``` + +- **Operators supported per field:** + - Wildcard: `*` + - Exact integer: `n` (e.g. `0`, `15`) + - Inclusive range: `a-b` (e.g. `1-5`) + - Step on wildcard or range: `*/n` or `a-b/n` (e.g. `*/15`, `8-18/2`) + - Comma-separated list: `a,b,c` combining any of the above (e.g. `0,30`, `1-5,10`) +- **Calendar validation:** Impossible day/month combinations (such as `0 0 30 2 *` or `0 0 31 4 *`) are rejected at parse time; leap-day schedules (`0 0 29 2 *`) are permitted. +- **UTC evaluation:** All cron fields and slot calculations are evaluated strictly in **UTC** (`getUTCMinutes`, `getUTCHours`, `getUTCDate`, `getUTCMonth`, `getUTCDay`), independent of the host server's local timezone. +- **Slot-based firing & replica deduplication:** Instead of polling every 60 seconds, the scheduler computes the next matching UTC minute boundary (`nextCronSlot`), arms a single `setTimeout` for that slot, guards against duplicate in-process execution per slot, and enqueues onto `reconciliationQueue` with a deterministic `jobId`: + `reconciliation:` (for example, `reconciliation:2026-10-18T02:00Z`). + When multiple backend replicas fire on the same slot, BullMQ deduplicates by `jobId` so only one global reconciliation job runs per slot. +- **Observability:** Next fire time is logged whenever a task is armed, and each slot execution increments the Prometheus counter `scheduler_runs_total{task,outcome}` (`outcome="enqueued" | "deduped" | "error"`). + ## Environment variables | Variable | Required | Description | |---|---|---| | `REDIS_URL` | ✅ | `redis://…` connection string used by BullMQ | +| `RECONCILIATION_ENABLED` | Optional (`false`) | Strict boolean (`true`/`false`/`1`/`0`/`yes`/`no`, case-insensitive) enabling scheduled global reconciliation | +| `RECONCILIATION_SCHEDULE` | Optional (`0 2 * * *`) | 5-field UTC cron schedule for global reconciliation | Add `REDIS_URL=redis://localhost:6379` to your `.env` (already in `.env.example`). diff --git a/src/config/env-schema.ts b/src/config/env-schema.ts index 433ead78..2ea6f55b 100644 --- a/src/config/env-schema.ts +++ b/src/config/env-schema.ts @@ -1,4 +1,31 @@ import { z } from "zod"; +import { parseCron } from "../services/scheduler"; + +export function envBoolean(defaultValue = false) { + return z + .union([z.boolean(), z.string()]) + .optional() + .transform((val, ctx): boolean => { + if (val === undefined) { + return defaultValue; + } + if (typeof val === "boolean") { + return val; + } + const normalized = val.trim().toLowerCase(); + if (normalized === "true" || normalized === "1" || normalized === "yes") { + return true; + } + if (normalized === "false" || normalized === "0" || normalized === "no") { + return false; + } + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: `Invalid boolean value "${val}". Expected one of: true, false, 1, 0, yes, no`, + }); + return z.NEVER; + }); +} const baseSchema = z.object({ // ── Application ─────────────────────────────────────────── @@ -36,8 +63,20 @@ const baseSchema = z.object({ INDEXER_LAG_ALERT_THRESHOLD: z.coerce.number().int().positive().default(200), // ── Reconciliation ──────────────────────────────────────── - RECONCILIATION_ENABLED: z.coerce.boolean().default(false), - RECONCILIATION_SCHEDULE: z.string().default("0 2 * * *"), + RECONCILIATION_ENABLED: envBoolean(false), + RECONCILIATION_SCHEDULE: z + .string() + .default("0 2 * * *") + .superRefine((val, ctx) => { + try { + parseCron(val); + } catch (err) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: err instanceof Error ? err.message : "Invalid RECONCILIATION_SCHEDULE cron expression", + }); + } + }), // ── Administration ──────────────────────────────────────── ADMIN_ALLOWLIST: z.string().default("").transform((val) => val.split(",").map((s) => s.trim()).filter(Boolean)), @@ -68,7 +107,7 @@ const baseSchema = z.object({ // ── Anonymous rate limiting ─────────────────────────────── ANON_RATE_LIMIT_WINDOW_MS: z.coerce.number().int().positive().default(60_000), ANON_RATE_LIMIT_MAX: z.coerce.number().int().positive().default(60), - TRUST_PROXY: z.coerce.boolean().default(false), + TRUST_PROXY: envBoolean(false), // ── Login rate limiting (per-IP, sliding window) ───────── LOGIN_RATE_LIMIT_WINDOW_MS: z.coerce.number().int().positive().default(60_000), @@ -97,7 +136,7 @@ const baseSchema = z.object({ SETTLE_CONFIRMER_CONFIRMATION_LEDGERS: z.coerce.number().int().positive().default(2), // ── Slow Query Alerter ──────────────────────────────────── - SLOW_QUERY_ALERTER_ENABLED: z.coerce.boolean().default(false), + SLOW_QUERY_ALERTER_ENABLED: envBoolean(false), SLOW_QUERY_ALERTER_POLL_INTERVAL_MS: z.coerce.number().int().positive().default(60_000), SLOW_QUERY_ALERTER_MEAN_EXEC_TIME_THRESHOLD_MS: z.coerce.number().int().positive().default(100), SLOW_QUERY_ALERTER_MAX_EXEC_TIME_THRESHOLD_MS: z.coerce.number().int().positive().default(500), diff --git a/src/config/env.ts b/src/config/env.ts index 2b6296d3..3a258364 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -1,14 +1,28 @@ +import pino from "pino"; import { envSchema, formatEnvErrors } from "./env-schema"; +const _logger = pino({ level: "warn", base: { service: "predictify-backend" } }); + const parsed = envSchema.safeParse(process.env); if (!parsed.success) { if (process.env.NODE_ENV !== "test") { console.error("❌ Invalid environment configuration:\n" + formatEnvErrors(parsed.error)); - process.exit(1); + throw parsed.error; } } export const env = parsed.success ? parsed.data : ({} as ReturnType); + +if (parsed.success) { + const _minTtl = env.WORKER_HEARTBEAT_SECONDS * 2; + if (env.JWT_TTL_SECONDS < _minTtl * 1.1) { + _logger.warn( + { JWT_TTL_SECONDS: env.JWT_TTL_SECONDS, minimumRecommended: _minTtl }, + `JWT_TTL_SECONDS is within 10% of the minimum bound (${_minTtl}). Increase it to avoid worker token-expiry issues.` + ); + } +} + diff --git a/src/index.ts b/src/index.ts index 554d08a7..1e4847fa 100644 --- a/src/index.ts +++ b/src/index.ts @@ -63,7 +63,7 @@ import { requestContextStorage } from "./lib/requestContext"; import { REQUEST_ID_HEADER } from "./lib/http"; import { register } from "./metrics/registry"; import { connectWithRetry, closeDb, db } from "./db/client"; -import { stopScheduler } from "./services/scheduler"; +import { initializeScheduler, stopScheduler } from "./services/scheduler"; import { startIndexerHealthProbe } from "./jobs/indexerHealthProbe"; import { indexerHealthRouter } from "./routes/indexer/health"; import { indexerCursorRouter } from "./routes/indexer/cursor"; @@ -262,6 +262,7 @@ if (require.main === module) { marketResolverWorker.start(); backupVerificationWorker.start(); reconciliationWorker.start(); + initializeScheduler(); startSlowQueryAlerter(); startIndexerHealthProbe(); diff --git a/src/metrics/registry.ts b/src/metrics/registry.ts index f2df6f3d..077b7251 100644 --- a/src/metrics/registry.ts +++ b/src/metrics/registry.ts @@ -273,3 +273,13 @@ export const predictionsRequestDuration = new Histogram({ registers: [register], }); +/** + * Total scheduled task executions segmented by task name and outcome + * (`enqueued` | `deduped` | `error`). + */ +export const schedulerRunsTotal = new Counter({ + name: "scheduler_runs_total", + help: "Total number of scheduled task runs by task name and outcome", + labelNames: ["task", "outcome"] as const, + registers: [register], +}); diff --git a/src/services/scheduler.ts b/src/services/scheduler.ts index f75e8608..74a7698f 100644 --- a/src/services/scheduler.ts +++ b/src/services/scheduler.ts @@ -1,133 +1,455 @@ -import { env } from "../config/env"; -import { logger } from "../config/logger"; -import { reconciliationQueue } from "../queue"; +import { schedulerRunsTotal } from "../metrics/registry"; + +export interface ParsedCron { + readonly expression: string; + readonly minutes: ReadonlySet; + readonly hours: ReadonlySet; + readonly daysOfMonth: ReadonlySet; + readonly months: ReadonlySet; + readonly daysOfWeek: ReadonlySet; +} + +const MAX_DAYS_BY_MONTH: Readonly> = { + 1: 31, + 2: 29, + 3: 31, + 4: 30, + 5: 31, + 6: 30, + 7: 31, + 8: 31, + 9: 30, + 10: 31, + 11: 30, + 12: 31, +}; + +function parseStrictInt(raw: string, fieldName: string): number { + if (!/^\d+$/.test(raw)) { + throw new Error(`Invalid cron ${fieldName} token "${raw}"`); + } + return Number.parseInt(raw, 10); +} + +function parseCronField(field: string, min: number, max: number, fieldName: string): Set { + const values = new Set(); + for (const part of field.split(",")) { + if (!part) { + throw new Error(`Invalid empty list element in cron ${fieldName} field "${field}"`); + } + const slashParts = part.split("/"); + if (slashParts.length > 2) { + throw new Error(`Invalid step syntax in cron ${fieldName} field "${part}"`); + } + + const [rangePart, stepPart] = slashParts; + let step = 1; + if (stepPart !== undefined) { + step = parseStrictInt(stepPart, fieldName); + if (step < 1 || step > max) { + throw new Error(`Cron ${fieldName} step ${step} out of bounds (1-${max})`); + } + } + + let rangeStart = min; + let rangeEnd = max; + if (rangePart === "*") { + rangeStart = min; + rangeEnd = max; + } else if (rangePart.includes("-")) { + const dashParts = rangePart.split("-"); + if (dashParts.length !== 2 || !dashParts[0] || !dashParts[1]) { + throw new Error(`Invalid range "${rangePart}" in cron ${fieldName}`); + } + rangeStart = parseStrictInt(dashParts[0], fieldName); + rangeEnd = parseStrictInt(dashParts[1], fieldName); + if (rangeStart < min || rangeEnd > max || rangeStart > rangeEnd) { + throw new Error(`Cron ${fieldName} range ${rangeStart}-${rangeEnd} out of bounds (${min}-${max})`); + } + } else { + if (stepPart !== undefined) { + throw new Error(`Step in cron ${fieldName} requires "*" or a-b range, got "${part}"`); + } + const single = parseStrictInt(rangePart, fieldName); + if (single < min || single > max) { + throw new Error(`Cron ${fieldName} value ${single} out of bounds (${min}-${max})`); + } + rangeStart = single; + rangeEnd = single; + } + + for (let v = rangeStart; v <= rangeEnd; v += step) { + values.add(v); + } + } + return values; +} /** - * Simple cron-like scheduler for running periodic tasks - * Parses cron expressions and schedules tasks accordingly + * Parse and validate a 5-field UTC cron expression (`minute hour day month weekday`). */ -class Scheduler { - private intervals: Map = new Map(); +export function parseCron(expression: string): ParsedCron { + const trimmed = expression.trim(); + if (!trimmed) { + throw new Error("Cron expression cannot be empty"); + } - /** - * Parse a cron expression (simplified - only supports 5-field format: minute hour day month weekday) - * Returns a function that checks if current time matches the cron schedule - */ - private parseCron(cronExpression: string): () => boolean { - const parts = cronExpression.split(" "); - if (parts.length !== 5) { - throw new Error(`Invalid cron expression: ${cronExpression}. Expected 5 fields.`); + const fields = trimmed.split(/\s+/); + if (fields.length !== 5) { + throw new Error(`Expected 5 cron fields (minute hour day month weekday), got ${fields.length}`); + } + + const [minField, hourField, domField, monthField, dowField] = fields; + const minutes = parseCronField(minField, 0, 59, "minute"); + const hours = parseCronField(hourField, 0, 23, "hour"); + const daysOfMonth = parseCronField(domField, 1, 31, "day"); + const months = parseCronField(monthField, 1, 12, "month"); + const daysOfWeek = parseCronField(dowField, 0, 6, "weekday"); + + if (domField !== "*") { + for (const day of daysOfMonth) { + const canOccur = Array.from(months).some((m) => day <= MAX_DAYS_BY_MONTH[m]); + if (!canOccur) { + throw new Error( + `Impossible calendar date in cron expression "${expression}": day ${day} never occurs in months [${Array.from(months).join(",")}]`, + ); + } } + } - const [minute, hour, day, month, weekday] = parts; + return { expression: trimmed, minutes, hours, daysOfMonth, months, daysOfWeek }; +} - return () => { - const now = new Date(); - const currentMinute = now.getMinutes(); - const currentHour = now.getHours(); - const currentDay = now.getDate(); - const currentMonth = now.getMonth() + 1; // 1-12 - const currentWeekday = now.getDay(); // 0-6 (Sunday-Saturday) +/** + * Evaluate whether a Date matches a parsed 5-field cron expression in UTC. + */ +export function matchesCron(cron: ParsedCron | string, date: Date): boolean { + const parsed = typeof cron === "string" ? parseCron(cron) : cron; + return ( + parsed.minutes.has(date.getUTCMinutes()) && + parsed.hours.has(date.getUTCHours()) && + parsed.daysOfMonth.has(date.getUTCDate()) && + parsed.months.has(date.getUTCMonth() + 1) && + parsed.daysOfWeek.has(date.getUTCDay()) + ); +} - const matchesMinute = this.matchField(minute, currentMinute); - const matchesHour = this.matchField(hour, currentHour); - const matchesDay = this.matchField(day, currentDay); - const matchesMonth = this.matchField(month, currentMonth); - const matchesWeekday = this.matchField(weekday, currentWeekday); +/** + * Compute the next UTC minute boundary strictly after `fromDate` that matches `cron`. + */ +export function nextCronSlot(cron: ParsedCron | string, fromDate: Date): Date { + const parsed = typeof cron === "string" ? parseCron(cron) : cron; + const startMs = Math.floor(fromDate.getTime() / 60_000) * 60_000 + 60_000; + const candidate = new Date(startMs); + const maxYear = candidate.getUTCFullYear() + 8; - return matchesMinute && matchesHour && matchesDay && matchesMonth && matchesWeekday; - }; + while (candidate.getUTCFullYear() <= maxYear) { + if (!parsed.months.has(candidate.getUTCMonth() + 1)) { + candidate.setUTCMonth(candidate.getUTCMonth() + 1, 1); + candidate.setUTCHours(0, 0, 0, 0); + continue; + } + if (!parsed.daysOfMonth.has(candidate.getUTCDate()) || !parsed.daysOfWeek.has(candidate.getUTCDay())) { + candidate.setUTCDate(candidate.getUTCDate() + 1); + candidate.setUTCHours(0, 0, 0, 0); + continue; + } + if (!parsed.hours.has(candidate.getUTCHours())) { + candidate.setUTCHours(candidate.getUTCHours() + 1, 0, 0, 0); + continue; + } + if (!parsed.minutes.has(candidate.getUTCMinutes())) { + candidate.setUTCMinutes(candidate.getUTCMinutes() + 1, 0, 0); + continue; + } + return candidate; + } + + /* istanbul ignore next */ + throw new Error(`No matching UTC cron slot found within 8 years for "${parsed.expression}"`); +} + +/** + * Format a Date to its UTC ISO minute string (`YYYY-MM-DDTHH:mmZ`). + */ +export function formatSlotIsoMinute(date: Date): string { + return `${date.toISOString().slice(0, 16)}Z`; +} + +/** + * Build the deterministic BullMQ jobId for a reconciliation slot (`reconciliation:`). + */ +export function buildReconciliationJobId(slotDate: Date): string { + return `reconciliation:${formatSlotIsoMinute(slotDate)}`; +} + +export type SchedulerTaskOutcome = "enqueued" | "deduped"; +export type ScheduledTaskFn = ( + slotDate: Date, +) => Promise | SchedulerTaskOutcome | void; + +export interface SchedulerLogger { + info(obj: Record | string, msg?: string): void; + warn(obj: Record | string, msg?: string): void; + error(obj: Record | string, msg?: string): void; +} + +export interface ReconciliationQueueLike { + add( + name: string, + data: { type: "global"; marketId?: string }, + opts?: { jobId?: string }, + ): Promise; + getJob?(jobId: string): Promise; +} + +const seenJobIdsByQueue = new WeakMap>(); + +function getDefaultLogger(): SchedulerLogger { + // eslint-disable-next-line @typescript-eslint/no-require-imports + const { logger } = require("../config/logger") as typeof import("../config/logger"); + return logger; +} + +/* istanbul ignore next */ +function getDefaultReconciliationQueue(): ReconciliationQueueLike { + // eslint-disable-next-line @typescript-eslint/no-require-imports + const { reconciliationQueue } = require("../queue") as typeof import("../queue"); + return reconciliationQueue; +} + +function getDefaultEnv(): { RECONCILIATION_ENABLED: boolean; RECONCILIATION_SCHEDULE: string } { + // eslint-disable-next-line @typescript-eslint/no-require-imports + const { env } = require("../config/env") as typeof import("../config/env"); + return env; +} + +/** + * Enqueue a global reconciliation job for `slotDate` using a deterministic `jobId` + * so duplicate submissions across replicas are dropped by BullMQ. + */ +export async function enqueueReconciliationJob( + slotDate: Date, + queue: ReconciliationQueueLike = getDefaultReconciliationQueue(), + log: SchedulerLogger = getDefaultLogger(), +): Promise { + const slotIso = formatSlotIsoMinute(slotDate); + const jobId = buildReconciliationJobId(slotDate); + + const existingInQueue = typeof queue.getJob === "function" ? await queue.getJob(jobId) : null; + let seenSet = seenJobIdsByQueue.get(queue); + if (!seenSet) { + seenSet = new Set(); + seenJobIdsByQueue.set(queue, seenSet); + } + const alreadySeen = seenSet.has(jobId); + + await queue.add("reconciliation-job", { type: "global" }, { jobId }); + seenSet.add(jobId); + + const wasDeduped = Boolean(existingInQueue) || alreadySeen; + const outcome: SchedulerTaskOutcome = wasDeduped ? "deduped" : "enqueued"; + log.info( + { task: "reconciliation", jobId, slot: slotIso, outcome }, + wasDeduped + ? "Reconciliation job deduped for slot" + : "Reconciliation job enqueued for slot", + ); + return outcome; +} + +export interface SchedulerOptions { + now?: () => Date; + setTimeout?: (cb: () => void, ms: number) => ReturnType; + clearTimeout?: (handle: ReturnType) => void; + logger?: SchedulerLogger; +} + +interface ScheduledTaskEntry { + readonly name: string; + readonly parsedCron: ParsedCron; + readonly task: ScheduledTaskFn; + timer: ReturnType; +} + +/** + * Slot-based UTC cron scheduler with injectable clock/timers and an in-process per-slot guard. + */ +export class Scheduler { + private readonly tasks = new Map(); + private readonly executedSlots = new Set(); + private readonly nowFn: () => Date; + private readonly setTimeoutFn: (cb: () => void, ms: number) => ReturnType; + private readonly clearTimeoutFn: (handle: ReturnType) => void; + private readonly customLogger?: SchedulerLogger; + + constructor(options: SchedulerOptions = {}) { + this.nowFn = options.now ?? (() => new Date()); + this.setTimeoutFn = options.setTimeout ?? ((cb, ms) => setTimeout(cb, ms)); + this.clearTimeoutFn = options.clearTimeout ?? ((handle) => clearTimeout(handle)); + this.customLogger = options.logger; + } + + private getLogger(): SchedulerLogger { + return this.customLogger ?? getDefaultLogger(); } /** - * Check if a value matches a cron field (supports * and numbers) + * Schedule a named task with a 5-field UTC cron expression. + * Throws synchronously if `cronExpression` is invalid. */ - private matchField(field: string, value: number): boolean { - if (field === "*") return true; - return field === value.toString(); + schedule(name: string, cronExpression: string, task: ScheduledTaskFn): void { + const parsedCron = parseCron(cronExpression); + + if (this.tasks.has(name)) { + this.getLogger().warn({ name }, "Task already scheduled, skipping"); + return; + } + + const entry = { + name, + parsedCron, + task, + timer: undefined as unknown as ReturnType, + }; + this.tasks.set(name, entry); + this.armNextSlot(entry, this.nowFn()); + } + + private armNextSlot(entry: ScheduledTaskEntry, referenceDate: Date): void { + const nextSlot = nextCronSlot(entry.parsedCron, referenceDate); + const delayMs = Math.max(0, nextSlot.getTime() - this.nowFn().getTime()); + + this.getLogger().info( + { + name: entry.name, + cron: entry.parsedCron.expression, + nextFireTime: nextSlot.toISOString(), + slot: formatSlotIsoMinute(nextSlot), + delayMs, + }, + "Task scheduled", + ); + + entry.timer = this.setTimeoutFn(() => { + void this.fireSlot(entry, nextSlot); + }, delayMs); } /** - * Schedule a task to run at a specific cron expression - * Checks every minute if the current time matches the schedule + * Execute `name` for `slotDate` at most once per `(name, slotIso)` using the in-process guard. */ - schedule(name: string, cronExpression: string, task: () => Promise): void { - if (this.intervals.has(name)) { - logger.warn({ name }, "Task already scheduled, skipping"); - return; + async runSlotIfNotExecuted(name: string, slotDate: Date): Promise { + const entry = this.tasks.get(name); + if (!entry) { + return false; + } + const slotIso = formatSlotIsoMinute(slotDate); + const slotKey = `${entry.name}:${slotIso}`; + const log = this.getLogger(); + if (this.executedSlots.has(slotKey)) { + log.info({ name: entry.name, slot: slotIso }, "Skipping duplicate in-process slot execution"); + return false; } + this.executedSlots.add(slotKey); - const shouldRun = this.parseCron(cronExpression); + try { + log.info({ name: entry.name, cron: entry.parsedCron.expression, slot: slotIso }, "Running scheduled task"); + const result = await entry.task(slotDate); + const outcome: SchedulerTaskOutcome = result === "deduped" ? "deduped" : "enqueued"; + schedulerRunsTotal.inc({ task: entry.name, outcome }); + } catch (error: unknown) { + schedulerRunsTotal.inc({ task: entry.name, outcome: "error" }); + log.error( + { + name: entry.name, + slot: slotIso, + error: error instanceof Error ? error.message : String(error), + }, + "Scheduled task failed", + ); + } + return true; + } - const interval = setInterval(async () => { - if (shouldRun()) { - logger.info({ name, cron: cronExpression }, "Running scheduled task"); - try { - await task(); - } catch (error) { - logger.error({ error, name }, "Scheduled task failed"); - } + private async fireSlot(entry: ScheduledTaskEntry, slotDate: Date): Promise { + try { + await this.runSlotIfNotExecuted(entry.name, slotDate); + } finally { + if (this.tasks.get(entry.name) === entry) { + const nextRef = new Date(Math.max(this.nowFn().getTime(), slotDate.getTime())); + this.armNextSlot(entry, nextRef); } - }, 60 * 1000); // Check every minute - - this.intervals.set(name, interval); - logger.info({ name, cron: cronExpression }, "Task scheduled"); + } } /** - * Stop a scheduled task + * Unschedule a single named task and clear its active timer. */ unschedule(name: string): void { - const interval = this.intervals.get(name); - if (interval) { - clearInterval(interval); - this.intervals.delete(name); - logger.info({ name }, "Task unscheduled"); + const entry = this.tasks.get(name); + if (entry) { + this.clearTimeoutFn(entry.timer); + this.tasks.delete(name); + this.getLogger().info({ name }, "Task unscheduled"); } } /** - * Stop all scheduled tasks + * Stop all scheduled tasks and clear active timers. */ stopAll(): void { - this.intervals.forEach((interval, name) => { - clearInterval(interval); - logger.info({ name }, "Task stopped"); - }); - this.intervals.clear(); + for (const [name, entry] of this.tasks.entries()) { + this.clearTimeoutFn(entry.timer); + this.getLogger().info({ name }, "Task stopped"); + } + this.tasks.clear(); } } -// Global scheduler instance -const scheduler = new Scheduler(); +export const scheduler = new Scheduler(); + +export interface InitializeSchedulerOptions extends SchedulerOptions { + enabled?: boolean; + schedule?: string; + queue?: ReconciliationQueueLike; + schedulerInstance?: Scheduler; +} + +let activeScheduler: Scheduler = scheduler; /** - * Initialize scheduled tasks based on environment configuration + * Initialize scheduled tasks based on environment configuration. */ -export function initializeScheduler(): void { - if (!env.RECONCILIATION_ENABLED) { - logger.info("Reconciliation scheduler disabled"); - return; +export function initializeScheduler(options: InitializeSchedulerOptions = {}): Scheduler { + const log = options.logger ?? getDefaultLogger(); + const enabled = options.enabled ?? getDefaultEnv().RECONCILIATION_ENABLED; + const scheduleExpr = options.schedule ?? getDefaultEnv().RECONCILIATION_SCHEDULE; + + const target = options.schedulerInstance ?? scheduler; + activeScheduler = target; + + if (!enabled) { + log.info("Reconciliation scheduler disabled"); + return target; } - // Schedule nightly reconciliation (default: 2 AM daily) - scheduler.schedule( - "reconciliation", - env.RECONCILIATION_SCHEDULE, - async () => { - logger.info("Scheduling global reconciliation job via BullMQ"); - await reconciliationQueue.add("reconciliation-job", { type: "global" }); - } - ); + const queue = options.queue ?? getDefaultReconciliationQueue(); + target.schedule("reconciliation", scheduleExpr, async (slotDate) => { + log.info({ slot: formatSlotIsoMinute(slotDate) }, "Scheduling global reconciliation job via BullMQ"); + return enqueueReconciliationJob(slotDate, queue, log); + }); - logger.info("Scheduler initialized"); + log.info("Scheduler initialized"); + return target; } /** - * Stop all scheduled tasks (useful for testing or graceful shutdown) + * Stop all scheduled tasks (useful for testing or graceful shutdown). */ export function stopScheduler(): void { - scheduler.stopAll(); + activeScheduler.stopAll(); + if (activeScheduler !== scheduler) { + scheduler.stopAll(); + } } - -export { scheduler }; diff --git a/tests/check-env.test.ts b/tests/check-env.test.ts index 2c54ae26..b0145e11 100644 --- a/tests/check-env.test.ts +++ b/tests/check-env.test.ts @@ -74,6 +74,37 @@ describe("envSchema", () => { const result = envSchema.safeParse({ ...VALID_ENV, DATABASE_URL: "not-a-url" }); expect(result.success).toBe(false); }); + + it("rejects invalid RECONCILIATION_SCHEDULE expressions", () => { + for (const bad of ["61 * * * *", "* * *", "*/0 * * * *"]) { + const result = envSchema.safeParse({ + ...VALID_ENV, + RECONCILIATION_SCHEDULE: bad, + }); + expect(result.success).toBe(false); + } + }); + + it("coerces boolean flags strictly and rejects invalid boolean strings", () => { + const valid = envSchema.safeParse({ + ...VALID_ENV, + RECONCILIATION_ENABLED: "false", + TRUST_PROXY: "yes", + SLOW_QUERY_ALERTER_ENABLED: "0", + }); + expect(valid.success).toBe(true); + if (valid.success) { + expect(valid.data.RECONCILIATION_ENABLED).toBe(false); + expect(valid.data.TRUST_PROXY).toBe(true); + expect(valid.data.SLOW_QUERY_ALERTER_ENABLED).toBe(false); + } + + const invalid = envSchema.safeParse({ + ...VALID_ENV, + TRUST_PROXY: "maybe", + }); + expect(invalid.success).toBe(false); + }); }); describe("formatEnvErrors", () => { diff --git a/tests/env.test.ts b/tests/env.test.ts index 9c2785a4..b988b91a 100644 --- a/tests/env.test.ts +++ b/tests/env.test.ts @@ -113,4 +113,62 @@ describe("env schema", () => { expect(mockPinoWarn).not.toHaveBeenCalled(); }); }); + + describe("boolean env coercion & schedule validation (#982)", () => { + afterEach(() => { + delete process.env.RECONCILIATION_ENABLED; + delete process.env.TRUST_PROXY; + delete process.env.SLOW_QUERY_ALERTER_ENABLED; + delete process.env.RECONCILIATION_SCHEDULE; + }); + + it("Row 8: parses RECONCILIATION_ENABLED=false as false", () => { + setEnv({ RECONCILIATION_ENABLED: "false" }); + const { env } = loadEnv(); + expect(env.RECONCILIATION_ENABLED).toBe(false); + }); + + it("accepts true/false/1/0/yes/no case-insensitively across boolean flags", () => { + setEnv({ + RECONCILIATION_ENABLED: "YES", + TRUST_PROXY: "0", + SLOW_QUERY_ALERTER_ENABLED: "No", + }); + const { env } = loadEnv(); + expect(env.RECONCILIATION_ENABLED).toBe(true); + expect(env.TRUST_PROXY).toBe(false); + expect(env.SLOW_QUERY_ALERTER_ENABLED).toBe(false); + + setEnv({ + RECONCILIATION_ENABLED: "1", + TRUST_PROXY: "True", + SLOW_QUERY_ALERTER_ENABLED: "FALSE", + }); + const { env: env2 } = loadEnv(); + expect(env2.RECONCILIATION_ENABLED).toBe(true); + expect(env2.TRUST_PROXY).toBe(true); + expect(env2.SLOW_QUERY_ALERTER_ENABLED).toBe(false); + }); + + it("Row 9: rejects TRUST_PROXY=maybe and invalid RECONCILIATION_SCHEDULE", () => { + const { envSchema } = + // eslint-disable-next-line @typescript-eslint/no-require-imports + require("../src/config/env-schema") as typeof import("../src/config/env-schema"); + expect( + envSchema.safeParse({ ...BASE_ENV, TRUST_PROXY: "maybe" }).success, + ).toBe(false); + expect( + envSchema.safeParse({ + ...BASE_ENV, + RECONCILIATION_SCHEDULE: "61 * * * *", + }).success, + ).toBe(false); + expect( + envSchema.safeParse({ + ...BASE_ENV, + RECONCILIATION_SCHEDULE: "*/15 * * * *", + }).success, + ).toBe(true); + }); + }); }); diff --git a/tests/scheduler.test.ts b/tests/scheduler.test.ts new file mode 100644 index 00000000..8ae7db62 --- /dev/null +++ b/tests/scheduler.test.ts @@ -0,0 +1,451 @@ +import { register } from "../src/metrics/registry"; +import { + Scheduler, + parseCron, + matchesCron, + nextCronSlot, + formatSlotIsoMinute, + buildReconciliationJobId, + enqueueReconciliationJob, + initializeScheduler, + stopScheduler, + scheduler, + type ReconciliationQueueLike, +} from "../src/services/scheduler"; + +interface PendingTimer { + id: number; + runAtMs: number; + callback: () => void; + cleared: boolean; +} + +class DeterministicClock { + private currentMs: number; + private nextTimerId = 1; + private readonly timers = new Map(); + + constructor(initialIso: string) { + this.currentMs = Date.parse(initialIso); + } + + now = (): Date => new Date(this.currentMs); + + setTimeout = ( + callback: () => void, + delayMs: number, + ): ReturnType => { + const id = this.nextTimerId++; + const safeDelay = Math.max(0, delayMs); + this.timers.set(id, { + id, + runAtMs: this.currentMs + safeDelay, + callback, + cleared: false, + }); + return id as unknown as ReturnType; + }; + + clearTimeout = (handle: ReturnType): void => { + const id = handle as unknown as number; + const timer = this.timers.get(id); + if (timer) { + timer.cleared = true; + this.timers.delete(id); + } + }; + + get activeTimerCount(): number { + return this.timers.size; + } + + async advanceTo(targetIso: string): Promise { + const targetMs = Date.parse(targetIso); + while (true) { + let earliest: PendingTimer | undefined; + for (const timer of this.timers.values()) { + if (!timer.cleared && timer.runAtMs <= targetMs) { + if (!earliest || timer.runAtMs < earliest.runAtMs) { + earliest = timer; + } + } + } + if (!earliest) break; + this.timers.delete(earliest.id); + this.currentMs = earliest.runAtMs; + earliest.callback(); + await Promise.resolve(); + await Promise.resolve(); + } + this.currentMs = targetMs; + } + + async advanceWithEventLoopDelay( + fireAtIso: string, + actualCallbackClockIso: string, + ): Promise { + const fireAtMs = Date.parse(fireAtIso); + const delayedMs = Date.parse(actualCallbackClockIso); + const due: PendingTimer[] = []; + for (const timer of this.timers.values()) { + if (!timer.cleared && timer.runAtMs <= fireAtMs) { + due.push(timer); + } + } + due.sort((a, b) => a.runAtMs - b.runAtMs); + this.currentMs = delayedMs; + for (const timer of due) { + this.timers.delete(timer.id); + timer.callback(); + await Promise.resolve(); + await Promise.resolve(); + } + } +} + +class FakeBullQueue implements ReconciliationQueueLike { + readonly jobs = new Map< + string, + { name: string; data: { type: "global" }; opts?: { jobId?: string } } + >(); + + readonly add = jest.fn( + async ( + name: string, + data: { type: "global" }, + opts?: { jobId?: string }, + ) => { + const id = opts?.jobId ?? `auto-${this.jobs.size + 1}`; + if (this.jobs.has(id)) { + return { id, name, data, opts, deduped: true }; + } + const created = { id, name, data, opts, deduped: false }; + this.jobs.set(id, created); + return created; + }, + ); + + readonly getJob = jest.fn(async (jobId: string) => { + return this.jobs.get(jobId) ?? null; + }); +} + +describe("Scheduler & Cron Parser (Issue #982)", () => { + const originalTz = process.env.TZ; + + afterEach(() => { + stopScheduler(); + process.env.TZ = originalTz; + }); + + it("Row 1: */15 * * * * at 10:14:59Z, then 10:15:00Z fires exactly once at 10:15Z", async () => { + const clock = new DeterministicClock("2026-10-18T10:14:59.000Z"); + const sched = new Scheduler({ + now: clock.now, + setTimeout: clock.setTimeout, + clearTimeout: clock.clearTimeout, + }); + + const firedAt: string[] = []; + sched.schedule("quarter-hour", "*/15 * * * *", async (slot) => { + firedAt.push(slot.toISOString()); + }); + + expect(firedAt).toEqual([]); + + await clock.advanceTo("2026-10-18T10:15:00.000Z"); + expect(firedAt).toEqual(["2026-10-18T10:15:00.000Z"]); + + // Advancing within the same 10:15 minute does not fire a second time + await clock.advanceTo("2026-10-18T10:29:59.000Z"); + expect(firedAt).toEqual(["2026-10-18T10:15:00.000Z"]); + + sched.stopAll(); + }); + + it("Row 2: 0 2 * * 1-5 on a Saturday does not fire on weekend; fires Monday 02:00Z", async () => { + // 2026-10-17 is Saturday (UTC day 6) + const clock = new DeterministicClock("2026-10-17T01:00:00.000Z"); + const sched = new Scheduler({ + now: clock.now, + setTimeout: clock.setTimeout, + clearTimeout: clock.clearTimeout, + }); + + const firedAt: string[] = []; + sched.schedule("weekday-2am", "0 2 * * 1-5", async (slot) => { + firedAt.push(slot.toISOString()); + }); + + // Saturday 02:00Z and Sunday 02:00Z must not fire + await clock.advanceTo("2026-10-17T12:00:00.000Z"); + expect(firedAt).toEqual([]); + + await clock.advanceTo("2026-10-18T23:59:59.000Z"); + expect(firedAt).toEqual([]); + + // Monday 2026-10-19T02:00:00Z fires + await clock.advanceTo("2026-10-19T02:00:00.000Z"); + expect(firedAt).toEqual(["2026-10-19T02:00:00.000Z"]); + + sched.stopAll(); + }); + + it("Row 3: 0 0,12 * * * fires at 00:00Z and 12:00Z only", async () => { + const clock = new DeterministicClock("2026-10-18T00:00:00.000Z"); + const sched = new Scheduler({ + now: clock.now, + setTimeout: clock.setTimeout, + clearTimeout: clock.clearTimeout, + }); + + const firedAt: string[] = []; + sched.schedule("twice-daily", "0 0,12 * * *", async (slot) => { + firedAt.push(slot.toISOString()); + }); + + await clock.advanceTo("2026-10-19T00:00:00.000Z"); + expect(firedAt).toEqual([ + "2026-10-18T12:00:00.000Z", + "2026-10-19T00:00:00.000Z", + ]); + + sched.stopAll(); + }); + + it("Row 4: invalid cron expressions throw on parseCron and schedule()", () => { + const clock = new DeterministicClock("2026-10-18T00:00:00.000Z"); + const sched = new Scheduler({ + now: clock.now, + setTimeout: clock.setTimeout, + clearTimeout: clock.clearTimeout, + }); + + const invalidExpressions = [ + "61 * * * *", + "* * *", + "*/0 * * * *", + "* 24 * * *", + "* * 0 * *", + "* * 32 * *", + "* * * 0 *", + "* * * 13 *", + "* * * * 7", + "10-5 * * * *", + "10/2 * * * *", + "1, * * * *", + "0 0 31 2 *", + "", + ]; + + for (const expr of invalidExpressions) { + expect(() => parseCron(expr)).toThrow(); + expect(() => + sched.schedule(`bad-${expr}`, expr, async () => undefined), + ).toThrow(); + } + expect(clock.activeTimerCount).toBe(0); + }); + + it("Row 5: evaluates fire times in UTC even when host TZ is America/New_York", async () => { + process.env.TZ = "America/New_York"; + const clock = new DeterministicClock("2026-10-18T01:59:30.000Z"); + const sched = new Scheduler({ + now: clock.now, + setTimeout: clock.setTimeout, + clearTimeout: clock.clearTimeout, + }); + + const firedAt: string[] = []; + sched.schedule("utc-2am", "0 2 * * *", async (slot) => { + firedAt.push(slot.toISOString()); + }); + + await clock.advanceTo("2026-10-18T02:00:00.000Z"); + expect(firedAt).toEqual(["2026-10-18T02:00:00.000Z"]); + + sched.stopAll(); + }); + + it("Row 6: two scheduler instances (two replicas) enqueue the same deterministic jobId and the queue keeps one job", async () => { + const clockA = new DeterministicClock("2026-10-18T01:59:50.000Z"); + const clockB = new DeterministicClock("2026-10-18T01:59:55.000Z"); + const sharedQueue = new FakeBullQueue(); + + const replicaA = new Scheduler({ + now: clockA.now, + setTimeout: clockA.setTimeout, + clearTimeout: clockA.clearTimeout, + }); + const replicaB = new Scheduler({ + now: clockB.now, + setTimeout: clockB.setTimeout, + clearTimeout: clockB.clearTimeout, + }); + + replicaA.schedule("reconciliation", "0 2 * * *", (slot) => + enqueueReconciliationJob(slot, sharedQueue), + ); + replicaB.schedule("reconciliation", "0 2 * * *", (slot) => + enqueueReconciliationJob(slot, sharedQueue), + ); + + await clockA.advanceTo("2026-10-18T02:00:00.000Z"); + await clockB.advanceTo("2026-10-18T02:00:00.000Z"); + + const expectedJobId = "reconciliation:2026-10-18T02:00Z"; + expect( + buildReconciliationJobId(new Date("2026-10-18T02:00:00.000Z")), + ).toBe(expectedJobId); + expect(sharedQueue.add).toHaveBeenCalledTimes(2); + expect(sharedQueue.add).toHaveBeenNthCalledWith( + 1, + "reconciliation-job", + { type: "global" }, + { jobId: expectedJobId }, + ); + expect(sharedQueue.add).toHaveBeenNthCalledWith( + 2, + "reconciliation-job", + { type: "global" }, + { jobId: expectedJobId }, + ); + expect(sharedQueue.jobs.size).toBe(1); + + replicaA.stopAll(); + replicaB.stopAll(); + }); + + it("Row 7: event-loop delay of 70 s around a slot runs the task once for that slot, not twice", async () => { + const clock = new DeterministicClock("2026-10-18T10:14:50.000Z"); + const sched = new Scheduler({ + now: clock.now, + setTimeout: clock.setTimeout, + clearTimeout: clock.clearTimeout, + }); + + const slotsRun: string[] = []; + sched.schedule("delayed-task", "*/15 * * * *", async (slot) => { + slotsRun.push(formatSlotIsoMinute(slot)); + }); + + // Timer was scheduled for 10:15:00Z, but event loop was stalled 70s until 10:16:10Z + await clock.advanceWithEventLoopDelay( + "2026-10-18T10:15:00.000Z", + "2026-10-18T10:16:10.000Z", + ); + expect(slotsRun).toEqual(["2026-10-18T10:15Z"]); + + // Attempting to execute the same slot again via the in-process guard is a no-op + const ranAgain = await sched.runSlotIfNotExecuted( + "delayed-task", + new Date("2026-10-18T10:15:00.000Z"), + ); + expect(ranAgain).toBe(false); + expect(slotsRun).toEqual(["2026-10-18T10:15Z"]); + + // Next slot fires at 10:30Z + await clock.advanceTo("2026-10-18T10:30:00.000Z"); + expect(slotsRun).toEqual(["2026-10-18T10:15Z", "2026-10-18T10:30Z"]); + + sched.stopAll(); + }); + + it("Row 8 & initializeScheduler: respects RECONCILIATION_ENABLED=false and schedules when enabled", async () => { + const clock = new DeterministicClock("2026-10-18T01:59:00.000Z"); + const fakeQueue = new FakeBullQueue(); + const targetScheduler = new Scheduler({ + now: clock.now, + setTimeout: clock.setTimeout, + clearTimeout: clock.clearTimeout, + }); + + initializeScheduler({ + enabled: false, + schedule: "0 2 * * *", + schedulerInstance: targetScheduler, + queue: fakeQueue, + }); + expect(clock.activeTimerCount).toBe(0); + + initializeScheduler({ + enabled: true, + schedule: "0 2 * * *", + schedulerInstance: targetScheduler, + queue: fakeQueue, + }); + expect(clock.activeTimerCount).toBe(1); + + await clock.advanceTo("2026-10-18T02:00:00.000Z"); + expect(fakeQueue.jobs.size).toBe(1); + expect(fakeQueue.jobs.has("reconciliation:2026-10-18T02:00Z")).toBe(true); + + targetScheduler.stopAll(); + }); + + it("Row 10: stopScheduler() and unschedule() clear all pending timers and handle errors/metrics", async () => { + const clock = new DeterministicClock("2026-10-18T10:00:00.000Z"); + const sched = new Scheduler({ + now: clock.now, + setTimeout: clock.setTimeout, + clearTimeout: clock.clearTimeout, + }); + + const runs: number[] = []; + sched.schedule("step-range", "10-40/15 * * * *", async (slot) => { + runs.push(slot.getUTCMinutes()); + }); + // Duplicate schedule call is ignored + sched.schedule("step-range", "10-40/15 * * * *", async () => undefined); + + sched.schedule("failing-task", "*/10 * * * *", async () => { + throw new Error("boom"); + }); + + expect(clock.activeTimerCount).toBe(2); + + await clock.advanceTo("2026-10-18T10:10:00.000Z"); + expect(runs).toEqual([10]); + + sched.unschedule("step-range"); + sched.unschedule("non-existent"); + expect(clock.activeTimerCount).toBe(1); + + sched.stopAll(); + expect(clock.activeTimerCount).toBe(0); + + // Global scheduler stopScheduler() also leaves zero active timers + expect(scheduler).toBeInstanceOf(Scheduler); + initializeScheduler(); + stopScheduler(); + + // Exercise matchesCron (string + ParsedCron), leap-day month-jumping, bare queue without getJob, and non-Error throw + const leapCron = parseCron("0 0 29 2 *"); + expect(matchesCron("0 0 29 2 *", new Date("2028-02-29T00:00:00.000Z"))).toBe(true); + expect(matchesCron(leapCron, new Date("2026-02-28T00:00:00.000Z"))).toBe(false); + expect(nextCronSlot("0 0 29 2 *", new Date("2026-10-18T00:00:00.000Z")).toISOString()).toBe( + "2028-02-29T00:00:00.000Z", + ); + expect(() => parseCron("0/2/3 * * * *")).toThrow(); + expect(() => parseCron("1-2-3 * * * *")).toThrow(); + expect(() => parseCron("abc * * * *")).toThrow(); + + const bareQueue: ReconciliationQueueLike = { + add: jest.fn().mockResolvedValue({ id: "reconciliation:2026-10-18T02:00Z" }), + }; + const slotDate = new Date("2026-10-18T02:00:00.000Z"); + expect(await enqueueReconciliationJob(slotDate, bareQueue)).toBe("enqueued"); + expect(await enqueueReconciliationJob(slotDate, bareQueue)).toBe("deduped"); + + expect(await sched.runSlotIfNotExecuted("missing-task", slotDate)).toBe(false); + sched.schedule("string-throw", "* * * * *", async () => { + throw "non-error-failure"; + }); + await sched.runSlotIfNotExecuted("string-throw", slotDate); + sched.stopAll(); + + const metricsText = await register.metrics(); + expect(metricsText).toContain("scheduler_runs_total"); + }); +}); + From 6551d319cca113b381936d48ee5b2976133a9f5b Mon Sep 17 00:00:00 2001 From: lhphong09t2 Date: Mon, 19 Oct 2026 04:34:22 +0700 Subject: [PATCH 2/3] fix(scheduler): clamp setTimeout delays to MAX_TIMEOUT_MS and bound slot tracking sets Why: Delays exceeding 2^31 - 1 ms (~24.85 days, e.g., monthly/yearly cron schedules) overflow Node's 32-bit timer and fire immediately at 1 ms, while unbounded slot sets grow over long uptime. What: Split long delays across MAX_TIMEOUT_MS hops with pre-slot wake check, bound executedSlots and seenJobIdsByQueue to 1024 entries, and document AND field semantics and BullMQ 3-segment jobId format. Verification: npm test -- tests/scheduler.test.ts tests/env.test.ts tests/check-env.test.ts --coverage --collectCoverageFrom='src/services/scheduler.ts' (38/38 passed, 99.48% lines / 96.59% branches, exit 0). --- docs/job-queue.md | 4 ++-- src/services/scheduler.ts | 42 +++++++++++++++++++++++++++++++++++---- tests/scheduler.test.ts | 28 ++++++++++++++++++++++++++ 3 files changed, 68 insertions(+), 6 deletions(-) diff --git a/docs/job-queue.md b/docs/job-queue.md index 613d0c41..b0128df3 100644 --- a/docs/job-queue.md +++ b/docs/job-queue.md @@ -186,9 +186,9 @@ when `scheduler.schedule()` is called using a dependency-free 5-field parser: - Inclusive range: `a-b` (e.g. `1-5`) - Step on wildcard or range: `*/n` or `a-b/n` (e.g. `*/15`, `8-18/2`) - Comma-separated list: `a,b,c` combining any of the above (e.g. `0,30`, `1-5,10`) -- **Calendar validation:** Impossible day/month combinations (such as `0 0 30 2 *` or `0 0 31 4 *`) are rejected at parse time; leap-day schedules (`0 0 29 2 *`) are permitted. +- **Calendar validation & field matching:** Impossible day/month combinations (such as `0 0 30 2 *` or `0 0 31 4 *`) are rejected at parse time; leap-day schedules (`0 0 29 2 *`) are permitted. All 5 fields (`minute`, `hour`, `day of month`, `month`, and `day of week`) are combined with logical **AND**. - **UTC evaluation:** All cron fields and slot calculations are evaluated strictly in **UTC** (`getUTCMinutes`, `getUTCHours`, `getUTCDate`, `getUTCMonth`, `getUTCDay`), independent of the host server's local timezone. -- **Slot-based firing & replica deduplication:** Instead of polling every 60 seconds, the scheduler computes the next matching UTC minute boundary (`nextCronSlot`), arms a single `setTimeout` for that slot, guards against duplicate in-process execution per slot, and enqueues onto `reconciliationQueue` with a deterministic `jobId`: +- **Slot-based firing, 32-bit timer safety & replica deduplication:** Instead of polling every 60 seconds, the scheduler computes the next matching UTC minute boundary (`nextCronSlot`), arms a `setTimeout` for that slot (clamping individual timer hops to `MAX_TIMEOUT_MS = 2,147,483,647` ms so monthly/yearly schedules never trigger `TimeoutOverflowWarning`), guards against duplicate in-process execution per slot, and enqueues onto `reconciliationQueue` with a deterministic `jobId`: `reconciliation:` (for example, `reconciliation:2026-10-18T02:00Z`). When multiple backend replicas fire on the same slot, BullMQ deduplicates by `jobId` so only one global reconciliation job runs per slot. - **Observability:** Next fire time is logged whenever a task is armed, and each slot execution increments the Prometheus counter `scheduler_runs_total{task,outcome}` (`outcome="enqueued" | "deduped" | "error"`). diff --git a/src/services/scheduler.ts b/src/services/scheduler.ts index 74a7698f..b0689763 100644 --- a/src/services/scheduler.ts +++ b/src/services/scheduler.ts @@ -169,6 +169,24 @@ export function nextCronSlot(cron: ParsedCron | string, fromDate: Date): Date { throw new Error(`No matching UTC cron slot found within 8 years for "${parsed.expression}"`); } +/** + * Maximum 32-bit signed integer delay supported by Node.js timers (~24.85 days). + * Delays exceeding this value are split across intermediate timer hops to prevent + * `TimeoutOverflowWarning` (which would otherwise clamp the delay to 1 ms). + */ +export const MAX_TIMEOUT_MS = 2_147_483_647; +export const MAX_TRACKED_SLOTS = 1_024; + +function addBoundedSlot(set: Set, key: string): void { + if (set.size >= MAX_TRACKED_SLOTS) { + const oldest = set.values().next().value; + if (oldest !== undefined) { + set.delete(oldest); + } + } + set.add(key); +} + /** * Format a Date to its UTC ISO minute string (`YYYY-MM-DDTHH:mmZ`). */ @@ -178,6 +196,8 @@ export function formatSlotIsoMinute(date: Date): string { /** * Build the deterministic BullMQ jobId for a reconciliation slot (`reconciliation:`). + * Uses `YYYY-MM-DDTHH:mmZ` (omitting `:ss`) so the `jobId` contains exactly 3 colon-separated + * segments, satisfying BullMQ 5.x custom `jobId` validation rules. */ export function buildReconciliationJobId(slotDate: Date): string { return `reconciliation:${formatSlotIsoMinute(slotDate)}`; @@ -245,7 +265,7 @@ export async function enqueueReconciliationJob( const alreadySeen = seenSet.has(jobId); await queue.add("reconciliation-job", { type: "global" }, { jobId }); - seenSet.add(jobId); + addBoundedSlot(seenSet, jobId); const wasDeduped = Boolean(existingInQueue) || alreadySeen; const outcome: SchedulerTaskOutcome = wasDeduped ? "deduped" : "enqueued"; @@ -331,9 +351,23 @@ export class Scheduler { "Task scheduled", ); + this.armTimerUntilSlot(entry, nextSlot); + } + + private armTimerUntilSlot(entry: ScheduledTaskEntry, targetSlot: Date): void { + const remainingMs = Math.max(0, targetSlot.getTime() - this.nowFn().getTime()); + const waitMs = Math.min(remainingMs, MAX_TIMEOUT_MS); + entry.timer = this.setTimeoutFn(() => { - void this.fireSlot(entry, nextSlot); - }, delayMs); + if (this.tasks.get(entry.name) !== entry) { + return; + } + if (this.nowFn().getTime() < targetSlot.getTime()) { + this.armTimerUntilSlot(entry, targetSlot); + return; + } + void this.fireSlot(entry, targetSlot); + }, waitMs); } /** @@ -351,7 +385,7 @@ export class Scheduler { log.info({ name: entry.name, slot: slotIso }, "Skipping duplicate in-process slot execution"); return false; } - this.executedSlots.add(slotKey); + addBoundedSlot(this.executedSlots, slotKey); try { log.info({ name: entry.name, cron: entry.parsedCron.expression, slot: slotIso }, "Running scheduled task"); diff --git a/tests/scheduler.test.ts b/tests/scheduler.test.ts index 8ae7db62..4c10de9b 100644 --- a/tests/scheduler.test.ts +++ b/tests/scheduler.test.ts @@ -434,9 +434,15 @@ describe("Scheduler & Cron Parser (Issue #982)", () => { add: jest.fn().mockResolvedValue({ id: "reconciliation:2026-10-18T02:00Z" }), }; const slotDate = new Date("2026-10-18T02:00:00.000Z"); + expect(buildReconciliationJobId(slotDate).split(":")).toHaveLength(3); expect(await enqueueReconciliationJob(slotDate, bareQueue)).toBe("enqueued"); expect(await enqueueReconciliationJob(slotDate, bareQueue)).toBe("deduped"); + // Exercise bounded slot eviction (> 1024 unique slots) + for (let i = 0; i < 1030; i++) { + await enqueueReconciliationJob(new Date(slotDate.getTime() + (i + 1) * 60_000), bareQueue); + } + expect(await sched.runSlotIfNotExecuted("missing-task", slotDate)).toBe(false); sched.schedule("string-throw", "* * * * *", async () => { throw "non-error-failure"; @@ -444,8 +450,30 @@ describe("Scheduler & Cron Parser (Issue #982)", () => { await sched.runSlotIfNotExecuted("string-throw", slotDate); sched.stopAll(); + // Monthly schedule > 24.85 days (MAX_TIMEOUT_MS = 2^31 - 1) hops safely without premature firing + const monthlyClock = new DeterministicClock("2026-01-01T02:01:00.000Z"); + const monthlySched = new Scheduler({ + now: monthlyClock.now, + setTimeout: monthlyClock.setTimeout, + clearTimeout: monthlyClock.clearTimeout, + }); + const monthlyFired: string[] = []; + monthlySched.schedule("monthly", "0 2 1 * *", async (slot) => { + monthlyFired.push(slot.toISOString()); + }); + + // At +25 days (after first 24.85d hop), task must NOT have fired yet + await monthlyClock.advanceTo("2026-01-26T12:00:00.000Z"); + expect(monthlyFired).toEqual([]); + + // At 2026-02-01T02:00:00.000Z, task fires once + await monthlyClock.advanceTo("2026-02-01T02:00:00.000Z"); + expect(monthlyFired).toEqual(["2026-02-01T02:00:00.000Z"]); + monthlySched.stopAll(); + const metricsText = await register.metrics(); expect(metricsText).toContain("scheduler_runs_total"); }); }); + From 9afcf906348a420b68d0e5b6f06a8a30fb3f686f Mon Sep 17 00:00:00 2001 From: lhphong09t2 Date: Mon, 19 Oct 2026 04:37:21 +0700 Subject: [PATCH 3/3] test(scheduler): enforce 32-bit timer overflow regression check in DeterministicClock Why: DeterministicClock accepted delays > 2^31 - 1 ms without clamping to 1 ms, so the monthly schedule test passed even against un-clamped setTimeout calls. What: Record requested timer delays, emulate Node 32-bit timer overflow (> 2_147_483_647 ms -> 1 ms) in DeterministicClock, and assert recordedDelays <= MAX_TIMEOUT_MS. Verification: Confirmed test fails on cd54e5cd and passes on HEAD (38/38 passed, 99.48% lines / 96.59% branches). --- tests/scheduler.test.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/scheduler.test.ts b/tests/scheduler.test.ts index 4c10de9b..a5c6bcc6 100644 --- a/tests/scheduler.test.ts +++ b/tests/scheduler.test.ts @@ -10,6 +10,7 @@ import { initializeScheduler, stopScheduler, scheduler, + MAX_TIMEOUT_MS, type ReconciliationQueueLike, } from "../src/services/scheduler"; @@ -24,6 +25,7 @@ class DeterministicClock { private currentMs: number; private nextTimerId = 1; private readonly timers = new Map(); + readonly recordedDelays: number[] = []; constructor(initialIso: string) { this.currentMs = Date.parse(initialIso); @@ -35,8 +37,10 @@ class DeterministicClock { callback: () => void, delayMs: number, ): ReturnType => { + this.recordedDelays.push(delayMs); const id = this.nextTimerId++; - const safeDelay = Math.max(0, delayMs); + // Match Node.js 32-bit signed int timer behavior: delays > 2^31 - 1 overflow to 1 ms + const safeDelay = delayMs > 2_147_483_647 ? 1 : Math.max(0, delayMs); this.timers.set(id, { id, runAtMs: this.currentMs + safeDelay, @@ -461,6 +465,7 @@ describe("Scheduler & Cron Parser (Issue #982)", () => { monthlySched.schedule("monthly", "0 2 1 * *", async (slot) => { monthlyFired.push(slot.toISOString()); }); + expect(monthlyClock.recordedDelays[0]).toBe(MAX_TIMEOUT_MS); // At +25 days (after first 24.85d hop), task must NOT have fired yet await monthlyClock.advanceTo("2026-01-26T12:00:00.000Z"); @@ -469,6 +474,7 @@ describe("Scheduler & Cron Parser (Issue #982)", () => { // At 2026-02-01T02:00:00.000Z, task fires once await monthlyClock.advanceTo("2026-02-01T02:00:00.000Z"); expect(monthlyFired).toEqual(["2026-02-01T02:00:00.000Z"]); + expect(Math.max(...monthlyClock.recordedDelays)).toBeLessThanOrEqual(MAX_TIMEOUT_MS); monthlySched.stopAll(); const metricsText = await register.metrics();