From dfa95262734a03be989ce6e1ca6e9be396510728 Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Thu, 5 Nov 2026 13:29:32 -0600 Subject: [PATCH 1/6] test: drive the run-executor watchdog tests with injected timers run-executor-timeout-honesty.test.ts armed the real watchdog with a 400 ms no-progress budget. The watchdog is armed before the connector subprocess is spawned, so that budget also had to cover node startup, the START handshake and the ingest round-trip. On a loaded host those exceeded it: the run was killed with run_timed_out before its records were ingested, so the records_emitted: 5 and "survives past the budget" tests failed (16 of 16 parallel runs under CPU load; the file passes alone). Let the attempt watchdog take its timer functions: createAttemptWatchdog and RunExecutorDeps gain an optional `watchdogTimers`, defaulting to the global timers, so production behaviour is unchanged. The tests now inject a manual timer, wait for the event they assert on (records durably ingested, the local-only boundary latched), assert the delay the watchdog armed (one budget, or the four-budget ceiling after the boundary) and fire it explicitly. The phase-boundary connector waits for a release marker instead of sleeping for a multiple of the budget. Under CPU load (3 busy loops per core, 16 parallel runs) the old file failed in 16 of 16 runs; the new one failed in 0 of 48. Replacing the ceiling with the primary budget in run-executor.ts makes the boundary tests fail. Assisted-by: AI Signed-off-by: Tim Nunamaker --- .../runtime/scheduler/run-executor.ts | 37 ++- .../test/run-executor-timeout-honesty.test.ts | 278 ++++++++++++------ 2 files changed, 215 insertions(+), 100 deletions(-) diff --git a/reference-implementation/runtime/scheduler/run-executor.ts b/reference-implementation/runtime/scheduler/run-executor.ts index b030e6f20..7f39c1380 100644 --- a/reference-implementation/runtime/scheduler/run-executor.ts +++ b/reference-implementation/runtime/scheduler/run-executor.ts @@ -115,6 +115,13 @@ export interface RunExecutorDeps { | null | undefined; setState: SetStateHandler; + /** + * Timer functions the attempt watchdog arms. Optional: defaults to the + * global timers, so callers that do not care are unaffected. A test injects a + * manual implementation to fire the watchdog at a chosen point instead of + * racing a real wall-clock budget against process startup. + */ + watchdogTimers?: WatchdogTimers; } // ─── Public interface ───────────────────────────────────────────────────────── @@ -356,6 +363,16 @@ async function invokeRunConnector(call: RunConnectorCall): Promise void; + setTimeout: (callback: () => void, ms: number) => NodeJS.Timeout; +} + +const GLOBAL_WATCHDOG_TIMERS: WatchdogTimers = { + clearTimeout: (timer) => clearTimeout(timer), + setTimeout: (callback, ms) => setTimeout(callback, ms), +}; + interface AttemptWatchdog { cancel: () => void; clear: () => void; @@ -383,7 +400,10 @@ function isBoundedWallClock(maxRunWallClockMs: number): boolean { return Number.isFinite(maxRunWallClockMs) && maxRunWallClockMs > 0; } -export function createAttemptWatchdog(maxRunWallClockMs: number): AttemptWatchdog { +export function createAttemptWatchdog( + maxRunWallClockMs: number, + timers: WatchdogTimers = GLOBAL_WATCHDOG_TIMERS +): AttemptWatchdog { const cancellation = new AbortController(); let timedOut = false; let timer: NodeJS.Timeout | null = null; @@ -409,9 +429,9 @@ export function createAttemptWatchdog(maxRunWallClockMs: number): AttemptWatchdo return; } if (timer) { - clearTimeout(timer); + timers.clearTimeout(timer); } - timer = setTimeout(() => { + timer = timers.setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); }, maxRunWallClockMs); @@ -427,9 +447,9 @@ export function createAttemptWatchdog(maxRunWallClockMs: number): AttemptWatchdo return; } if (timer) { - clearTimeout(timer); + timers.clearTimeout(timer); } - timer = setTimeout(() => { + timer = timers.setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); }, localOnlyPhaseMaxRunWallClockMs); @@ -448,7 +468,7 @@ export function createAttemptWatchdog(maxRunWallClockMs: number): AttemptWatchdo }, clear() { if (timer) { - clearTimeout(timer); + timers.clearTimeout(timer); timer = null; } }, @@ -456,7 +476,7 @@ export function createAttemptWatchdog(maxRunWallClockMs: number): AttemptWatchdo if (extra?.phase_boundary === "local_only_phase_started" && !localOnlyPhase) { localOnlyPhase = true; if (timer) { - clearTimeout(timer); + timers.clearTimeout(timer); } armLocalOnlyPhaseCeiling(); // TRUE only on the latching call — the caller emits the durable trace. @@ -772,6 +792,7 @@ export function createRunExecutor(deps: RunExecutorDeps): RunExecutor { runManagedConnectorViaController, schedulerStore, setState, + watchdogTimers, } = deps; async function appendRunHistoryBestEffort(record: RunRecord, connectorId: string): Promise { @@ -1046,7 +1067,7 @@ export function createRunExecutor(deps: RunExecutorDeps): RunExecutor { const { connectorId, connectorInstanceId = connectorId } = schedule; const originalOnProgress = call.onProgress; const activeRunStore = selectActiveRunStore(); - const watchdog = createAttemptWatchdog(maxRunWallClockMs); + const watchdog = createAttemptWatchdog(maxRunWallClockMs, watchdogTimers); // A cryptographically strong UUID keeps this fallback globally // collision-resistant: run_id is the sole uniqueness key for the durable // STREAM_EVIDENCE claim registry, and a millisecond-granularity fallback diff --git a/reference-implementation/test/run-executor-timeout-honesty.test.ts b/reference-implementation/test/run-executor-timeout-honesty.test.ts index d207bed27..06c33cc41 100644 --- a/reference-implementation/test/run-executor-timeout-honesty.test.ts +++ b/reference-implementation/test/run-executor-timeout-honesty.test.ts @@ -25,7 +25,7 @@ // ingest round-trip that production hit. import assert from "node:assert/strict"; -import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import test from "node:test"; @@ -35,6 +35,7 @@ import { createRunExecutor, type RunExecutorDeps, type RunExecutorRuntimeState, + type WatchdogTimers, } from "../runtime/scheduler/run-executor.ts"; import type { ConnectorSchedule } from "../runtime/scheduler-domain-types.ts"; import { getDb } from "../server/db.ts"; @@ -146,30 +147,75 @@ const SCHEDULED_POLICY = projectRunAutomationPolicy({ refreshPolicy: null, trigg // `maxRunWallClockMs` is a RESETTABLE no-progress allowance, not a total-run // budget: `createAttemptWatchdog`'s `markProgress` re-arms the timer on every -// PROGRESS message (run-executor.ts `arm()`). The watchdog is armed at -// construction in `createActiveRunAttemptLease`, BEFORE the connector -// subprocess is spawned, so the FIRST allowance must cover `node` startup, the -// START handshake and the ingest round-trip together. +// PROGRESS message (run-executor.ts `arm()`), and a `local_only_phase_started` +// boundary replaces it with a ceiling of four allowances. // -// Measured on a 24-core host (the run-executor's own `elapsed_ms` at the -// phase-boundary latch — total attempt time, spawn included): 84-124 ms idle, -// up to 148 ms under load. Against the previous 150 ms allowance that is a 2 ms -// margin, which is why this file was the one remaining load-sensitive test in -// the suite. -// -// This raises the no-progress allowance to 400 ms and resets it on connector -// readiness; the completion case stays at three allowances and the secondary -// ceiling at four. Startup remains subject to the initial allowance — the ready -// message widens the allowance available AFTER it, it does not make startup -// unbounded or remove timing sensitivity. -// -// Deadline-honesty is unweakened: every "must time out" test still hangs -// forever with no further progress, and the phase-boundary test's post-boundary -// sleep is still a multiple of the allowance. +// These tests do not race that allowance against the clock. Process startup, +// the START handshake and the ingest round-trip take a host-dependent time, so +// any real budget either flakes on a loaded host or makes the test slow. The +// watchdog's timers are injected (`WatchdogTimers`): the test waits for the +// event it cares about (records durably ingested, the phase boundary latched), +// asserts the delay the watchdog armed, and fires it explicitly. The value +// below only has to be recognisable as the armed delay. const WALL_CLOCK_BUDGET_MS = 400; -// The post-boundary sleep must be unambiguously longer than the budget, so a -// re-armed (not disarmed) watchdog would still kill the run and fail the test. -const POST_BOUNDARY_SLEEP_MS = WALL_CLOCK_BUDGET_MS * 3; +const LOCAL_ONLY_CEILING_MS = WALL_CLOCK_BUDGET_MS * 4; +const CONNECTOR_SENT_MARKER = "connector-sent"; +const CONNECTOR_RELEASE_MARKER = "connector-release"; +const WAIT_FOR_TIMEOUT_MS = 30_000; + +interface ManualWatchdogTimers { + /** Delay of the single armed timer, or null when none is armed. */ + armedDelayMs: () => number | null; + /** Runs the single armed timer's callback, as if its delay had elapsed. */ + fire: () => void; + timers: WatchdogTimers; +} + +function manualWatchdogTimers(): ManualWatchdogTimers { + const armed = new Map void; ms: number }>(); + let nextId = 0; + return { + armedDelayMs: () => { + assert.ok(armed.size <= 1, `the watchdog arms at most one timer at a time, got ${armed.size}`); + const [only] = [...armed.values()]; + return only ? only.ms : null; + }, + fire: () => { + assert.equal(armed.size, 1, "expected exactly one armed watchdog timer to fire"); + const [entry] = [...armed.entries()]; + assert.ok(entry, "expected an armed watchdog timer"); + const [id, timer] = entry; + armed.delete(id); + timer.callback(); + }, + timers: { + clearTimeout: (handle) => { + armed.delete((handle as unknown as { id: number }).id); + }, + setTimeout: (callback, ms) => { + nextId += 1; + armed.set(nextId, { callback, ms }); + return { + id: nextId, + unref() { + return this; + }, + } as unknown as NodeJS.Timeout; + }, + }, + }; +} + +async function waitFor(condition: () => boolean, what: string): Promise { + const deadline = Date.now() + WAIT_FOR_TIMEOUT_MS; + while (!condition()) { + if (Date.now() > deadline) { + assert.fail(`timed out waiting for ${what}`); + } + // biome-ignore lint/performance/noAwaitInLoops: Polling is sequential by nature. + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} /** * Writes a connector that emits `recordCount` RECORD messages + a STATE @@ -179,6 +225,7 @@ const POST_BOUNDARY_SLEEP_MS = WALL_CLOCK_BUDGET_MS * 3; * and exits immediately (used as the "genuinely completed" control). */ function writeConnector(tmpDir: string, recordCount: number, hang: boolean): string { + const sentMarker = join(tmpDir, CONNECTOR_SENT_MARKER); const connectorPath = join(tmpDir, "connector.mjs"); const records = Array.from({ length: recordCount }, (_, i) => ({ data: { id: `i${i}` }, @@ -190,6 +237,7 @@ function writeConnector(tmpDir: string, recordCount: number, hang: boolean): str writeFileSync( connectorPath, ` +import { writeFileSync } from "node:fs"; import { createInterface } from "node:readline"; const rl = createInterface({ input: process.stdin, crlfDelay: Infinity }); rl.on("line", (line) => { @@ -205,8 +253,9 @@ rl.on("line", (line) => { } process.stdout.write(JSON.stringify({ cursor: { last: records.length ? records[records.length - 1].key : null }, stream: "items", type: "STATE" }) + "\\n"); if (${hang ? "true" : "false"}) { - // Never send DONE, never exit — only the scheduler watchdog's - // maxRunWallClockMs setTimeout can end this run. + // Never send DONE, never exit — only the scheduler watchdog can end this + // run. The marker tells the test every message has been written. + writeFileSync(${JSON.stringify(sentMarker)}, ""); setInterval(() => {}, 1000); } else { process.stdout.write(JSON.stringify({ records_emitted: records.length, status: "succeeded", type: "DONE" }) + "\\n"); @@ -222,19 +271,14 @@ rl.on("line", (line) => { /** * Writes a connector that emits `recordCount` records, then sends a PROGRESS - * message with `phase_boundary: "local_only_phase_started"`, then sleeps for - * `sleepMs` (simulating local-only work with no external dependency) before - * completing successfully. Used to prove the watchdog stops applying - * `maxRunWallClockMs` once a connector declares it has left the - * provider-rate-limited phase — `sleepMs` is deliberately set well past - * `maxRunWallClockMs` in the test that uses this. + * message with `phase_boundary: "local_only_phase_started"`, then waits until + * the test releases it (simulating local-only work with no external + * dependency) and it completes successfully. Used to prove the watchdog stops + * applying `maxRunWallClockMs` once a connector declares it has left the + * provider-rate-limited phase. */ -function writeConnectorWithPhaseBoundary( - tmpDir: string, - recordCount: number, - sleepMs: number, - hangAfterBoundary = false -): string { +function writeConnectorWithPhaseBoundary(tmpDir: string, recordCount: number, hangAfterBoundary = false): string { + const releaseMarker = join(tmpDir, CONNECTOR_RELEASE_MARKER); const connectorPath = join(tmpDir, "connector.mjs"); const records = Array.from({ length: recordCount }, (_, i) => ({ data: { id: `i${i}` }, @@ -246,6 +290,7 @@ function writeConnectorWithPhaseBoundary( writeFileSync( connectorPath, ` +import { existsSync } from "node:fs"; import { createInterface } from "node:readline"; const rl = createInterface({ input: process.stdin, crlfDelay: Infinity }); rl.on("line", async (line) => { @@ -267,7 +312,10 @@ rl.on("line", async (line) => { setInterval(() => {}, 1000); return; } - await new Promise((resolve) => setTimeout(resolve, ${sleepMs})); + // Local-only work: finish when the test releases it, not after a wall-clock sleep. + while (!existsSync(${JSON.stringify(releaseMarker)})) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } process.stdout.write(JSON.stringify({ records_emitted: records.length, status: "succeeded", type: "DONE" }) + "\\n"); rl.close(); process.exit(0); @@ -319,7 +367,8 @@ function schedule(connectorPath: string, ownerToken: string): ConnectorSchedule function makeHarness( runtime: RunExecutorRuntimeState, rsUrl: string, - maxRunWallClockMs: number + maxRunWallClockMs: number, + watchdogTimers?: WatchdogTimers ): ReturnType["launchRun"] { const deps: RunExecutorDeps = { admitRunConnection: async ({ connectorId, connectorInstanceId, ownerSubjectId }) => ({ @@ -357,6 +406,7 @@ function makeHarness( setState: async () => { // State persistence is out of scope for this oracle. }, + ...(watchdogTimers ? { watchdogTimers } : {}), }; return createRunExecutor(deps).launchRun; } @@ -388,19 +438,51 @@ function withServerAndTmpDir( }; } +function ingestedRecordCount(): number { + const row = getDb().prepare("SELECT count(*) AS n FROM records WHERE connector_id = ?").get(CONNECTOR_ID) as { + n: number; + }; + return Number(row.n); +} + +function localOnlyPhaseLatched(): boolean { + const row = getDb() + .prepare("SELECT count(*) AS n FROM spine_events WHERE event_type = 'run.local_only_phase_started'") + .get() as { n: number }; + return Number(row.n) > 0; +} + +/** + * Launches a connector that emits `recordCount` records and then hangs, waits + * until every record is durably ingested, and fires the watchdog's armed + * timer. Returns the terminal record and the delay the watchdog had armed. + */ +async function runHangingConnectorUntilWatchdogFires( + ctx: { ownerToken: string; rsUrl: string; tmpDir: string }, + recordCount: number +) { + const watchdog = manualWatchdogTimers(); + const launchRun = makeHarness(freshRuntime(), ctx.rsUrl, WALL_CLOCK_BUDGET_MS, watchdog.timers); + const pending = launchRun( + schedule(writeConnector(ctx.tmpDir, recordCount, true), ctx.ownerToken), + false, + SCHEDULED_POLICY + ); + await waitFor(() => existsSync(join(ctx.tmpDir, CONNECTOR_SENT_MARKER)), "the connector to write all its messages"); + await waitFor(() => ingestedRecordCount() === recordCount, `${recordCount} records to be durably ingested`); + const armedDelayMs = watchdog.armedDelayMs(); + watchdog.fire(); + return { armedDelayMs, record: await pending }; +} + test( "a run that durably emits N records before being killed by the wall-clock watchdog reports records_emitted: N, not 0", - withServerAndTmpDir(async ({ ownerToken, rsUrl, tmpDir }) => { - const runtime = freshRuntime(); - // The allowance resets on the connector's "ready" PROGRESS, so the ingest - // round-trip for 5 records gets a full allowance of its own rather than - // sharing one with startup. It is still short enough that the test doesn't - // hang; the connector hangs forever after emitting, so ONLY the watchdog - // can end this run. - const launchRun = makeHarness(runtime, rsUrl, WALL_CLOCK_BUDGET_MS); - - const record = await launchRun(schedule(writeConnector(tmpDir, 5, true), ownerToken), false, SCHEDULED_POLICY); + withServerAndTmpDir(async (ctx) => { + // The connector hangs forever after emitting, so ONLY the watchdog can end + // this run; it is fired once all 5 records are durably ingested. + const { armedDelayMs, record } = await runHangingConnectorUntilWatchdogFires(ctx, 5); + assert.equal(armedDelayMs, WALL_CLOCK_BUDGET_MS); assert.equal(record.status, "failed"); assert.equal(record.terminalReason, "run_timed_out"); assert.equal( @@ -413,11 +495,8 @@ test( test( "a run that emits zero records before being killed by the watchdog still reports records_emitted: 0 (no fabrication in either direction)", - withServerAndTmpDir(async ({ ownerToken, rsUrl, tmpDir }) => { - const runtime = freshRuntime(); - const launchRun = makeHarness(runtime, rsUrl, WALL_CLOCK_BUDGET_MS); - - const record = await launchRun(schedule(writeConnector(tmpDir, 0, true), ownerToken), false, SCHEDULED_POLICY); + withServerAndTmpDir(async (ctx) => { + const { record } = await runHangingConnectorUntilWatchdogFires(ctx, 0); assert.equal(record.status, "failed"); assert.equal(record.terminalReason, "run_timed_out"); @@ -431,11 +510,8 @@ test( test( "a run killed by the watchdog with unfinished work records a typed gap, not an empty known_gaps array", - withServerAndTmpDir(async ({ ownerToken, rsUrl, tmpDir }) => { - const runtime = freshRuntime(); - const launchRun = makeHarness(runtime, rsUrl, WALL_CLOCK_BUDGET_MS); - - const record = await launchRun(schedule(writeConnector(tmpDir, 5, true), ownerToken), false, SCHEDULED_POLICY); + withServerAndTmpDir(async (ctx) => { + const { record } = await runHangingConnectorUntilWatchdogFires(ctx, 5); assert.equal(record.status, "failed"); assert.ok(Array.isArray(record.knownGaps), "knownGaps must be an array"); @@ -466,20 +542,32 @@ test( test( "a connector that declares a local-only phase boundary is not truncated by maxRunWallClockMs, even though it runs well past that budget", withServerAndTmpDir(async ({ ownerToken, rsUrl, tmpDir }) => { - const runtime = freshRuntime(); - // The budget is far shorter than the connector's post-boundary sleep - // (POST_BOUNDARY_SLEEP_MS = 3x the budget). Without the phase-boundary fix - // this run would be killed by run_timed_out one budget after the boundary; - // with the fix, the watchdog disarms itself once the PROGRESS - // phase_boundary message arrives and the run is free to run past the budget - // to genuine completion. - const launchRun = makeHarness(runtime, rsUrl, WALL_CLOCK_BUDGET_MS); - - const record = await launchRun( - schedule(writeConnectorWithPhaseBoundary(tmpDir, 5, POST_BOUNDARY_SLEEP_MS), ownerToken), + // Without the phase-boundary fix the watchdog would stay armed at one + // budget and kill the run once local-only work outlasted it; with the fix + // the boundary replaces that timer with the four-budget ceiling. The + // connector keeps working until released, so the only thing that could end + // the run is the armed timer, which this test asserts is the ceiling. + const watchdog = manualWatchdogTimers(); + const launchRun = makeHarness(freshRuntime(), rsUrl, WALL_CLOCK_BUDGET_MS, watchdog.timers); + + const pending = launchRun( + schedule(writeConnectorWithPhaseBoundary(tmpDir, 5), ownerToken), false, SCHEDULED_POLICY ); + try { + await waitFor(localOnlyPhaseLatched, "the local-only phase boundary to be latched"); + assert.equal( + watchdog.armedDelayMs(), + LOCAL_ONLY_CEILING_MS, + "after the boundary the primary budget must be replaced by the local-only ceiling" + ); + } finally { + // Release the connector even when an assertion above failed, so a failure + // reports instead of hanging on a connector that waits for this marker. + writeFileSync(join(tmpDir, CONNECTOR_RELEASE_MARKER), ""); + } + const record = await pending; assert.equal( record.status, @@ -524,11 +612,8 @@ test( test( "a run with no phase-boundary declaration never emits a local_only_phase_started spine event (control for the durable-trace test above)", - withServerAndTmpDir(async ({ ownerToken, rsUrl, tmpDir }) => { - const runtime = freshRuntime(); - const launchRun = makeHarness(runtime, rsUrl, WALL_CLOCK_BUDGET_MS); - - const record = await launchRun(schedule(writeConnector(tmpDir, 5, true), ownerToken), false, SCHEDULED_POLICY); + withServerAndTmpDir(async (ctx) => { + const { record } = await runHangingConnectorUntilWatchdogFires(ctx, 5); assert.equal(record.status, "failed"); assert.equal(record.terminalReason, "run_timed_out"); @@ -545,19 +630,18 @@ test( test( "without a phase-boundary declaration, a connector running past maxRunWallClockMs IS killed (control for the phase-boundary test above)", - withServerAndTmpDir(async ({ ownerToken, rsUrl, tmpDir }) => { - const runtime = freshRuntime(); - const launchRun = makeHarness(runtime, rsUrl, WALL_CLOCK_BUDGET_MS); - + withServerAndTmpDir(async (ctx) => { // Same shape as the phase-boundary connector but with no phase_boundary // PROGRESS message — proves the watchdog's default behavior (kill at // maxRunWallClockMs) is still intact and that it is specifically the // phase_boundary signal, not merely "the connector kept emitting - // PROGRESS", that disarms it. This connector DOES emit a plain "ready" - // PROGRESS, so that distinction is now exercised rather than assumed: - // a plain PROGRESS re-arms the timer, it does not disarm it. - const record = await launchRun(schedule(writeConnector(tmpDir, 5, true), ownerToken), false, SCHEDULED_POLICY); + // PROGRESS", that replaces the primary budget. This connector DOES emit a + // plain "ready" PROGRESS, so that distinction is exercised rather than + // assumed: a plain PROGRESS re-arms the primary budget, it does not + // replace it with the local-only ceiling. + const { armedDelayMs, record } = await runHangingConnectorUntilWatchdogFires(ctx, 5); + assert.equal(armedDelayMs, WALL_CLOCK_BUDGET_MS); assert.equal(record.status, "failed"); assert.equal(record.terminalReason, "run_timed_out"); }) @@ -566,18 +650,28 @@ test( test( "a connector that hangs after the local-only phase boundary is still killed by the secondary ceiling", withServerAndTmpDir(async ({ ownerToken, rsUrl, tmpDir }) => { - const runtime = freshRuntime(); - const launchRun = makeHarness(runtime, rsUrl, WALL_CLOCK_BUDGET_MS); - - // hangAfterBoundary=true, so the connector never reaches its sleep and the - // sleepMs argument is unreachable — the secondary ceiling - // (WALL_CLOCK_BUDGET_MS * 4, see createAttemptWatchdog) is what must end - // this attempt. - const record = await launchRun( - schedule(writeConnectorWithPhaseBoundary(tmpDir, 1, POST_BOUNDARY_SLEEP_MS, true), ownerToken), + const watchdog = manualWatchdogTimers(); + const launchRun = makeHarness(freshRuntime(), rsUrl, WALL_CLOCK_BUDGET_MS, watchdog.timers); + + // hangAfterBoundary=true, so the connector never completes — the secondary + // ceiling (WALL_CLOCK_BUDGET_MS * 4, see createAttemptWatchdog) is what must + // end this attempt. + const pending = launchRun( + schedule(writeConnectorWithPhaseBoundary(tmpDir, 1, true), ownerToken), false, SCHEDULED_POLICY ); + try { + await waitFor(localOnlyPhaseLatched, "the local-only phase boundary to be latched"); + assert.equal(watchdog.armedDelayMs(), LOCAL_ONLY_CEILING_MS); + } finally { + // Fire even when an assertion above failed, so a failure reports instead + // of hanging on a connector that never exits. + if (watchdog.armedDelayMs() !== null) { + watchdog.fire(); + } + } + const record = await pending; assert.equal(record.status, "failed"); assert.equal(record.terminalReason, "run_timed_out"); From dd554cef3b661e22d5de563e7b2a64b95ded8e3e Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Thu, 5 Nov 2026 16:25:53 -0600 Subject: [PATCH 2/6] test: give the watchdog timer seam its own handle type and cover the default timers WatchdogTimers.setTimeout now returns a canceller instead of a Node timer, so a test implementation no longer has to fabricate a NodeJS.Timeout and cast it back in clearTimeout. unref moves into the production default. The test header now says what the suite proves with injected timers. One real-timer test covers the default wiring: a tiny budget and a hanging connector must end as run_timed_out. Assisted-by: AI Signed-off-by: Tim Nunamaker --- .../runtime/scheduler/run-executor.ts | 43 ++++++++--------- .../test/run-executor-timeout-honesty.test.ts | 48 ++++++++++++------- 2 files changed, 53 insertions(+), 38 deletions(-) diff --git a/reference-implementation/runtime/scheduler/run-executor.ts b/reference-implementation/runtime/scheduler/run-executor.ts index 7f39c1380..93026165f 100644 --- a/reference-implementation/runtime/scheduler/run-executor.ts +++ b/reference-implementation/runtime/scheduler/run-executor.ts @@ -363,14 +363,23 @@ async function invokeRunConnector(call: RunConnectorCall): Promise void; - setTimeout: (callback: () => void, ms: number) => NodeJS.Timeout; + setTimeout: (callback: () => void, ms: number) => () => void; } +// The production default. `unref` keeps a pending watchdog from holding the +// process open on its own. const GLOBAL_WATCHDOG_TIMERS: WatchdogTimers = { - clearTimeout: (timer) => clearTimeout(timer), - setTimeout: (callback, ms) => setTimeout(callback, ms), + setTimeout: (callback, ms) => { + const handle = setTimeout(callback, ms); + handle.unref(); + return () => clearTimeout(handle); + }, }; interface AttemptWatchdog { @@ -406,7 +415,7 @@ export function createAttemptWatchdog( ): AttemptWatchdog { const cancellation = new AbortController(); let timedOut = false; - let timer: NodeJS.Timeout | null = null; + let cancelTimer: (() => void) | null = null; // Local-only work no longer has provider progress to reset the primary // watchdog. Keep the exemption bounded: four primary budgets is long enough // for the existing local materialization path while still guaranteeing that @@ -428,14 +437,11 @@ export function createAttemptWatchdog( if (!isBoundedWallClock(maxRunWallClockMs) || timedOut || localOnlyPhase || cancellation.signal.aborted) { return; } - if (timer) { - timers.clearTimeout(timer); - } - timer = timers.setTimeout(() => { + cancelTimer?.(); + cancelTimer = timers.setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); }, maxRunWallClockMs); - timer.unref?.(); }; const armLocalOnlyPhaseCeiling = () => { @@ -446,14 +452,11 @@ export function createAttemptWatchdog( ) { return; } - if (timer) { - timers.clearTimeout(timer); - } - timer = timers.setTimeout(() => { + cancelTimer?.(); + cancelTimer = timers.setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); }, localOnlyPhaseMaxRunWallClockMs); - timer.unref?.(); }; if (isBoundedWallClock(maxRunWallClockMs)) { @@ -467,17 +470,13 @@ export function createAttemptWatchdog( } }, clear() { - if (timer) { - timers.clearTimeout(timer); - timer = null; - } + cancelTimer?.(); + cancelTimer = null; }, markProgress(extra) { if (extra?.phase_boundary === "local_only_phase_started" && !localOnlyPhase) { localOnlyPhase = true; - if (timer) { - timers.clearTimeout(timer); - } + cancelTimer?.(); armLocalOnlyPhaseCeiling(); // TRUE only on the latching call — the caller emits the durable trace. return true; diff --git a/reference-implementation/test/run-executor-timeout-honesty.test.ts b/reference-implementation/test/run-executor-timeout-honesty.test.ts index 06c33cc41..246541c26 100644 --- a/reference-implementation/test/run-executor-timeout-honesty.test.ts +++ b/reference-implementation/test/run-executor-timeout-honesty.test.ts @@ -17,12 +17,14 @@ // 91,202 reactions + 81,857 message_attachments + 19,936 files, and still // recorded records_emitted: 0, known_gaps_json: []. // -// This suite drives a REAL timeout: a connector that durably emits records -// through a real in-memory ingest server, then hangs forever (never sends -// DONE), with maxRunWallClockMs set low enough that the watchdog's real -// setTimeout fires and kills it. This is not a mock of the timeout path — -// it is the actual watchdog, the actual child-process kill, and the actual -// ingest round-trip that production hit. +// This suite runs the real executor, a real child process, a real in-memory +// ingest server and the real kill path against a connector that durably emits +// records and then hangs forever (never sends DONE). The watchdog's timers are +// injected (`WatchdogTimers`), so each test fires the timeout at a chosen point +// rather than racing a wall-clock budget against process startup. What it +// proves: the timed-out result carries the durable counts and gaps. What it +// does not prove: that a real timer elapses. The production default timer path +// is covered only by the single real-timer test at the end of this file. import assert from "node:assert/strict"; import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; @@ -189,18 +191,13 @@ function manualWatchdogTimers(): ManualWatchdogTimers { timer.callback(); }, timers: { - clearTimeout: (handle) => { - armed.delete((handle as unknown as { id: number }).id); - }, setTimeout: (callback, ms) => { nextId += 1; - armed.set(nextId, { callback, ms }); - return { - id: nextId, - unref() { - return this; - }, - } as unknown as NodeJS.Timeout; + const id = nextId; + armed.set(id, { callback, ms }); + return () => { + armed.delete(id); + }; }, }, }; @@ -677,3 +674,22 @@ test( assert.equal(record.terminalReason, "run_timed_out"); }) ); + +test( + "the production default timers arm a real timer that kills a hanging connector with run_timed_out", + withServerAndTmpDir(async (ctx) => { + // No injected timers: this exercises GLOBAL_WATCHDOG_TIMERS. The budget is + // tiny, and the assertion waits on the run's own terminal event, so it does + // not depend on how long startup takes. Only the classification is asserted + // (the record count depends on how far the connector got before the kill). + const launchRun = makeHarness(freshRuntime(), ctx.rsUrl, 20); + const record = await launchRun( + schedule(writeConnector(ctx.tmpDir, 0, true), ctx.ownerToken), + false, + SCHEDULED_POLICY + ); + + assert.equal(record.status, "failed"); + assert.equal(record.terminalReason, "run_timed_out"); + }) +); From 78e4d562fd130272764547385c9840f6e5ff9b8a Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Fri, 6 Nov 2026 00:03:34 -0600 Subject: [PATCH 3/6] fix(scheduler): reject an invalid watchdog budget and clamp timer delays A NaN or negative maxRunWallClockMs read as "no ceiling", so a hanging connector was never killed and nothing recorded why. createAttemptWatchdog and createRunExecutor now throw on it. Infinity and 0 stay the documented explicit opt-outs. Timer delays are clamped to 2^31 - 1 ms: budget * 4 could overflow to Infinity (read as unbounded) and Node fires a longer setTimeout almost at once. Assisted-by: AI Signed-off-by: Tim Nunamaker --- .../runtime/scheduler/run-executor.ts | 30 +++++++++++-- .../test/run-executor-timeout-honesty.test.ts | 44 +++++++++++++++++++ 2 files changed, 71 insertions(+), 3 deletions(-) diff --git a/reference-implementation/runtime/scheduler/run-executor.ts b/reference-implementation/runtime/scheduler/run-executor.ts index 93026165f..cfab762c9 100644 --- a/reference-implementation/runtime/scheduler/run-executor.ts +++ b/reference-implementation/runtime/scheduler/run-executor.ts @@ -409,10 +409,30 @@ function isBoundedWallClock(maxRunWallClockMs: number): boolean { return Number.isFinite(maxRunWallClockMs) && maxRunWallClockMs > 0; } +// `0` and `Infinity` are the documented explicit opt-outs (see +// runtime/scheduler-config.ts). NaN and negative values are not: they used to +// read as "no ceiling", so a hanging connector was never killed and nothing +// said why. Reject them where the budget enters the watchdog. +function assertValidWallClockBudget(maxRunWallClockMs: number): void { + if (Number.isNaN(maxRunWallClockMs) || maxRunWallClockMs < 0) { + throw new Error(`maxRunWallClockMs must be a non-negative number or Infinity; got ${maxRunWallClockMs}`); + } +} + +// `setTimeout` cannot delay past 2^31 - 1 ms: Node fires it almost at once. +// `budget * 4` can also overflow to Infinity, which reads as "unbounded", so a +// huge finite budget would silently lose its ceiling. +const MAX_TIMER_DELAY_MS = 2 ** 31 - 1; + +function clampTimerDelay(ms: number): number { + return Math.min(ms, MAX_TIMER_DELAY_MS); +} + export function createAttemptWatchdog( maxRunWallClockMs: number, timers: WatchdogTimers = GLOBAL_WATCHDOG_TIMERS ): AttemptWatchdog { + assertValidWallClockBudget(maxRunWallClockMs); const cancellation = new AbortController(); let timedOut = false; let cancelTimer: (() => void) | null = null; @@ -420,7 +440,9 @@ export function createAttemptWatchdog( // watchdog. Keep the exemption bounded: four primary budgets is long enough // for the existing local materialization path while still guaranteeing that // a connector that stops making progress cannot survive until restart. - const localOnlyPhaseMaxRunWallClockMs = maxRunWallClockMs * 4; + const localOnlyPhaseMaxRunWallClockMs = isBoundedWallClock(maxRunWallClockMs) + ? clampTimerDelay(maxRunWallClockMs * 4) + : maxRunWallClockMs; // Set once a connector reports `phase_boundary: "local_only_phase_started"` // and never cleared for the rest of this attempt. `maxRunWallClockMs` is // sized for provider-rate-limited walks (external API pagination); once a @@ -441,7 +463,7 @@ export function createAttemptWatchdog( cancelTimer = timers.setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); - }, maxRunWallClockMs); + }, clampTimerDelay(maxRunWallClockMs)); }; const armLocalOnlyPhaseCeiling = () => { @@ -456,7 +478,7 @@ export function createAttemptWatchdog( cancelTimer = timers.setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); - }, localOnlyPhaseMaxRunWallClockMs); + }, clampTimerDelay(localOnlyPhaseMaxRunWallClockMs)); }; if (isBoundedWallClock(maxRunWallClockMs)) { @@ -793,6 +815,8 @@ export function createRunExecutor(deps: RunExecutorDeps): RunExecutor { setState, watchdogTimers, } = deps; + // Fail at construction, not on the first run that happens to hang. + assertValidWallClockBudget(maxRunWallClockMs); async function appendRunHistoryBestEffort(record: RunRecord, connectorId: string): Promise { if (!schedulerStore) { diff --git a/reference-implementation/test/run-executor-timeout-honesty.test.ts b/reference-implementation/test/run-executor-timeout-honesty.test.ts index 246541c26..567f17361 100644 --- a/reference-implementation/test/run-executor-timeout-honesty.test.ts +++ b/reference-implementation/test/run-executor-timeout-honesty.test.ts @@ -34,6 +34,7 @@ import test from "node:test"; import { projectRunAutomationPolicy } from "../runtime/run-automation-policy.ts"; import { + createAttemptWatchdog, createRunExecutor, type RunExecutorDeps, type RunExecutorRuntimeState, @@ -693,3 +694,46 @@ test( assert.equal(record.terminalReason, "run_timed_out"); }) ); + +const BUDGET_ERROR = /maxRunWallClockMs/; + +// The watchdog budget contract: a positive finite number arms the ceiling, +// `Infinity` and `0` are the documented ways to disable it +// (runtime/scheduler-config.ts), and anything else is a configuration error. +// A NaN or negative budget used to read as "no ceiling", so a hanging connector +// was never killed and nothing recorded why. +for (const invalid of [Number.NaN, -1, Number.NEGATIVE_INFINITY]) { + test(`createAttemptWatchdog rejects the invalid budget ${String(invalid)} instead of silently disabling the ceiling`, () => { + const watchdog = manualWatchdogTimers(); + assert.throws(() => createAttemptWatchdog(invalid, watchdog.timers), BUDGET_ERROR); + assert.equal(watchdog.armedDelayMs(), null); + }); +} + +for (const optOut of [0, Number.POSITIVE_INFINITY]) { + test(`createAttemptWatchdog treats ${String(optOut)} as the explicit opt-out: nothing is armed`, () => { + const watchdog = manualWatchdogTimers(); + const attempt = createAttemptWatchdog(optOut, watchdog.timers); + attempt.markProgress({ phase_boundary: "local_only_phase_started" }); + assert.equal(watchdog.armedDelayMs(), null); + assert.equal(attempt.timedOut(), false); + }); +} + +// setTimeout cannot delay past 2^31 - 1 ms (Node fires it almost at once), and +// `budget * 4` can overflow to Infinity, which reads as "unbounded". A huge but +// finite budget must keep a real, finite ceiling. +const MAX_TIMER_DELAY_MS = 2 ** 31 - 1; +for (const hugeBudget of [2 ** 30, 1e308]) { + test(`the local-only ceiling for a finite budget of ${String(hugeBudget)} ms is clamped to a finite timer delay`, () => { + const watchdog = manualWatchdogTimers(); + const attempt = createAttemptWatchdog(hugeBudget, watchdog.timers); + attempt.markProgress({ phase_boundary: "local_only_phase_started" }); + assert.equal(watchdog.armedDelayMs(), MAX_TIMER_DELAY_MS); + }); +} + +test("createRunExecutor rejects an invalid wall-clock budget at construction", () => { + assert.throws(() => makeHarness(freshRuntime(), "http://127.0.0.1:1", Number.NaN), BUDGET_ERROR); + assert.throws(() => makeHarness(freshRuntime(), "http://127.0.0.1:1", -1), BUDGET_ERROR); +}); From 255107a764d0dcdf067c55d4f064cae75bbbeb8b Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Fri, 6 Nov 2026 06:16:08 -0600 Subject: [PATCH 4/6] fix(scheduler): latch the local-only phase only when a timer was armed markProgress latched the local-only phase and reported a suppression even when nothing was armed: with an opt-out budget (0 or Infinity), after the watchdog fired, or after an owner cancel. The caller then recorded a "watchdog suppressed" event for a suppression that never happened. It now latches only while a primary timer could still fire. The WatchdogTimers canceller is documented as idempotent, and the cancelled timer is dropped after the latch. Assisted-by: AI Signed-off-by: Tim Nunamaker --- .../runtime/scheduler/run-executor.ts | 12 +++++-- .../test/run-executor-timeout-honesty.test.ts | 33 ++++++++++++++++++- 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/reference-implementation/runtime/scheduler/run-executor.ts b/reference-implementation/runtime/scheduler/run-executor.ts index cfab762c9..178aaea6f 100644 --- a/reference-implementation/runtime/scheduler/run-executor.ts +++ b/reference-implementation/runtime/scheduler/run-executor.ts @@ -366,7 +366,8 @@ async function invokeRunConnector(call: RunConnectorCall): Promise void, ms: number) => () => void; @@ -466,6 +467,12 @@ export function createAttemptWatchdog( }, clampTimerDelay(maxRunWallClockMs)); }; + // Suppression is only real while a primary timer could still fire: with an + // opt-out budget, after the watchdog fired, or after an owner cancel there is + // nothing to suppress, and the caller must not record that it did. + const canSuppressPrimaryBudget = () => + isBoundedWallClock(maxRunWallClockMs) && !timedOut && !cancellation.signal.aborted; + const armLocalOnlyPhaseCeiling = () => { if ( !isBoundedWallClock(localOnlyPhaseMaxRunWallClockMs) || @@ -496,9 +503,10 @@ export function createAttemptWatchdog( cancelTimer = null; }, markProgress(extra) { - if (extra?.phase_boundary === "local_only_phase_started" && !localOnlyPhase) { + if (extra?.phase_boundary === "local_only_phase_started" && !localOnlyPhase && canSuppressPrimaryBudget()) { localOnlyPhase = true; cancelTimer?.(); + cancelTimer = null; armLocalOnlyPhaseCeiling(); // TRUE only on the latching call — the caller emits the durable trace. return true; diff --git a/reference-implementation/test/run-executor-timeout-honesty.test.ts b/reference-implementation/test/run-executor-timeout-honesty.test.ts index 567f17361..e45078a5f 100644 --- a/reference-implementation/test/run-executor-timeout-honesty.test.ts +++ b/reference-implementation/test/run-executor-timeout-honesty.test.ts @@ -714,12 +714,43 @@ for (const optOut of [0, Number.POSITIVE_INFINITY]) { test(`createAttemptWatchdog treats ${String(optOut)} as the explicit opt-out: nothing is armed`, () => { const watchdog = manualWatchdogTimers(); const attempt = createAttemptWatchdog(optOut, watchdog.timers); - attempt.markProgress({ phase_boundary: "local_only_phase_started" }); + // Nothing was armed, so nothing was suppressed: the caller must not write a + // "watchdog suppressed" spine event or progress message for it. + assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), false); assert.equal(watchdog.armedDelayMs(), null); assert.equal(attempt.timedOut(), false); }); } +test("a phase boundary arriving after the watchdog already fired suppresses nothing", () => { + const watchdog = manualWatchdogTimers(); + const attempt = createAttemptWatchdog(WALL_CLOCK_BUDGET_MS, watchdog.timers); + watchdog.fire(); + assert.equal(attempt.timedOut(), true); + assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), false); + assert.equal(watchdog.armedDelayMs(), null); +}); + +test("a phase boundary arriving after an owner cancel suppresses nothing", () => { + const watchdog = manualWatchdogTimers(); + const attempt = createAttemptWatchdog(WALL_CLOCK_BUDGET_MS, watchdog.timers); + attempt.cancel(); + assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), false); + assert.equal( + watchdog.armedDelayMs(), + WALL_CLOCK_BUDGET_MS, + "cancel does not disarm the primary timer; the latch must not re-arm" + ); +}); + +test("a phase boundary on an armed watchdog still latches exactly once", () => { + const watchdog = manualWatchdogTimers(); + const attempt = createAttemptWatchdog(WALL_CLOCK_BUDGET_MS, watchdog.timers); + assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), true); + assert.equal(watchdog.armedDelayMs(), LOCAL_ONLY_CEILING_MS); + assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), false); +}); + // setTimeout cannot delay past 2^31 - 1 ms (Node fires it almost at once), and // `budget * 4` can overflow to Infinity, which reads as "unbounded". A huge but // finite budget must keep a real, finite ceiling. From ae531fef46f026fb9642d81340555bf37a0d0d60 Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Fri, 6 Nov 2026 10:59:55 -0600 Subject: [PATCH 5/6] refactor(scheduler): drop the duplicate wall-clock budget check The scheduler already rejects a NaN or negative maxRunWallClockMs with resolveNonNegativeMsOrInfinity before createRunExecutor sees it, and that is the only production caller. The watchdog's own copy had the same message, so it and its tests are removed. Behaviour is unchanged. Assisted-by: AI Signed-off-by: Tim Nunamaker --- .../runtime/scheduler/run-executor.ts | 13 ---------- .../test/run-executor-timeout-honesty.test.ts | 24 ++++--------------- 2 files changed, 4 insertions(+), 33 deletions(-) diff --git a/reference-implementation/runtime/scheduler/run-executor.ts b/reference-implementation/runtime/scheduler/run-executor.ts index 178aaea6f..c3406efa5 100644 --- a/reference-implementation/runtime/scheduler/run-executor.ts +++ b/reference-implementation/runtime/scheduler/run-executor.ts @@ -410,16 +410,6 @@ function isBoundedWallClock(maxRunWallClockMs: number): boolean { return Number.isFinite(maxRunWallClockMs) && maxRunWallClockMs > 0; } -// `0` and `Infinity` are the documented explicit opt-outs (see -// runtime/scheduler-config.ts). NaN and negative values are not: they used to -// read as "no ceiling", so a hanging connector was never killed and nothing -// said why. Reject them where the budget enters the watchdog. -function assertValidWallClockBudget(maxRunWallClockMs: number): void { - if (Number.isNaN(maxRunWallClockMs) || maxRunWallClockMs < 0) { - throw new Error(`maxRunWallClockMs must be a non-negative number or Infinity; got ${maxRunWallClockMs}`); - } -} - // `setTimeout` cannot delay past 2^31 - 1 ms: Node fires it almost at once. // `budget * 4` can also overflow to Infinity, which reads as "unbounded", so a // huge finite budget would silently lose its ceiling. @@ -433,7 +423,6 @@ export function createAttemptWatchdog( maxRunWallClockMs: number, timers: WatchdogTimers = GLOBAL_WATCHDOG_TIMERS ): AttemptWatchdog { - assertValidWallClockBudget(maxRunWallClockMs); const cancellation = new AbortController(); let timedOut = false; let cancelTimer: (() => void) | null = null; @@ -823,8 +812,6 @@ export function createRunExecutor(deps: RunExecutorDeps): RunExecutor { setState, watchdogTimers, } = deps; - // Fail at construction, not on the first run that happens to hang. - assertValidWallClockBudget(maxRunWallClockMs); async function appendRunHistoryBestEffort(record: RunRecord, connectorId: string): Promise { if (!schedulerStore) { diff --git a/reference-implementation/test/run-executor-timeout-honesty.test.ts b/reference-implementation/test/run-executor-timeout-honesty.test.ts index e45078a5f..9c872fe60 100644 --- a/reference-implementation/test/run-executor-timeout-honesty.test.ts +++ b/reference-implementation/test/run-executor-timeout-honesty.test.ts @@ -695,21 +695,10 @@ test( }) ); -const BUDGET_ERROR = /maxRunWallClockMs/; - -// The watchdog budget contract: a positive finite number arms the ceiling, -// `Infinity` and `0` are the documented ways to disable it -// (runtime/scheduler-config.ts), and anything else is a configuration error. -// A NaN or negative budget used to read as "no ceiling", so a hanging connector -// was never killed and nothing recorded why. -for (const invalid of [Number.NaN, -1, Number.NEGATIVE_INFINITY]) { - test(`createAttemptWatchdog rejects the invalid budget ${String(invalid)} instead of silently disabling the ceiling`, () => { - const watchdog = manualWatchdogTimers(); - assert.throws(() => createAttemptWatchdog(invalid, watchdog.timers), BUDGET_ERROR); - assert.equal(watchdog.armedDelayMs(), null); - }); -} - +// The watchdog budget contract: a positive finite number arms the ceiling and +// `Infinity` and `0` are the documented explicit opt-outs +// (runtime/scheduler-config.ts). The scheduler rejects NaN and negative budgets +// before they reach the watchdog (resolveNonNegativeMsOrInfinity). for (const optOut of [0, Number.POSITIVE_INFINITY]) { test(`createAttemptWatchdog treats ${String(optOut)} as the explicit opt-out: nothing is armed`, () => { const watchdog = manualWatchdogTimers(); @@ -763,8 +752,3 @@ for (const hugeBudget of [2 ** 30, 1e308]) { assert.equal(watchdog.armedDelayMs(), MAX_TIMER_DELAY_MS); }); } - -test("createRunExecutor rejects an invalid wall-clock budget at construction", () => { - assert.throws(() => makeHarness(freshRuntime(), "http://127.0.0.1:1", Number.NaN), BUDGET_ERROR); - assert.throws(() => makeHarness(freshRuntime(), "http://127.0.0.1:1", -1), BUDGET_ERROR); -}); From 141f6fafc0f39983a610f6627339d75ecc665cf1 Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Fri, 6 Nov 2026 11:03:54 -0600 Subject: [PATCH 6/6] test(scheduler): keep the watchdog timer seam only Move the timer-delay clamp and the local-only latch guard out of this change so the production diff stays within the mutation cohort bound. They return as their own change with their own tests. Assisted-by: AI Signed-off-by: Tim Nunamaker --- .../runtime/scheduler/run-executor.ts | 29 ++------- .../test/run-executor-timeout-honesty.test.ts | 59 ------------------- 2 files changed, 5 insertions(+), 83 deletions(-) diff --git a/reference-implementation/runtime/scheduler/run-executor.ts b/reference-implementation/runtime/scheduler/run-executor.ts index c3406efa5..93026165f 100644 --- a/reference-implementation/runtime/scheduler/run-executor.ts +++ b/reference-implementation/runtime/scheduler/run-executor.ts @@ -366,8 +366,7 @@ async function invokeRunConnector(call: RunConnectorCall): Promise void, ms: number) => () => void; @@ -410,15 +409,6 @@ function isBoundedWallClock(maxRunWallClockMs: number): boolean { return Number.isFinite(maxRunWallClockMs) && maxRunWallClockMs > 0; } -// `setTimeout` cannot delay past 2^31 - 1 ms: Node fires it almost at once. -// `budget * 4` can also overflow to Infinity, which reads as "unbounded", so a -// huge finite budget would silently lose its ceiling. -const MAX_TIMER_DELAY_MS = 2 ** 31 - 1; - -function clampTimerDelay(ms: number): number { - return Math.min(ms, MAX_TIMER_DELAY_MS); -} - export function createAttemptWatchdog( maxRunWallClockMs: number, timers: WatchdogTimers = GLOBAL_WATCHDOG_TIMERS @@ -430,9 +420,7 @@ export function createAttemptWatchdog( // watchdog. Keep the exemption bounded: four primary budgets is long enough // for the existing local materialization path while still guaranteeing that // a connector that stops making progress cannot survive until restart. - const localOnlyPhaseMaxRunWallClockMs = isBoundedWallClock(maxRunWallClockMs) - ? clampTimerDelay(maxRunWallClockMs * 4) - : maxRunWallClockMs; + const localOnlyPhaseMaxRunWallClockMs = maxRunWallClockMs * 4; // Set once a connector reports `phase_boundary: "local_only_phase_started"` // and never cleared for the rest of this attempt. `maxRunWallClockMs` is // sized for provider-rate-limited walks (external API pagination); once a @@ -453,15 +441,9 @@ export function createAttemptWatchdog( cancelTimer = timers.setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); - }, clampTimerDelay(maxRunWallClockMs)); + }, maxRunWallClockMs); }; - // Suppression is only real while a primary timer could still fire: with an - // opt-out budget, after the watchdog fired, or after an owner cancel there is - // nothing to suppress, and the caller must not record that it did. - const canSuppressPrimaryBudget = () => - isBoundedWallClock(maxRunWallClockMs) && !timedOut && !cancellation.signal.aborted; - const armLocalOnlyPhaseCeiling = () => { if ( !isBoundedWallClock(localOnlyPhaseMaxRunWallClockMs) || @@ -474,7 +456,7 @@ export function createAttemptWatchdog( cancelTimer = timers.setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); - }, clampTimerDelay(localOnlyPhaseMaxRunWallClockMs)); + }, localOnlyPhaseMaxRunWallClockMs); }; if (isBoundedWallClock(maxRunWallClockMs)) { @@ -492,10 +474,9 @@ export function createAttemptWatchdog( cancelTimer = null; }, markProgress(extra) { - if (extra?.phase_boundary === "local_only_phase_started" && !localOnlyPhase && canSuppressPrimaryBudget()) { + if (extra?.phase_boundary === "local_only_phase_started" && !localOnlyPhase) { localOnlyPhase = true; cancelTimer?.(); - cancelTimer = null; armLocalOnlyPhaseCeiling(); // TRUE only on the latching call — the caller emits the durable trace. return true; diff --git a/reference-implementation/test/run-executor-timeout-honesty.test.ts b/reference-implementation/test/run-executor-timeout-honesty.test.ts index 9c872fe60..246541c26 100644 --- a/reference-implementation/test/run-executor-timeout-honesty.test.ts +++ b/reference-implementation/test/run-executor-timeout-honesty.test.ts @@ -34,7 +34,6 @@ import test from "node:test"; import { projectRunAutomationPolicy } from "../runtime/run-automation-policy.ts"; import { - createAttemptWatchdog, createRunExecutor, type RunExecutorDeps, type RunExecutorRuntimeState, @@ -694,61 +693,3 @@ test( assert.equal(record.terminalReason, "run_timed_out"); }) ); - -// The watchdog budget contract: a positive finite number arms the ceiling and -// `Infinity` and `0` are the documented explicit opt-outs -// (runtime/scheduler-config.ts). The scheduler rejects NaN and negative budgets -// before they reach the watchdog (resolveNonNegativeMsOrInfinity). -for (const optOut of [0, Number.POSITIVE_INFINITY]) { - test(`createAttemptWatchdog treats ${String(optOut)} as the explicit opt-out: nothing is armed`, () => { - const watchdog = manualWatchdogTimers(); - const attempt = createAttemptWatchdog(optOut, watchdog.timers); - // Nothing was armed, so nothing was suppressed: the caller must not write a - // "watchdog suppressed" spine event or progress message for it. - assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), false); - assert.equal(watchdog.armedDelayMs(), null); - assert.equal(attempt.timedOut(), false); - }); -} - -test("a phase boundary arriving after the watchdog already fired suppresses nothing", () => { - const watchdog = manualWatchdogTimers(); - const attempt = createAttemptWatchdog(WALL_CLOCK_BUDGET_MS, watchdog.timers); - watchdog.fire(); - assert.equal(attempt.timedOut(), true); - assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), false); - assert.equal(watchdog.armedDelayMs(), null); -}); - -test("a phase boundary arriving after an owner cancel suppresses nothing", () => { - const watchdog = manualWatchdogTimers(); - const attempt = createAttemptWatchdog(WALL_CLOCK_BUDGET_MS, watchdog.timers); - attempt.cancel(); - assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), false); - assert.equal( - watchdog.armedDelayMs(), - WALL_CLOCK_BUDGET_MS, - "cancel does not disarm the primary timer; the latch must not re-arm" - ); -}); - -test("a phase boundary on an armed watchdog still latches exactly once", () => { - const watchdog = manualWatchdogTimers(); - const attempt = createAttemptWatchdog(WALL_CLOCK_BUDGET_MS, watchdog.timers); - assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), true); - assert.equal(watchdog.armedDelayMs(), LOCAL_ONLY_CEILING_MS); - assert.equal(attempt.markProgress({ phase_boundary: "local_only_phase_started" }), false); -}); - -// setTimeout cannot delay past 2^31 - 1 ms (Node fires it almost at once), and -// `budget * 4` can overflow to Infinity, which reads as "unbounded". A huge but -// finite budget must keep a real, finite ceiling. -const MAX_TIMER_DELAY_MS = 2 ** 31 - 1; -for (const hugeBudget of [2 ** 30, 1e308]) { - test(`the local-only ceiling for a finite budget of ${String(hugeBudget)} ms is clamped to a finite timer delay`, () => { - const watchdog = manualWatchdogTimers(); - const attempt = createAttemptWatchdog(hugeBudget, watchdog.timers); - attempt.markProgress({ phase_boundary: "local_only_phase_started" }); - assert.equal(watchdog.armedDelayMs(), MAX_TIMER_DELAY_MS); - }); -}