diff --git a/src/lib/db/migrations/139_job_registry.sql b/src/lib/db/migrations/139_job_registry.sql index e35519a418..3bf29ac3d2 100644 --- a/src/lib/db/migrations/139_job_registry.sql +++ b/src/lib/db/migrations/139_job_registry.sql @@ -1,4 +1,4 @@ --- Migration 136: generic job registry (jobs + job_runs tables) +-- Migration 139: generic job registry (jobs + job_runs tables) -- Job registry (#8848): centralized periodic-job scheduling + run history. -- -- jobs: one row per registered job (interval or cron), with env-flag gating @@ -33,10 +33,14 @@ CREATE INDEX IF NOT EXISTS idx_jr_started_at ON job_runs(started_at); -- Built-in job registration (idempotent - INSERT OR IGNORE). -- env_flag = NULL means "no registry-level boolean gate"; per-job disable semantics -- live inside the handler itself (see token_health_check wrapper). -INSERT OR IGNORE INTO jobs (id, type, cron, interval_ms, env_flag, config) VALUES - ('budget_reset', 'interval', NULL, 600000, NULL, '{}'), - ('warmup', 'cron', '0 7 * * *', NULL, 'OMNIROUTE_WARMUP_ENABLED', '{"timezone":"America/Los_Angeles","envDefault":false}'), - ('token_health_check', 'interval', NULL, 60000, NULL, '{}'); +-- 'warmup' is seeded disabled because its handler is not part of this change. +-- startAll() filters on enabled before it looks for a handler, so a disabled row +-- stays quiet instead of warning on every boot; the change that brings the warmup +-- handler flips it on. +INSERT OR IGNORE INTO jobs (id, type, cron, interval_ms, enabled, env_flag, config) VALUES + ('budget_reset', 'interval', NULL, 600000, 1, NULL, '{}'), + ('warmup', 'cron', '0 7 * * *', NULL, 0, 'OMNIROUTE_WARMUP_ENABLED', '{"timezone":"America/Los_Angeles","envDefault":false}'), + ('token_health_check', 'interval', NULL, 60000, 1, NULL, '{}'); -- records_affected semantics: -- budget_reset = number of budget records reset (UPDATE ... SET budget_used=0 row count) diff --git a/src/lib/initCloudSync.ts b/src/lib/initCloudSync.ts index 9032452cd0..3464ea0d29 100644 --- a/src/lib/initCloudSync.ts +++ b/src/lib/initCloudSync.ts @@ -1,8 +1,9 @@ import initializeCloudSync from "@/shared/services/initializeCloudSync"; -import { startBudgetResetJob } from "@/lib/jobs/budgetResetJob"; import { startModelSyncScheduler } from "@/shared/services/modelSyncScheduler"; -import { startWarmupScheduler } from "@/lib/warmupScheduler"; import { isAutomatedTestProcess } from "@/shared/utils/testProcess"; +import { getJobRegistry } from "@/lib/jobRegistry"; +import { registerBudgetResetJob } from "@/lib/jobs/budgetResetJob"; +import { registerTokenHealthCheck } from "@/lib/jobs/tokenHealthCheckJob"; // Initialize runtime background sync services once per server process. let initialized = false; @@ -29,12 +30,19 @@ export async function ensureCloudSyncInitialized() { } if (!initialized) { try { - const { initTokenHealthCheck } = await import("@/lib/tokenHealthCheck"); - initTokenHealthCheck(); await initializeCloudSync(); startModelSyncScheduler(); - startBudgetResetJob(); - startWarmupScheduler(); + + // startAll() runs each interval job's first tick synchronously, so it has to + // come after initializeCloudSync(). The old wiring got that ordering two + // different ways: the budget reset was started right here, and the health + // check's first sweep sat behind a 10s timer. Awaiting the init is a firmer + // guarantee than the timer was. + const registry = getJobRegistry(); + registerBudgetResetJob(registry); + registerTokenHealthCheck(registry); + await registry.startAll(); + initialized = true; } catch (error) { console.error("[ServerInit] Error initializing background sync services:", error); diff --git a/src/lib/jobs/budgetResetJob.ts b/src/lib/jobs/budgetResetJob.ts index bd640569e8..f29ae383a1 100644 --- a/src/lib/jobs/budgetResetJob.ts +++ b/src/lib/jobs/budgetResetJob.ts @@ -1,40 +1,42 @@ import { syncAllBudgetSchedules } from "@/domain/costRules"; +import type { JobRegistry } from "../jobRegistry/registry"; const DEFAULT_INTERVAL_MS = 10 * 60 * 1000; -let timer: NodeJS.Timeout | null = null; - function getIntervalMs() { const raw = process.env.OMNIROUTE_BUDGET_RESET_JOB_INTERVAL_MS; const parsed = raw ? Number(raw) : Number.NaN; return Number.isFinite(parsed) && parsed >= 10_000 ? parsed : DEFAULT_INTERVAL_MS; } -export function startBudgetResetJob() { - if (timer) { - return timer; +/** + * Budget-reset handler - resets all budget schedules once per interval. + * + * Migrated to the JobRegistry: the old start/stop timer logic is + * gone; the registry owns scheduling + run recording. This module exports just the + * `run` handler and a `registerBudgetResetJob` wiring helper. Errors are no longer + * caught here - safeRun records them as a failure run. + */ +export async function run(): Promise<{ success: boolean; recordsAffected: number }> { + const result = syncAllBudgetSchedules(Date.now()); + if (result.resetCount > 0) { + console.log(`[BudgetReset] processed=${result.processed} reset=${result.resetCount}`); } - - const run = () => { - try { - const result = syncAllBudgetSchedules(Date.now()); - if (result.resetCount > 0) { - console.log(`[BudgetReset] processed=${result.processed} reset=${result.resetCount}`); - } - } catch (error) { - console.error("[BudgetReset] Job failed:", error); - } - }; - - run(); - timer = setInterval(run, getIntervalMs()); - timer.unref?.(); - return timer; + return { success: true, recordsAffected: result.resetCount }; } -export function stopBudgetResetJob() { - if (timer) { - clearInterval(timer); - timer = null; - } +/** Wire the budget_reset job into a JobRegistry (idempotent - call at boot). */ +export function registerBudgetResetJob(registry: JobRegistry): void { + registry.register({ + id: "budget_reset", + type: "interval", + cron: null, + intervalMs: getIntervalMs(), + enabled: true, + envFlag: null, + config: {}, + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + handler: run, + }); } diff --git a/src/lib/jobs/tokenHealthCheckJob.ts b/src/lib/jobs/tokenHealthCheckJob.ts new file mode 100644 index 0000000000..73e15d784c --- /dev/null +++ b/src/lib/jobs/tokenHealthCheckJob.ts @@ -0,0 +1,40 @@ +/** + * Registry adapter for the token health check. + * + * The sweep used to start itself on import. The registry owns the schedule now, + * so the adapter lives here next to the other job registrations rather than in + * tokenHealthCheck.ts, which is already at its size ceiling. + * + * Disable semantics are unchanged: isHealthCheckDisabled() still honours + * OMNIROUTE_DISABLE_TOKEN_HEALTHCHECK, NEXT_PHASE=phase-production-build and + * isAutomatedTestProcess(). The registry only fires the interval; the handler + * decides whether there is anything to do. + */ +import { isHealthCheckDisabled, sweep } from "@/lib/tokenHealthCheck"; +import type { JobRegistry } from "@/lib/jobRegistry/registry"; + +const TOKEN_HEALTH_CHECK_INTERVAL_MS = 60_000; + +export function registerTokenHealthCheck(registry: JobRegistry): void { + const now = new Date().toISOString(); + registry.register({ + id: "token_health_check", + type: "interval", + cron: null, + intervalMs: TOKEN_HEALTH_CHECK_INTERVAL_MS, + enabled: true, + envFlag: null, + config: {}, + createdAt: now, + updatedAt: now, + handler: async () => { + if (isHealthCheckDisabled()) { + return { success: true, recordsAffected: 0 }; + } + // Errors are not caught here: safeRun records a thrown error as a failure run + // with its message, the same as the budget reset job. + const swept = await sweep(); + return { success: true, recordsAffected: swept }; + }, + }); +} diff --git a/src/lib/tokenHealthCheck.ts b/src/lib/tokenHealthCheck.ts index ce54c2e675..a368b85c2a 100644 --- a/src/lib/tokenHealthCheck.ts +++ b/src/lib/tokenHealthCheck.ts @@ -267,7 +267,7 @@ function isEnvFlagEnabled(name: string): boolean { return TRUE_ENV_VALUES.has(value.trim().toLowerCase()); } -function isHealthCheckDisabled(): boolean { +export function isHealthCheckDisabled(): boolean { return ( isEnvFlagEnabled("OMNIROUTE_DISABLE_TOKEN_HEALTHCHECK") || isBuildProcess() || @@ -421,16 +421,18 @@ export function stopTokenHealthCheck() { } // ── Core sweep (batch concurrent) ────────────────────────────────────────── -export async function sweep() { +/** Returns the number of connections swept, which the job registry records. */ +export async function sweep(): Promise { const state = getHCState(); if (state.sweeping) { - return log(`${LOG_PREFIX} Sweep skipped — previous sweep still in progress`); + log(`${LOG_PREFIX} Sweep skipped — previous sweep still in progress`); + return 0; } state.sweeping = true; try { const connections = await getProviderConnections({ authType: "oauth" }); - if (!connections || connections.length === 0) return; + if (!connections || connections.length === 0) return 0; const staggerMs = parseInt(process.env.HEALTHCHECK_STAGGER_MS || "3000", 10); const total = connections.length; @@ -471,8 +473,10 @@ export async function sweep() { await new Promise((resolve) => setTimeout(resolve, 0)); } } + return total; } catch (err) { logError(`${LOG_PREFIX} Sweep error:`, err.message); + return 0; } finally { state.sweeping = false; } @@ -1045,8 +1049,3 @@ export async function checkConnection(conn) { ); } } - -// Auto-start when imported -initTokenHealthCheck(); - -export default initTokenHealthCheck; diff --git a/src/server-init.ts b/src/server-init.ts index bdbcb64dde..04e6190d5a 100644 --- a/src/server-init.ts +++ b/src/server-init.ts @@ -4,7 +4,8 @@ import { enforceWebRuntimeEnv } from "./lib/env/runtimeEnv"; import { enforceSecrets } from "./shared/utils/secretsValidator"; import { initAuditLog, cleanupExpiredLogs, logAuditEvent } from "./lib/compliance/index"; import { initConsoleInterceptor } from "./lib/consoleInterceptor"; -import { startBudgetResetJob } from "./lib/jobs/budgetResetJob"; +import { registerBudgetResetJob } from "./lib/jobs/budgetResetJob"; +import { registerTokenHealthCheck } from "./lib/jobs/tokenHealthCheckJob"; import { startReasoningCacheCleanupJob } from "./lib/jobs/reasoningCacheCleanupJob"; import { startCleanupScheduler } from "./lib/db/cleanup"; import { getSettings } from "./lib/db/settings"; @@ -113,7 +114,16 @@ async function startServer() { } await initializeCloudSync(); - startBudgetResetJob(); + // register() only persists the definition; startAll() is what arms the timers. + // This path does not call ensureCloudSyncInitialized(), so nothing else here + // would start the jobs on our behalf. It registers the same set as that path: + // starting one job and not the other is how a background job goes missing + // without anything failing. + const { getJobRegistry } = await import("./lib/jobRegistry"); + const jobRegistry = getJobRegistry(); + registerBudgetResetJob(jobRegistry); + registerTokenHealthCheck(jobRegistry); + await jobRegistry.startAll(); startReasoningCacheCleanupJob(); startCleanupScheduler(); startRuntimeConfigHotReload(); diff --git a/tests/unit/db/jobRegistryDb.test.ts b/tests/unit/db/jobRegistryDb.test.ts index 2c428a47f2..08aa22a9c7 100644 --- a/tests/unit/db/jobRegistryDb.test.ts +++ b/tests/unit/db/jobRegistryDb.test.ts @@ -1,5 +1,5 @@ /** - * Tests for jobRegistry persistence (migration 136 + jobRegistryDb.ts). + * Tests for jobRegistry persistence (migration 139 + jobRegistryDb.ts). * * Verifies: * - 3 built-in jobs seeded by the migration @@ -54,7 +54,9 @@ test("seed: warmup is cron type with env gate + envDefault=false", () => { assert.equal(warmup.type, "cron"); assert.equal(warmup.cron, "0 7 * * *"); assert.equal(warmup.envFlag, "OMNIROUTE_WARMUP_ENABLED"); - assert.equal(warmup.enabled, true); + // Seeded disabled: the warmup handler is not registered by this change, and + // startAll() would otherwise warn about the missing handler on every boot. + assert.equal(warmup.enabled, false); assert.equal(warmup.config.envDefault, false); assert.equal(warmup.config.timezone, "America/Los_Angeles"); }); diff --git a/tests/unit/lib/jobRegistry/boot-wiring.test.ts b/tests/unit/lib/jobRegistry/boot-wiring.test.ts new file mode 100644 index 0000000000..4aa09dada4 --- /dev/null +++ b/tests/unit/lib/jobRegistry/boot-wiring.test.ts @@ -0,0 +1,61 @@ +/** + * Boot wiring for the job registry. + * + * startAll() runs each interval job's first tick synchronously (see startInterval + * in registry.ts), so starting a job before initializeCloudSync() has been awaited + * runs its handler against a half-initialised app. The previous wiring avoided that + * two different ways: the budget reset was started after the init call, and the + * token health check's first sweep sat behind a 10s timer. + * + * Neither entry point can be driven from a unit test - both are process-boot + * functions that reach the real cloud sync - so these read the wiring, the same + * way tests/unit/model-sync-scheduler.test.ts already asserts on this file. + */ + +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import path from "node:path"; + +function readSource(relativePath: string): string { + return fs.readFileSync(path.join(process.cwd(), relativePath), "utf8"); +} + +test("initCloudSync: cloud sync is initialised before the registry starts any job", () => { + const source = readSource("src/lib/initCloudSync.ts"); + const initCall = source.indexOf("await initializeCloudSync("); + const startAllCall = source.indexOf("registry.startAll("); + + assert.notEqual(initCall, -1, "initializeCloudSync() call not found"); + assert.notEqual(startAllCall, -1, "registry.startAll() call not found"); + assert.ok( + initCall < startAllCall, + "startAll() fires each interval job's first tick immediately, so it must come after initializeCloudSync()" + ); +}); + +test("server-init: cloud sync is initialised before the registry starts any job", () => { + const source = readSource("src/server-init.ts"); + const initCall = source.indexOf("await initializeCloudSync("); + const startAllCall = source.indexOf("jobRegistry.startAll("); + + assert.notEqual(initCall, -1, "initializeCloudSync() call not found"); + assert.notEqual(startAllCall, -1, "jobRegistry.startAll() call not found"); + assert.ok(initCall < startAllCall, "startAll() must come after initializeCloudSync()"); +}); + +test("both boot paths register the same jobs", () => { + for (const relativePath of ["src/lib/initCloudSync.ts", "src/server-init.ts"]) { + const source = readSource(relativePath); + assert.match( + source, + /registerBudgetResetJob\s*\(/, + `${relativePath} must register the budget reset job` + ); + assert.match( + source, + /registerTokenHealthCheck\s*\(/, + `${relativePath} must register the token health check job` + ); + } +}); diff --git a/tests/unit/token-health-check-sweep.test.ts b/tests/unit/token-health-check-sweep.test.ts index 87e2d548a4..afce89cde8 100644 --- a/tests/unit/token-health-check-sweep.test.ts +++ b/tests/unit/token-health-check-sweep.test.ts @@ -104,9 +104,13 @@ test("sweep() skips re-entrant calls while a previous sweep is still in flight", assert.equal(isSweeping(), true, "first sweep should mark sweeping=true immediately"); const start = Date.now(); - await sweep(); // second, concurrent call — must be skipped by the guard + const skippedCount = await sweep(); // second, concurrent call — must be skipped by the guard const elapsedMs = Date.now() - start; + // The job registry records this as records_affected, so a skipped sweep has + // to report nothing rather than inheriting the in-flight sweep's count. + assert.equal(skippedCount, 0, "a skipped sweep reports 0 connections swept"); + // With 21 connections and a 300ms inter-batch stagger, a REAL second // sweep would take >= 300ms to clear the single batch gap. The guard must // make this call return near- @@ -130,8 +134,9 @@ test("sweep() resets the sweeping flag after a normal completion, allowing the n process.env.HEALTHCHECK_STAGGER_MS = "0"; await createNoOpOauthConnections(2, "sequential"); - await sweep(); + const sweptCount = await sweep(); assert.equal(isSweeping(), false, "flag resets after first sweep completes"); + assert.equal(sweptCount, 2, "sweep reports how many connections it swept"); // A second, fully sequential call must not be skipped by leftover state — // it should run and complete normally rather than hang or short-circuit. @@ -144,7 +149,8 @@ test("sweep() resets the sweeping flag even when there are no connections to pro process.env.HEALTHCHECK_STAGGER_MS = "0"; // No connections created — sweep() takes the early `connections.length === 0` return. - await sweep(); + const emptyCount = await sweep(); assert.equal(isSweeping(), false, "sweeping flag must be false after an empty sweep"); + assert.equal(emptyCount, 0, "an empty sweep reports 0 connections swept"); });