diff --git a/reference-implementation/runtime/scheduler/run-executor.ts b/reference-implementation/runtime/scheduler/run-executor.ts index b030e6f20..88c0ef808 100644 --- a/reference-implementation/runtime/scheduler/run-executor.ts +++ b/reference-implementation/runtime/scheduler/run-executor.ts @@ -383,6 +383,15 @@ 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): AttemptWatchdog { const cancellation = new AbortController(); let timedOut = false; @@ -391,7 +400,9 @@ export function createAttemptWatchdog(maxRunWallClockMs: number): AttemptWatchdo // 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 @@ -414,10 +425,16 @@ export function createAttemptWatchdog(maxRunWallClockMs: number): AttemptWatchdo timer = setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); - }, maxRunWallClockMs); + }, clampTimerDelay(maxRunWallClockMs)); timer.unref?.(); }; + // 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) || @@ -432,7 +449,7 @@ export function createAttemptWatchdog(maxRunWallClockMs: number): AttemptWatchdo timer = setTimeout(() => { timedOut = true; cancellation.abort("run_timed_out"); - }, localOnlyPhaseMaxRunWallClockMs); + }, clampTimerDelay(localOnlyPhaseMaxRunWallClockMs)); timer.unref?.(); }; @@ -453,7 +470,7 @@ export function createAttemptWatchdog(maxRunWallClockMs: number): AttemptWatchdo } }, markProgress(extra) { - if (extra?.phase_boundary === "local_only_phase_started" && !localOnlyPhase) { + if (extra?.phase_boundary === "local_only_phase_started" && !localOnlyPhase && canSuppressPrimaryBudget()) { localOnlyPhase = true; if (timer) { clearTimeout(timer); diff --git a/reference-implementation/test/run-executor-watchdog-limits.test.ts b/reference-implementation/test/run-executor-watchdog-limits.test.ts new file mode 100644 index 000000000..e3eabf7fc --- /dev/null +++ b/reference-implementation/test/run-executor-watchdog-limits.test.ts @@ -0,0 +1,90 @@ +// Copyright The PDP-Connect Contributors +// SPDX-License-Identifier: Apache-2.0 + +import assert from "node:assert/strict"; +import { mock, test } from "node:test"; +import { createAttemptWatchdog } from "../runtime/scheduler/run-executor.ts"; + +const PHASE_BOUNDARY = { phase_boundary: "local_only_phase_started" }; +const MAX_TIMER_DELAY_MS = 2 ** 31 - 1; +const SHORT_BUDGET_MS = 5; + +function waitForAbort(signal: AbortSignal): Promise { + return new Promise((resolve) => { + signal.addEventListener("abort", () => resolve(), { once: true }); + }); +} + +// `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. +for (const hugeBudget of [2 ** 30, 1e308]) { + test(`a finite budget of ${String(hugeBudget)} ms arms timers no longer than the timer limit`, () => { + const realSetTimeout = globalThis.setTimeout; + const delays: number[] = []; + const spy = mock.method(globalThis, "setTimeout", ((callback: () => void, ms?: number) => { + delays.push(Number(ms)); + // Never let the test arm a real long timer. + return realSetTimeout(callback, 60_000); + }) as typeof setTimeout); + const watchdog = createAttemptWatchdog(hugeBudget); + try { + watchdog.markProgress(PHASE_BOUNDARY); + assert.ok(delays.length >= 2, "the primary timer and the local-only ceiling are both armed"); + for (const delay of delays) { + assert.ok( + Number.isFinite(delay) && delay <= MAX_TIMER_DELAY_MS, + `armed delay ${delay} exceeds the timer limit` + ); + } + assert.equal(delays.at(-1), MAX_TIMER_DELAY_MS, "the local-only ceiling stays bounded at the limit"); + } finally { + watchdog.clear(); + spy.mock.restore(); + } + }); +} + +// A "watchdog suppressed" record is written when markProgress reports true. +// That must be true only when a timer was armed and could still have fired. +for (const optOut of [0, Number.POSITIVE_INFINITY]) { + test(`a phase boundary with the opt-out budget ${String(optOut)} suppresses nothing`, () => { + const watchdog = createAttemptWatchdog(optOut); + try { + assert.equal(watchdog.markProgress(PHASE_BOUNDARY), false); + } finally { + watchdog.clear(); + } + }); +} + +test("a phase boundary arriving after the watchdog fired suppresses nothing", async () => { + const watchdog = createAttemptWatchdog(SHORT_BUDGET_MS); + try { + await waitForAbort(watchdog.signal); + assert.equal(watchdog.timedOut(), true); + assert.equal(watchdog.markProgress(PHASE_BOUNDARY), false); + } finally { + watchdog.clear(); + } +}); + +test("a phase boundary arriving after an owner cancel suppresses nothing", () => { + const watchdog = createAttemptWatchdog(60_000); + try { + watchdog.cancel(); + assert.equal(watchdog.markProgress(PHASE_BOUNDARY), false); + } finally { + watchdog.clear(); + } +}); + +test("a phase boundary on an armed watchdog latches exactly once", () => { + const watchdog = createAttemptWatchdog(60_000); + try { + assert.equal(watchdog.markProgress(PHASE_BOUNDARY), true); + assert.equal(watchdog.markProgress(PHASE_BOUNDARY), false); + } finally { + watchdog.clear(); + } +});