feat(benchsdk-runner): deadlineMs — a run-level deadline drains and finishes, not fails

.changeset/runner-deadline-ms.md

@@ -0,0 +1,7 @@
+---
+"@benchsdk/runner": patch
+"@benchsdk/worker": patch
+"@benchsdk/api": patch
+---
+
+Add `deadlineMs` / `--deadline-ms`: a run-level wall-clock budget for task execution. On expiry the run stops launching new tasks, drains in-flight ones, and still flushes, reports, and exits 0 — a deadline is a planned end, not a failure. `BenchmarkRunOutcome.stoppedEarly` flags the truncation; `RunWorkerOptions.stopRequested` is the plumbing that stops new task starts inside `runWorker`.

.github/workflows/snapshot-fork-benchmarks.yml

@@ -141,13 +141,17 @@ jobs:
           # All matrix jobs for this workflow run share a single platform run. The
           # attempt number is included so re-runs do not rejoin completed runs.
           RUN_KEY="${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-          # Cap the benchmark process at 60 minutes so the job can still upload
-          # partial results and complete instead of being cancelled by GitHub.
-          timeout 60m npx tsx packages/benchsdk-runner/dist/bin.js run benchmarks/storage/snapshot-fork.bench.ts \
+          # The benchmark gets a 60-minute wall-clock budget via the runner's own
+          # deadline: on expiry it stops launching iterations, drains in-flight
+          # work, and submits partial results, so the job completes instead of
+          # failing on a timeout(1) kill. `timeout 75m` stays as a hard backstop
+          # (under the job's 90-minute cap) for a drain that itself hangs.
+          timeout 75m npx tsx packages/benchsdk-runner/dist/bin.js run benchmarks/storage/snapshot-fork.bench.ts \
             --provider ${{ matrix.provider }} \
             --dataset ${{ github.event.inputs.dataset || 'small' }} \
             --iterations ${{ (github.event_name == 'schedule' && '100') || github.event.inputs.iterations || (github.event_name == 'push' && '1') || '3' }} \
             --run-key "$RUN_KEY" \
+            --deadline-ms 3600000 \
             ${{ (github.event_name == 'push' || github.event.inputs.dry_run == 'true') && '--no-ingest' || '' }}
       - name: Upload results
         if: always()

packages/benchsdk-api/src/types.ts

@@ -758,6 +758,13 @@ export interface RunWorkerOptions {
    * values are ever captured.
    */
   metricsIntervalMs?: number;
+  /**
+   * Polled before each new task start: once it returns true the worker stops
+   * launching tasks, drains the in-flight ones, flushes and finishes normally
+   * with the records it produced — a graceful stop, not an error. Used by the
+   * runner's run-level `deadlineMs`.
+   */
+  stopRequested?: () => boolean;
   task: TaskFunction;
 }
 

packages/benchsdk-runner/src/__tests__/runner.test.ts

@@ -129,6 +129,14 @@ describe('parseCliArgs', () => {
     expect(() => parseCliArgs(['--stagger-delay-ms', '-1'])).toThrow('--stagger-delay-ms');
   });
 
+  it('parses --deadline-ms and rejects non-positive values', () => {
+    expect(parseCliArgs(['--deadline-ms', '3600000'])).toEqual({ deadlineMs: 3600000 });
+    expect(parseCliArgs(['--deadline-ms=250'])).toEqual({ deadlineMs: 250 });
+    expect(() => parseCliArgs(['--deadline-ms', '0'])).toThrow('--deadline-ms');
+    expect(() => parseCliArgs(['--deadline-ms', '-5'])).toThrow('--deadline-ms');
+    expect(() => parseCliArgs(['--deadline-ms', 'abc'])).toThrow('--deadline-ms');
+  });
+
   it('returns empty object for no args', () => {
     expect(parseCliArgs([])).toEqual({});
   });
@@ -173,6 +181,12 @@ describe('mergeConfig', () => {
     });
   });
 
+  it('resolves deadlineMs from config or --deadline-ms, CLI winning', () => {
+    expect(mergeConfig(config, {}).deadlineMs).toBeUndefined();
+    expect(mergeConfig({ ...config, deadlineMs: 60_000 }, {}).deadlineMs).toBe(60_000);
+    expect(mergeConfig({ ...config, deadlineMs: 60_000 }, { deadlineMs: 30_000 }).deadlineMs).toBe(30_000);
+  });
+
   it('falls back to knob defaults of 1/1/0/participant when neither config nor CLI set them', () => {
     const bare: BenchmarkConfig = { benchmarkSlug: 's', benchmarkName: 'n', participants: [] };
     expect(mergeConfig(bare, {})).toEqual({ iterations: 1, concurrency: 1, staggerDelayMs: 0, groupBy: 'participant', providers: undefined });
@@ -520,6 +534,43 @@ describe('runBenchmark', () => {
     expect(calls.runWorker[0].concurrency).toBe(5);
   });
 
+  it('forwards a stopRequested predicate to the platform worker when deadlineMs is set', async () => {
+    const config: BenchmarkConfig<typeof participants[number]> = {
+      benchmarkSlug: 's',
+      benchmarkName: 'n',
+      iterations: 1,
+      deadlineMs: 1,
+      participants: [participants[0]],
+    };
+
+    await runBenchmark(
+      config,
+      defineTask(async () => {
+        await new Promise((resolve) => setTimeout(resolve, 10));
+        return {};
+      }),
+      [],
+    );
+
+    expect(calls.runWorker).toHaveLength(1);
+    expect(calls.runWorker[0].stopRequested).toBeInstanceOf(Function);
+    // The mock ignores the predicate, but by now the 1ms deadline has passed.
+    expect(calls.runWorker[0].stopRequested!()).toBe(true);
+  });
+
+  it('omits stopRequested from the platform worker when no deadline is set', async () => {
+    const config: BenchmarkConfig<typeof participants[number]> = {
+      benchmarkSlug: 's',
+      benchmarkName: 'n',
+      iterations: 1,
+      participants: [participants[0]],
+    };
+
+    await runBenchmark(config, defineTask(async () => ({})), []);
+
+    expect(calls.runWorker[0].stopRequested!()).toBe(false);
+  });
+
   it('reports under --slug/--name instead of the config slug and name', async () => {
     const config: BenchmarkConfig<typeof participants[number]> = {
       benchmarkSlug: 'sandbox-tti-local',
@@ -1306,6 +1357,47 @@ describe('runBenchmark', () => {
       expect(outcome.participants[0].records).toHaveLength(2);
     });
 
+    it('deadlineMs stops launching tasks and finishes with partial results, not an error', async () => {
+      const task = vi.fn(async () => {
+        await new Promise((resolve) => setTimeout(resolve, 30));
+        return { data: { ok: true } };
+      });
+      const config: BenchmarkConfig<typeof participants[number]> = {
+        benchmarkSlug: 's',
+        benchmarkName: 'n',
+        iterations: 5,
+        concurrency: 1,
+        deadlineMs: 10,
+        participants: [participants[0]],
+      };
+
+      const outcome = await runBenchmark(config, defineTask(task), ['--no-ingest']);
+
+      // The 10ms deadline lets the first ~30ms task finish, then blocks the rest.
+      expect(outcome.participants[0].records).toHaveLength(1);
+      expect(outcome.participants[0].records[0].status).toBe('success');
+      expect(outcome.stoppedEarly).toBe(true);
+    });
+
+    it('deadlineMs via --deadline-ms flag stops round-mode runs too', async () => {
+      const task = vi.fn(async () => {
+        await new Promise((resolve) => setTimeout(resolve, 30));
+        return { data: { ok: true } };
+      });
+      const config: BenchmarkConfig<typeof participants[number]> = {
+        benchmarkSlug: 's',
+        benchmarkName: 'n',
+        iterations: 5,
+        groupBy: 'round',
+        participants: [participants[0]],
+      };
+
+      const outcome = await runBenchmark(config, defineTask(task), ['--no-ingest', '--deadline-ms', '10']);
+
+      expect(outcome.participants[0].records).toHaveLength(1);
+      expect(outcome.stoppedEarly).toBe(true);
+    });
+
     it('honors BENCHSDK_NO_INGEST=1 without a CLI flag', async () => {
       process.env.BENCHSDK_NO_INGEST = '1';
       const task = vi.fn(async () => ({}));

packages/benchsdk-runner/src/bench-config.ts

@@ -13,6 +13,7 @@
  *   iterations       total tasks to run (default 1)
  *   concurrency      max tasks in flight at once — 1 = sequential, N = burst (default 1)
  *   staggerDelayMs   delay each task's start by taskIndex * staggerDelayMs (default 0)
+ *   deadlineMs       wall-clock budget; on expiry the run drains and finishes, not fails (default none)
  *
  * Common shapes:
  *   sequential  { iterations: N, concurrency: 1 }
@@ -190,6 +191,8 @@ export interface ParticipantRecords {
 /** The orchestration knobs a run actually used, after CLI overrides. */
 export interface ResolvedRunConfig {
   iterations: number;
+  /** Wall-clock budget for task execution in ms, from `deadlineMs`/`--deadline-ms`. */
+  deadlineMs?: number;
   /**
    * Iterations each phase runs, when the benchmark declares `phases` and
    * `--iterations` overrode their configured counts. A phase is one arm of a
@@ -213,6 +216,8 @@ export interface BenchmarkRunOutcome {
   dashboardUrl: string;
   participants: ParticipantRecords[];
   config: ResolvedRunConfig;
+  /** True when the run's `deadlineMs` expired and unstarted tasks were skipped. */
+  stoppedEarly?: boolean;
 }
 
 /**
@@ -252,6 +257,14 @@ export interface BenchmarkConfig<T extends BaseParticipant = BaseParticipant> {
   concurrency?: number;
   /** Delay each task's start by `taskIndex * staggerDelayMs`. Default: 0. */
   staggerDelayMs?: number;
+  /**
+   * Wall-clock budget for task execution in ms (e.g. 60 * 60_000 for an hour),
+   * overridable with `--deadline-ms`. When it expires the run stops launching
+   * new tasks, lets in-flight ones finish, and still flushes, reports, and
+   * exits normally — a deadline is a planned end, not a failure. The outcome's
+   * `stoppedEarly` flags the truncation. Omit for no deadline.
+   */
+  deadlineMs?: number;
   /**
    * Task ordering across participants. Default: 'participant' (run each
    * participant's tasks to completion, then the next). 'round' takes turns:
@@ -693,6 +706,9 @@ export function validateBenchmarkConfig<T extends BaseParticipant = BaseParticip
   if (config.staggerDelayMs !== undefined && (!Number.isFinite(config.staggerDelayMs) || config.staggerDelayMs < 0)) {
     issues.push({ field: 'staggerDelayMs', message: `must be a number >= 0 (got ${config.staggerDelayMs})` });
   }
+  if (config.deadlineMs !== undefined && (!Number.isFinite(config.deadlineMs) || config.deadlineMs < 1)) {
+    issues.push({ field: 'deadlineMs', message: `must be a number >= 1 (got ${config.deadlineMs})` });
+  }
   if (config.groupBy !== undefined && config.groupBy !== 'participant' && config.groupBy !== 'round') {
     issues.push({ field: 'groupBy', message: `must be 'participant' or 'round' (got ${config.groupBy})` });
   }

packages/benchsdk-runner/src/runner.ts

@@ -69,6 +69,8 @@ export interface CliArgs {
   iterations?: number;
   concurrency?: number;
   staggerDelayMs?: number;
+  /** Wall-clock budget for task execution in ms (`--deadline-ms`). */
+  deadlineMs?: number;
   groupBy?: GroupBy;
   /** Participant names from `--provider a,b` (repeatable). */
   providers?: string[];
@@ -81,6 +83,32 @@ function isEnvNoIngest(): boolean {
   return v === '1' || v?.toLowerCase() === 'true';
 }
 
+/**
+ * A run-level wall-clock deadline (`deadlineMs`/`--deadline-ms`). Once expired
+ * the schedulers stop launching new tasks; in-flight tasks still finish and
+ * the run still flushes, reports, and exits 0 — a deadline is a planned end,
+ * not a failure.
+ */
+interface RunDeadline {
+  expired(): boolean;
+}
+
+function createDeadline(deadlineMs: number | undefined): RunDeadline | undefined {
+  if (deadlineMs === undefined) return undefined;
+  const expiresAt = Date.now() + deadlineMs;
+  let announced = false;
+  return {
+    expired() {
+      if (Date.now() < expiresAt) return false;
+      if (!announced) {
+        announced = true;
+        console.log(`  Run deadline reached (${deadlineMs}ms) — finishing in-flight tasks; no new tasks will start.`);
+      }
+      return true;
+    },
+  };
+}
+
 function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
@@ -326,6 +354,12 @@ export function parseCliArgs(argv: string[], allowedCustomFlags?: readonly strin
         i = nextIndex;
         break;
       }
+      case '--deadline-ms': {
+        const { value, nextIndex } = readValue(arg, i);
+        args.deadlineMs = intFlag(value, '--deadline-ms');
+        i = nextIndex;
+        break;
+      }
       case '--group-by': {
         const { value, nextIndex } = readValue(arg, i);
         if (value !== 'participant' && value !== 'round') {
@@ -411,6 +445,7 @@ export function mergeConfig<T extends BaseParticipant>(
     phaseIterations,
     concurrency: args.concurrency ?? config.concurrency ?? 1,
     staggerDelayMs: args.staggerDelayMs ?? config.staggerDelayMs ?? 0,
+    deadlineMs: args.deadlineMs ?? config.deadlineMs,
     groupBy: args.groupBy ?? config.groupBy ?? 'participant',
     providers: args.providers ?? config.defaultProviders,
   };
@@ -581,6 +616,7 @@ export function runConfigToJson<T extends BaseParticipant>(
     ...(!config.phases ? { iterations: resolved.iterations } : {}),
     concurrency: resolved.concurrency,
     staggerDelayMs: resolved.staggerDelayMs,
+    ...(resolved.deadlineMs !== undefined ? { deadlineMs: resolved.deadlineMs } : {}),
     groupBy: resolved.groupBy,
     ...(config.dimensions ? { dimensions: config.dimensions } : {}),
     ...(config.scoring ? { scoring: config.scoring } : {}),
@@ -649,7 +685,8 @@ export async function runBenchmark<T extends BaseParticipant>(
   }
   console.log(
     `Knobs: iterations=${totalTasks}, concurrency=${concurrencyLabel}, ` +
-      `staggerDelayMs=${resolved.staggerDelayMs}, groupBy=${resolved.groupBy}\n`,
+      `staggerDelayMs=${resolved.staggerDelayMs}, groupBy=${resolved.groupBy}` +
+      `${resolved.deadlineMs !== undefined ? `, deadlineMs=${resolved.deadlineMs}` : ''}\n`,
   );
 
   // Declaratively materialize the benchmark from the file/shape identity, which
@@ -730,20 +767,28 @@ export async function runBenchmark<T extends BaseParticipant>(
   }
 
   const onResult = defaultOnResult;
+  // The deadline covers task execution only — the setup above (auth, run
+  // creation) and the reporting below always run to completion.
+  const deadline = createDeadline(resolved.deadlineMs);
 
   let participantRecords: ParticipantRecords[];
   if (resolved.groupBy === 'round') {
-    participantRecords = await runGroupedByRound(config, schedule, available, resolved, client, runId, baseUrl, apiKey, token, orgSlug, orgId, onResult, noIngest);
+    participantRecords = await runGroupedByRound(config, schedule, available, resolved, client, runId, baseUrl, apiKey, token, orgSlug, orgId, onResult, noIngest, deadline);
   } else {
-    participantRecords = await runGroupedByParticipant(config, schedule, available, resolved, client, runId, onResult, noIngest);
+    participantRecords = await runGroupedByParticipant(config, schedule, available, resolved, client, runId, onResult, noIngest, deadline);
   }
 
-  console.log(`All done. ${noIngest ? 'No platform run created.' : `View at: ${dashboardUrl}`}`);
+  const stoppedEarly =
+    deadline !== undefined &&
+    deadline.expired() &&
+    participantRecords.some((p) => p.records.length < schedule.length);
+  console.log(`All done${stoppedEarly ? ' (stopped at deadline — partial results)' : ''}. ${noIngest ? 'No platform run created.' : `View at: ${dashboardUrl}`}`);
   const outcome: BenchmarkRunOutcome = {
     runId,
     dashboardUrl,
     participants: participantRecords,
     config: resolved,
+    ...(stoppedEarly ? { stoppedEarly: true } : {}),
   };
   if (!noIngest && client && (config.onScore || config.scoring)) {
     try {
@@ -850,9 +895,11 @@ async function runGroupedByParticipant<T extends BaseParticipant>(
   runId: string,
   onResult: OnResult,
   noIngest: boolean,
+  deadline?: RunDeadline,
 ): Promise<ParticipantRecords[]> {
   const participantRecords: ParticipantRecords[] = [];
   for (const participant of available) {
+    if (deadline?.expired()) break;
     console.log(`${'='.repeat(70)}`);
     console.log(`  Participant: ${participant.name}`);
     console.log('='.repeat(70));
@@ -877,7 +924,7 @@ async function runGroupedByParticipant<T extends BaseParticipant>(
       };
 
       const worker = async () => {
-        while (nextIndex < schedule.length) {
+        while (nextIndex < schedule.length && !deadline?.expired()) {
           const index = nextIndex++;
           await runSlot(index);
         }
@@ -939,6 +986,7 @@ async function runGroupedByParticipant<T extends BaseParticipant>(
         }
       },
       onResult: (record) => onResult(record, { iterations: schedule.length, participant: participant.name }),
+      stopRequested: () => deadline?.expired() ?? false,
     });
 
     if (!result.assignment) {
@@ -975,6 +1023,7 @@ async function runGroupedByRound<T extends BaseParticipant>(
   orgId: string | undefined,
   onResult: OnResult,
   noIngest: boolean = false,
+  deadline?: RunDeadline,
 ): Promise<ParticipantRecords[]> {
   const reporters = new Map<string, BenchmarkReporter | null>();
   const logBuffers = new Map<string, LogBuffer>();
@@ -1041,6 +1090,7 @@ async function runGroupedByRound<T extends BaseParticipant>(
   console.log(`Interleaving ${available.length} participant(s), ${schedule.length} round(s) each.\n`);
 
   for (let i = 0; i < schedule.length; i++) {
+    if (deadline?.expired()) break;
     const slot = schedule[i];
     // Round mode staggers a fixed delay between rounds (vs. participant mode's per-task stagger).
     if (resolved.staggerDelayMs > 0 && i > 0) {

packages/benchsdk-worker/src/worker.ts

@@ -145,10 +145,10 @@ function handleTelemetryError(handler: ((error: unknown, operation: string) => v
   console.warn(`[benchsdk] telemetry failure (${operation}): ${error instanceof Error ? error.message : String(error)}`);
 }
 
-async function mapPool<T>(items: T[], concurrency: number, fn: (item: T) => Promise<void>): Promise<void> {
+async function mapPool<T>(items: T[], concurrency: number, fn: (item: T) => Promise<void>, shouldStop?: () => boolean): Promise<void> {
   let nextIndex = 0;
   const workers = Array.from({ length: Math.min(concurrency, items.length) }, async () => {
-    while (nextIndex < items.length) {
+    while (nextIndex < items.length && !(shouldStop?.() ?? false)) {
       const item = items[nextIndex];
       nextIndex += 1;
       await fn(item);
@@ -591,7 +591,7 @@ export async function runWorker(client: BenchmarkClient, options: RunWorkerOptio
       pending.push(record);
       options.onResult?.(record);
       if (pending.length >= batchSize) await flush(false);
-    });
+    }, options.stopRequested);
 
     await flush(true);
     if (flushFailed || pending.length > 0) {