fix(scheduler): run reconciliation cron in UTC with replica dedupe

docs/job-queue.md

@@ -133,19 +133,21 @@ 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(() => {
   webhookWorker.start();
   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 & 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, 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:<YYYY-MM-DDTHH:mmZ>` (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`).

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),

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<typeof envSchema.parse>);
+
+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.`
+    );
+  }
+}
+

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();
 

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],
+});