diff --git a/apps/hub/src/config.ts b/apps/hub/src/config.ts index dd2c0e48f..adb73fc8e 100644 --- a/apps/hub/src/config.ts +++ b/apps/hub/src/config.ts @@ -101,6 +101,9 @@ const HubEnv = type({ "WORKBENCH_ALLOWED_EMAIL_DOMAINS?": type("string").describe( "comma-separated email domains allowed when WORKBENCH_SIGNUP=open, e.g. acme.example", ), + "ROUTINE_SCHEDULER_POLL_INTERVAL_MS?": type(/^[1-9]\d*$/).describe( + "dev/test-only override for the routine scheduler's poll interval, in milliseconds — unset (default) runs the real 30s production cadence; the e2e harness sets this to a fast interval so a scheduled-routine test doesn't wait out the real cadence", + ), "ANTHROPIC_API_KEY?": type("string > 0").describe( "your Anthropic API key; optional, enables the default workflow set for freshly self-served benches, and auto-plants a probed catalog credential on the operator bench at hub start", ), @@ -354,6 +357,10 @@ export type HubConfig = { /** Dev/test-only opt-in to skip @workbench/access-policy's email- * verification requirement. */ readonly allowUnverifiedEmails: boolean; + /** Dev/test-only override for the routine scheduler's poll interval — + * see `ROUTINE_SCHEDULER_POLL_INTERVAL_MS` above. Unset runs the real + * production cadence (`routine-scheduler.ts`'s own default). */ + readonly routineSchedulerPollIntervalMs?: number; /** Every sidecar-allocation backend registered for exclusive placement, * zero or more, each addressable by its provisioner id. Empty when * unconfigured — the registry then holds no provisioners and every @@ -660,6 +667,10 @@ export function readHubConfig( if (parsed.OPERATOR_TENANT_ID !== undefined) hubConfig.operatorTenantId = parsed.OPERATOR_TENANT_ID; if (parsed.PORT !== undefined) hubConfig.listenPort = Number(parsed.PORT); + if (parsed.ROUTINE_SCHEDULER_POLL_INTERVAL_MS !== undefined) + hubConfig.routineSchedulerPollIntervalMs = Number( + parsed.ROUTINE_SCHEDULER_POLL_INTERVAL_MS, + ); if (seedModel !== undefined) hubConfig.seedModel = seedModel; if (parsed.HUGGINGFACE_OAUTH_CLIENT_ID !== undefined) hubConfig.huggingfaceOAuthClientId = parsed.HUGGINGFACE_OAUTH_CLIENT_ID; diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index c277a9f73..6c271a570 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -2948,6 +2948,9 @@ export async function createHub(config: HubConfig) { store: routineStore, launcher: routineLauncher, deliveryWorkbenchRequired: routineDeliveryWorkbenchRequired, + ...(config.routineSchedulerPollIntervalMs !== undefined + ? { pollIntervalMs: config.routineSchedulerPollIntervalMs } + : {}), }); // The inventory Myra is offered when drafting a new agent definition diff --git a/apps/hub/src/routine-scheduler.ts b/apps/hub/src/routine-scheduler.ts index 81a106a6f..cb2ac28ee 100644 --- a/apps/hub/src/routine-scheduler.ts +++ b/apps/hub/src/routine-scheduler.ts @@ -45,9 +45,15 @@ export type RoutineSchedulerDeps = { ) => Promise; /** Injectable for deterministic tests; defaults to `Date.now`-backed wall time. */ now?: () => Date; + /** How often the poller sweeps for due routines. Defaults to + * `DEFAULT_ROUTINE_SCHEDULER_POLL_INTERVAL_MS`. Test/dev-only override + * (see `ROUTINE_SCHEDULER_POLL_INTERVAL_MS` in `config.ts`) — a live + * hub always runs the real cadence unless that env var is set, which + * only the e2e harness does. */ + pollIntervalMs?: number; }; -const POLL_INTERVAL_MS = 30_000; +export const DEFAULT_ROUTINE_SCHEDULER_POLL_INTERVAL_MS = 30_000; const log = getLogger(["hub", "routine-scheduler"]); /** @@ -131,7 +137,10 @@ export function createRoutineScheduler(deps: RoutineSchedulerDeps) { } } - const interval = setInterval(() => void tick(), POLL_INTERVAL_MS); + const interval = setInterval( + () => void tick(), + deps.pollIntervalMs ?? DEFAULT_ROUTINE_SCHEDULER_POLL_INTERVAL_MS, + ); if (typeof interval.unref === "function") interval.unref(); return { diff --git a/apps/hub/test/config.test.ts b/apps/hub/test/config.test.ts index df27ff071..72052fe53 100644 --- a/apps/hub/test/config.test.ts +++ b/apps/hub/test/config.test.ts @@ -184,6 +184,18 @@ describe("readHubConfig", () => { expect(config.signupRateLimit).toEqual({ windowSeconds: 30, max: 2 }); }); + test("ROUTINE_SCHEDULER_POLL_INTERVAL_MS is optional and absent by default", () => { + expect( + readHubConfig(validEnv).routineSchedulerPollIntervalMs, + ).toBeUndefined(); + expect( + readHubConfig({ + ...validEnv, + ROUTINE_SCHEDULER_POLL_INTERVAL_MS: "500", + }).routineSchedulerPollIntervalMs, + ).toBe(500); + }); + test("the sign-in rate limit is configurable and defaults well above better-auth's built-in 3-per-10-seconds special rule (CL-6494)", () => { expect(readHubConfig(validEnv).signInRateLimit).toEqual({ windowSeconds: 60, diff --git a/apps/hub/test/routine-scheduler.test.ts b/apps/hub/test/routine-scheduler.test.ts index f6c3d523f..2970b73da 100644 --- a/apps/hub/test/routine-scheduler.test.ts +++ b/apps/hub/test/routine-scheduler.test.ts @@ -1,12 +1,16 @@ // Scheduler poller: claim → fire, backoff on failure, dead-letter at max. -import { describe, expect, test } from "bun:test"; +import { afterEach, describe, expect, jest, test } from "bun:test"; import { backoffMsForFailure, createInMemoryRoutineStore, MAX_ROUTINE_FIRE_FAILURES, type RoutineLauncher, } from "@corbits/routines"; -import { tickRoutineScheduler } from "../src/routine-scheduler"; +import { + createRoutineScheduler, + DEFAULT_ROUTINE_SCHEDULER_POLL_INTERVAL_MS, + tickRoutineScheduler, +} from "../src/routine-scheduler"; const CRON = { kind: "cron" as const, expression: "0 * * * *" }; @@ -236,3 +240,88 @@ describe("tickRoutineScheduler", () => { ).toEqual([]); }); }); + +// `createRoutineScheduler`'s own `setInterval` wrapper — the piece an +// e2e test cannot drive with fake timers because it lives in a spawned +// child process (see CL-7250). Proven here instead, entirely +// in-process with Bun's fake timers, so the real 30s production +// cadence and the injectable override are both covered in +// milliseconds rather than by waiting either one out for real. +describe("createRoutineScheduler's setInterval wiring", () => { + afterEach(() => { + jest.useRealTimers(); + }); + + test("defaults to the real 30s cadence when pollIntervalMs is unset", async () => { + jest.useFakeTimers(); + const store = createInMemoryRoutineStore(); + await store.createRoutine({ + tenantId: "t1", + name: "hourly", + definitionId: "def_1", + trigger: CRON, + scope: "bench", + input: {}, + deliveryWorkbenchId: "ch_delivery", + createdBy: "user_1", + }); + let launches = 0; + const scheduler = createRoutineScheduler({ + store, + launcher: launcher(async () => { + launches += 1; + return { runId: "run_1" }; + }), + now: () => new Date(Date.now() + 365 * 24 * 60 * 60 * 1000), + }); + try { + jest.advanceTimersByTime(DEFAULT_ROUTINE_SCHEDULER_POLL_INTERVAL_MS - 1); + await Promise.resolve(); + expect(launches).toBe(0); + + jest.advanceTimersByTime(1); + await Promise.resolve(); + await Promise.resolve(); + expect(launches).toBe(1); + } finally { + scheduler.stop(); + } + }); + + test("an injected pollIntervalMs overrides the default cadence", async () => { + jest.useFakeTimers(); + const store = createInMemoryRoutineStore(); + await store.createRoutine({ + tenantId: "t1", + name: "hourly", + definitionId: "def_1", + trigger: CRON, + scope: "bench", + input: {}, + deliveryWorkbenchId: "ch_delivery", + createdBy: "user_1", + }); + let launches = 0; + const scheduler = createRoutineScheduler({ + store, + launcher: launcher(async () => { + launches += 1; + return { runId: "run_1" }; + }), + now: () => new Date(Date.now() + 365 * 24 * 60 * 60 * 1000), + pollIntervalMs: 500, + }); + try { + jest.advanceTimersByTime(499); + await Promise.resolve(); + expect(launches).toBe(0); + + jest.advanceTimersByTime(1); + await Promise.resolve(); + await Promise.resolve(); + expect(launches).toBe(1); + } finally { + scheduler.stop(); + } + }); +}); diff --git a/scripts/e2e/greeting-delivery.test.ts b/scripts/e2e/greeting-delivery.test.ts index aebb9f596..80e894821 100644 --- a/scripts/e2e/greeting-delivery.test.ts +++ b/scripts/e2e/greeting-delivery.test.ts @@ -258,7 +258,7 @@ describe.skipIf(databaseUrl === undefined)( break; } catch (cause) { if (Date.now() > deadline) throw cause; - await Bun.sleep(1000); + await Bun.sleep(200); } } }, @@ -307,7 +307,7 @@ describe.skipIf(databaseUrl === undefined)( return; } catch (cause) { if (Date.now() > deadline) throw cause; - await Bun.sleep(1000); + await Bun.sleep(200); } } } @@ -343,7 +343,7 @@ describe.skipIf(databaseUrl === undefined)( `"assistant" never appeared as invitable: ${JSON.stringify(res.data)}`, ); } - await Bun.sleep(1000); + await Bun.sleep(200); } }, ); @@ -373,7 +373,7 @@ describe.skipIf(databaseUrl === undefined)( `${JSON.stringify(res.data)}\nsidecar output:\n${sidecar.output()}`, ); } - await Bun.sleep(1000); + await Bun.sleep(200); } expectStatus("create chat", res, 201); const id = stringField(res.data, "id", "create chat"); @@ -449,7 +449,7 @@ describe.skipIf(databaseUrl === undefined)( `${JSON.stringify(items)}\nsidecar output:\n${sidecar.output()}`, ); } - await Bun.sleep(1000); + await Bun.sleep(200); } }, ); diff --git a/scripts/e2e/harness.ts b/scripts/e2e/harness.ts index a4d7f0caf..0b40b031d 100644 --- a/scripts/e2e/harness.ts +++ b/scripts/e2e/harness.ts @@ -354,6 +354,15 @@ export async function startHub(options: { WORKBENCH_SIGNUP: "open", HUB_STATIC_DIR: "public", BASE_URL: baseUrl, + // The routine scheduler's real production cadence is a 30s + // setInterval (routine-scheduler.ts); waiting that out for real on + // every e2e test that touches a scheduled routine fire is exactly + // the sleep-dominated cost CL-7250 exists to cut. The default and + // the override are both proven in-process with fake timers in + // apps/hub/test/routine-scheduler.test.ts, so no e2e test needs the + // real cadence — a caller's extraEnv can still opt back into it by + // omitting or overriding this key. + ROUTINE_SCHEDULER_POLL_INTERVAL_MS: "300", ...options.extraEnv, DATABASE_URL: options.databaseUrl, // Always the real bind port, independent of whatever port BASE_URL's diff --git a/scripts/e2e/heartbeat.test.ts b/scripts/e2e/heartbeat.test.ts index f53998ffc..8d55adec9 100644 --- a/scripts/e2e/heartbeat.test.ts +++ b/scripts/e2e/heartbeat.test.ts @@ -236,7 +236,7 @@ describe.skipIf(databaseUrl === undefined)("heartbeat workflow", () => { `${JSON.stringify(res.data)}\nsidecar output:\n${sidecar.output()}`, ); } - await Bun.sleep(1000); + await Bun.sleep(200); } expectStatus("deploy heartbeat workflow", res, 201); return stringField(res.data, "id", "deploy heartbeat workflow"); @@ -284,7 +284,7 @@ describe.skipIf(databaseUrl === undefined)("heartbeat workflow", () => { "heartbeat trigger was accepted but no run started within 30s", ); } - await Bun.sleep(500); + await Bun.sleep(200); } }, ); diff --git a/scripts/e2e/local-rip.test.ts b/scripts/e2e/local-rip.test.ts index 320b74d8c..8684c9276 100644 --- a/scripts/e2e/local-rip.test.ts +++ b/scripts/e2e/local-rip.test.ts @@ -367,7 +367,7 @@ describe.skipIf(databaseUrl === undefined)( }); } catch (cause) { if (Date.now() > deadline) throw cause; - await Bun.sleep(1000); + await Bun.sleep(200); } } } @@ -414,7 +414,7 @@ describe.skipIf(databaseUrl === undefined)( break; } catch (cause) { if (Date.now() > deadline) throw cause; - await Bun.sleep(1000); + await Bun.sleep(200); } } }, diff --git a/scripts/e2e/routine-repeat.test.ts b/scripts/e2e/routine-repeat.test.ts index d466058ec..f13a5d0e7 100644 --- a/scripts/e2e/routine-repeat.test.ts +++ b/scripts/e2e/routine-repeat.test.ts @@ -271,7 +271,7 @@ describe.skipIf(databaseUrl === undefined)("routine repeat fires", () => { `${JSON.stringify(res.data)}\nsidecar output:\n${sidecar.output()}`, ); } - await Bun.sleep(1000); + await Bun.sleep(200); continue; } expectStatus("deploy heartbeat workflow", res, 201); diff --git a/scripts/e2e/routine-trigger-input.test.ts b/scripts/e2e/routine-trigger-input.test.ts index 270fbf5c9..8915364cc 100644 --- a/scripts/e2e/routine-trigger-input.test.ts +++ b/scripts/e2e/routine-trigger-input.test.ts @@ -328,7 +328,7 @@ describe.skipIf(databaseUrl === undefined)( `sidecar never became deployable (hub kept answering 502): ${JSON.stringify(res.data)}\nsidecar output:\n${sidecar.output()}`, ); } - await Bun.sleep(1000); + await Bun.sleep(200); continue; } expectStatus("deploy heartbeat workflow", res, 201); @@ -406,7 +406,7 @@ describe.skipIf(databaseUrl === undefined)( while (Date.now() < deadline) { mailbox = await readMailbox(url, runId); if (mailbox.length > 0) break; - await Bun.sleep(500); + await Bun.sleep(200); } if ( @@ -443,8 +443,9 @@ describe.skipIf(databaseUrl === undefined)( const routineId = stringField(routine.data, "id", "create routine"); // Force the routine due immediately rather than waiting out its - // own cadence — the scheduler poll interval (30s) is the only - // wait this needs. + // own cadence — the only wait left is the scheduler's own poll + // interval, which the e2e harness sets to 300ms (CL-7250), not + // the real 30s production cadence. const sql = await connectE2eDb(url); try { await sql.unsafe( @@ -471,7 +472,7 @@ describe.skipIf(databaseUrl === undefined)( runId = items[0]?.runId; break; } - await Bun.sleep(1000); + await Bun.sleep(200); } if (runId === undefined) { throw new Error( @@ -485,7 +486,7 @@ describe.skipIf(databaseUrl === undefined)( while (Date.now() < mailboxDeadline) { mailbox = await readMailbox(url, runId); if (mailbox.length > 0) break; - await Bun.sleep(500); + await Bun.sleep(200); } if ( @@ -572,7 +573,7 @@ describe.skipIf(databaseUrl === undefined)( while (Date.now() < deadline) { mailbox = await readMailbox(url, deliveredData.instanceId); if (mailbox.length > 0) break; - await Bun.sleep(500); + await Bun.sleep(200); } if ( diff --git a/scripts/e2e/smoke-webhook.test.ts b/scripts/e2e/smoke-webhook.test.ts index c7746496a..c11cecb50 100644 --- a/scripts/e2e/smoke-webhook.test.ts +++ b/scripts/e2e/smoke-webhook.test.ts @@ -285,7 +285,7 @@ describe.skipIf(databaseUrl === undefined)("smoke: webhook trigger", () => { `sidecar never became deployable (hub kept answering 502): ${JSON.stringify(res.data)}\nsidecar output:\n${sidecar.output()}`, ); } - await Bun.sleep(1000); + await Bun.sleep(200); continue; } expectStatus("deploy heartbeat workflow", res, 201); @@ -515,7 +515,7 @@ describe.skipIf(databaseUrl === undefined)("smoke: webhook trigger", () => { `run ${instanceId} recorded no inference_turn within the deadline`, ); } - await Bun.sleep(500); + await Bun.sleep(200); } } finally { await sql.end(); diff --git a/scripts/e2e/walking-skeleton.test.ts b/scripts/e2e/walking-skeleton.test.ts index feeb26def..d0799f3ef 100644 --- a/scripts/e2e/walking-skeleton.test.ts +++ b/scripts/e2e/walking-skeleton.test.ts @@ -259,7 +259,7 @@ describe.skipIf(databaseUrl === undefined)("walking skeleton", () => { `${JSON.stringify(res.data)}\nsidecar output:\n${sidecar.output()}`, ); } - await Bun.sleep(1000); + await Bun.sleep(200); } expectStatus("deploy workflow", res, 201); return stringField(res.data, "id", "deploy workflow"); diff --git a/scripts/e2e/workbench-digest.test.ts b/scripts/e2e/workbench-digest.test.ts index f9eb3c00a..212a73824 100644 --- a/scripts/e2e/workbench-digest.test.ts +++ b/scripts/e2e/workbench-digest.test.ts @@ -231,7 +231,7 @@ describe.skipIf(databaseUrl === undefined)("workbench-digest workflow", () => { `${JSON.stringify(res.data)}\nsidecar output:\n${sidecar.output()}`, ); } - await Bun.sleep(1000); + await Bun.sleep(200); } expectStatus("deploy workbench-digest workflow", res, 201); return stringField(res.data, "id", "deploy workbench-digest workflow"); @@ -279,7 +279,7 @@ describe.skipIf(databaseUrl === undefined)("workbench-digest workflow", () => { "workbench-digest trigger was accepted but no run started within 30s", ); } - await Bun.sleep(500); + await Bun.sleep(200); } }, );