Support partial benchmark retries and increase HTTP timeout

Signed-off-by: Ludvig Liljenberg <4257730+ludfjig@users.noreply.github.com>

Author: ludfjig

.github/workflows/benchmark.yml

@@ -67,6 +67,7 @@ jobs:
           name: run-identity
           path: artifacts/run.json
           if-no-files-found: error
+          overwrite: true
 
   # Build guest inputs once on KVM for all measurement configurations.
   producer:
@@ -101,6 +102,7 @@ jobs:
       - uses: actions/upload-artifact@v7
         with:
           name: guest-inputs
+          overwrite: true
           path: |
             js/src/wit/handler_wit.wasm
             js/default/handler.wasm
@@ -191,6 +193,7 @@ jobs:
         if: always()
         with:
           name: result-${{ matrix.platform }}-${{ matrix.runtime }}-${{ matrix.strategy }}
+          overwrite: true
           path: results/
           if-no-files-found: error
           retention-days: 14
@@ -216,8 +219,8 @@ jobs:
         with:
           pattern: result-*
           path: results/
-          merge-multiple: true
-      - run: node scripts/merge-results.ts --output run.json results/*/bundle.json
+      - name: Merge measurements
+        run: node scripts/merge-results.ts --output run.json results/*/*/bundle.json
       - uses: actions/upload-artifact@v7
         with:
           name: run-${{ github.run_id }}-${{ github.run_attempt }}

docs/README.md

@@ -272,7 +272,13 @@ complete benchmark run for the current candidate.
 
 `Benchmark Publication` runs trusted `main` scripts on hosted runners. Successful
 PR attempts are checked against GitHub workflow and commit metadata. All jobs
-must succeed in one attempt. Use **Re-run all jobs** for publishable retries.
+must have a successful latest result through the selected attempt. Use
+**Re-run failed jobs** to retain successful configurations, or rerun an individual
+job and its dependents. Each retry replaces its configuration's result artifact.
+Collection combines the artifacts from that workflow run.
+A failed retry blocks publication even if an earlier
+attempt succeeded. Source revisions and benchmark definitions must match.
+Use **Re-run all jobs** when required artifacts have expired.
 The executed Benchmark workflow must match the trusted workflow. A workflow
 change can require archival recovery after maintainer review and merge.
 

docs/results.md

@@ -9,6 +9,11 @@ parser when loading bundles.
 Store one immutable JSON bundle per workflow run and attempt. The identity is
 `run.id` plus `run.attempt`. A retry has its own bundle. Commit SHAs may repeat.
 
+`run.attempt` identifies the collection attempt. Partial retries retain earlier
+successful measurements from that workflow run. Source revisions and benchmark
+definitions must match across all shards. Each configuration has a stable artifact
+name. A retry replaces that artifact. Collection rejects duplicate configurations.
+
 | Field | Contents |
 | --- | --- |
 | `schemaVersion` | Storage format version, currently `1` |
@@ -20,7 +25,7 @@ Store one immutable JSON bundle per workflow run and attempt. The identity is
 | `catalog.platforms` | Stable platform IDs and display labels |
 | `catalog.metrics` | Metric IDs, units, direction and measurement method versions |
 | `benchmark` | Workload ID, version, settings, required metrics and expected matrix |
-| `runners` | Runner records scoped to this workflow attempt |
+| `runners` | Runner records for the selected measurements |
 | `measurements` | Results linked to a runner and runtime |
 
 Runtime and metric IDs are extensible strings. Lifecycle IDs are `reload`,

scripts/benchmark.ts

@@ -26,6 +26,11 @@ mkdirSync(directory, { recursive: true })
 const local = values['local-bundle'] ? parseRunBundle(JSON.parse(readFileSync(values['local-bundle'], 'utf8'))) : undefined
 if (local && local.source !== 'local') throw new Error('Local collection requires a local bundle')
 const run = local?.run ?? JSON.parse(readFileSync(values.run, 'utf8')) as RunBundle['run']
+if (!local && process.env.GITHUB_ACTIONS) {
+  if (run.id !== process.env.GITHUB_RUN_ID || run.commit.sha !== process.env.GITHUB_SHA) throw new Error('Run identity differs from the current workflow')
+  run.attempt = z.coerce.number().int().positive().parse(process.env.GITHUB_RUN_ATTEMPT)
+  run.workflow.url = `https://github.com/${run.pullRequest!.repository}/actions/runs/${run.id}/attempts/${run.attempt}`
+}
 const executable = resolve('artifacts/bin', serverFlavor(runtime.id), 'http-bench')
 const oha = 'oha'
 const runner = await runnerMetadata(platform, !!local)

scripts/merge-results.ts

@@ -1,16 +1,42 @@
 import { readFileSync, writeFileSync } from 'node:fs'
+import { resolve } from 'node:path'
+import { pathToFileURL } from 'node:url'
 import { parseArgs } from 'node:util'
-import { canonicalJson, parseRunBundle, validatePublishableRun } from '../shared/results.ts'
+import { canonicalJson, parseRunBundle, validatePublishableRun, type RunBundle } from '../shared/results.ts'
 import { publicationPolicy } from '../shared/catalog.ts'
 
-const { values, positionals } = parseArgs({ allowPositionals: true, options: { output: { type: 'string', default: 'run.json' } } })
-if (!positionals.length) throw new Error('Provide bundle files to merge')
-const bundles = positionals.map(path => parseRunBundle(JSON.parse(readFileSync(path, 'utf8'))))
-const merged = structuredClone(bundles[0]!)
-for (const bundle of bundles.slice(1)) {
-  if (canonicalJson({ run: bundle.run, catalog: bundle.catalog, benchmark: bundle.benchmark, source: bundle.source }) !== canonicalJson({ run: merged.run, catalog: merged.catalog, benchmark: merged.benchmark, source: merged.source })) throw new Error('Shard definitions differ')
-  merged.runners.push(...bundle.runners)
-  merged.measurements.push(...bundle.measurements)
+export function mergeBundles(inputs: unknown[], attempt?: number, runId?: string): RunBundle {
+  if (!inputs.length) throw new Error('Provide bundle files to merge')
+  const bundles = inputs.map(parseRunBundle)
+  const merged = structuredClone(bundles[0]!)
+  const collectionAttempt = attempt ?? Math.max(...bundles.map(bundle => bundle.run.attempt))
+  if (!Number.isSafeInteger(collectionAttempt) || collectionAttempt < 1) throw new Error('Invalid collection attempt')
+  if (runId !== undefined && merged.run.id !== runId) throw new Error('Run identity differs from the current workflow')
+  const definition = (bundle: RunBundle) => {
+    const { attempt, createdAt, workflow, ...run } = bundle.run
+    const { url, ...workflowDefinition } = workflow
+    return canonicalJson({ run, workflow: workflowDefinition, catalog: bundle.catalog, benchmark: bundle.benchmark, source: bundle.source })
+  }
+  const expected = definition(merged)
+  merged.runners = []
+  merged.measurements = []
+  for (const bundle of bundles) {
+    if (definition(bundle) !== expected) throw new Error('Shard definitions differ')
+    if (bundle.run.attempt > collectionAttempt) throw new Error('Shard attempt exceeds collection attempt')
+    if (bundle.source === 'published' && bundle.run.workflow.url !== `https://github.com/${bundle.run.pullRequest?.repository}/actions/runs/${bundle.run.id}/attempts/${bundle.run.attempt}`) throw new Error('Shard workflow URL differs')
+    if (bundle.run.createdAt < merged.run.createdAt) merged.run.createdAt = bundle.run.createdAt
+    merged.runners.push(...bundle.runners)
+    merged.measurements.push(...bundle.measurements)
+  }
+  merged.run.attempt = collectionAttempt
+  if (merged.source === 'published') merged.run.workflow.url = `https://github.com/${merged.run.pullRequest!.repository}/actions/runs/${merged.run.id}/attempts/${collectionAttempt}`
+  return validatePublishableRun(merged, publicationPolicy)
 }
-writeFileSync(values.output, `${JSON.stringify(validatePublishableRun(merged, publicationPolicy), null, 2)}\n`, { flag: 'wx' })
-console.log(`Merged ${merged.measurements.length} successful measurements`)
+
+if (process.argv[1] && import.meta.url === pathToFileURL(resolve(process.argv[1])).href) {
+  const { values, positionals } = parseArgs({ allowPositionals: true, options: { output: { type: 'string', default: 'run.json' } } })
+  const attempt = process.env.GITHUB_ACTIONS ? Number(process.env.GITHUB_RUN_ATTEMPT) : undefined
+  const merged = mergeBundles(positionals.map(path => JSON.parse(readFileSync(path, 'utf8'))), attempt, process.env.GITHUB_ACTIONS ? process.env.GITHUB_RUN_ID : undefined)
+  writeFileSync(values.output, `${JSON.stringify(merged, null, 2)}\n`, { flag: 'wx' })
+  console.log(`Merged ${merged.measurements.length} successful measurements`)
+}

scripts/publication.test.ts

@@ -13,10 +13,59 @@ import { readJson, writeJson, withStoreLock } from './result-store.ts'
 import { assembleSite } from './assemble-site.ts'
 import { loadHistory } from '../src/data.ts'
 import { promoteRun, storeRun } from './publish-results.ts'
+import { mergeBundles } from './merge-results.ts'
 
 const repository = 'example/benchmarks'
 const root = resolve(import.meta.dirname, '..')
 
+test('partial retries retain successful measurements from the same source revision', () => {
+  const bundle = fixture()
+  const shards = bundle.measurements.map(measurement => {
+    const runner = structuredClone(bundle.runners.find(entry => entry.id === measurement.runnerId)!)
+    runner.id = measurement.id
+    return { ...structuredClone(bundle), runners: [runner], measurements: [{ ...structuredClone(measurement), runnerId: runner.id }] }
+  })
+  const retry = structuredClone(shards[0]!)
+  retry.run.attempt = 2
+  retry.run.workflow.url = `https://github.com/${repository}/actions/runs/${retry.run.id}/attempts/2`
+  retry.runners[0]!.name = 'retry-runner'
+  const measurement = shards[0]!.measurements[0]!
+  shards[0]!.measurements = [{ id: measurement.id, runnerId: measurement.runnerId, runtimeId: measurement.runtimeId, strategy: measurement.strategy, status: 'failed', reason: 'Transient failure', rawOutputs: [] }]
+  const retained = shards.slice(1)
+  const merged = mergeBundles([retry, ...retained], 2, bundle.run.id)
+  assert.equal(merged.run.attempt, 2)
+  assert.equal(merged.run.workflow.url, retry.run.workflow.url)
+  assert.equal(merged.measurements.length, 108)
+  assert.deepEqual(merged.measurements[0], retry.measurements[0])
+  assert.deepEqual(merged.measurements[1], shards[1]!.measurements[0])
+  assert.equal(merged.runners.find(runner => runner.id === retry.runners[0]!.id)!.name, 'retry-runner')
+  assert.throws(() => mergeBundles([retry, ...retained, structuredClone(retry)], 2), /Duplicate/)
+  assert.equal(mergeBundles([retry, ...retained], 3).run.attempt, 3)
+
+  const failedRetry = structuredClone(retry)
+  failedRetry.measurements = structuredClone(shards[0]!.measurements)
+  assert.throws(() => mergeBundles([failedRetry, ...retained], 2), /Missing successful configurations|Only successful/)
+  assert.throws(() => mergeBundles(shards, 2), /Missing successful configurations|Only successful/)
+  assert.throws(() => mergeBundles([...shards.slice(1, -1), retry], 2), /Missing successful configurations/)
+  assert.throws(() => mergeBundles([...shards, retry], 2), /Duplicate/)
+  assert.throws(() => mergeBundles([retry, ...retained], 1), /exceeds collection attempt/)
+  assert.throws(() => mergeBundles([retry, ...retained], 2, 'another-run'), /Run identity differs/)
+  for (const mutate of [
+    (input: typeof retry) => { input.run.id = 'another-run' },
+    (input: typeof retry) => { input.run.commit.sha = 'f'.repeat(40) },
+    (input: typeof retry) => { input.run.commit.tree = 'f'.repeat(40) },
+    (input: typeof retry) => { input.run.pullRequest!.base = 'f'.repeat(40) },
+    (input: typeof retry) => { input.benchmark.settings.concurrency = 1 },
+  ]) {
+    const invalid = structuredClone(retry)
+    mutate(invalid)
+    assert.throws(() => mergeBundles([...retained, invalid], 2), /Shard definitions differ/)
+  }
+  const invalidUrl = structuredClone(retry)
+  invalidUrl.run.workflow.url = bundle.run.workflow.url
+  assert.throws(() => mergeBundles([...retained, invalidUrl], 2), /Shard workflow URL differs/)
+})
+
 test('local history loads measured results but cannot be published', async context => {
   const bundle = fixture()
   bundle.source = 'local'
@@ -327,13 +376,14 @@ test('CI archival and promotion with simulated GitHub and Git', async context =>
     ...Array.from({ length: 2 }, (_, index) => `prepare (${index})`),
     ...Array.from({ length: 108 }, (_, index) => `measure (${index})`),
   ].map(name => ({ name, conclusion: 'success' }))
+  let retryJobs: typeof jobs = []
   context.mock.method(globalThis, 'fetch', async (url: string) => {
     const parsed = new URL(url)
     const path = parsed.pathname.replace(`/repos/${repository}/`, '')
     if (path.endsWith('/jobs')) {
       const page = Number(parsed.searchParams.get('page'))
-      const response = structuredClone(jobs)
-      if (failedJob) response[0]!.conclusion = 'failure'
+      const response = structuredClone(path.includes('/attempts/2/') ? retryJobs : jobs)
+      if (failedJob && !path.includes('/attempts/2/')) response[0]!.conclusion = 'failure'
       return Response.json({ jobs: response.slice((page - 1) * 100, page * 100), total_count: response.length })
     }
     const responses: Record<string, unknown> = {
@@ -344,6 +394,7 @@ test('CI archival and promotion with simulated GitHub and Git', async context =>
       'issues/7/events': [{ event: 'labeled', label: { name: 'benchmarks: skip' }, actor: { login: 'maintainer' } }],
       'collaborators/maintainer/permission': { permission: 'write' },
       'actions/runs/12345/attempts/1': { id: 12345, event: 'pull_request', conclusion: 'success', path: '.github/workflows/benchmark.yml', head_sha: workflowHead, created_at: bundle.run.createdAt },
+      'actions/runs/12345/attempts/2': { id: 12345, event: 'pull_request', conclusion: 'success', path: '.github/workflows/benchmark.yml', head_sha: workflowHead, created_at: bundle.run.createdAt },
       'contents/.github/workflows/benchmark.yml': { encoding: 'base64', content: Buffer.from(workflow).toString('base64') },
       [`git/commits/${bundle.run.commit.sha}`]: { tree: { sha: bundle.run.commit.tree }, parents: [{ sha: bundle.run.pullRequest!.base }, { sha: bundle.run.pullRequest!.head }] },
       [`git/commits/${pr.merge_commit_sha}`]: { tree: { sha: mergedTree } },
@@ -424,6 +475,36 @@ test('CI archival and promotion with simulated GitHub and Git', async context =>
     assert.equal(downloads, 0)
     assert.equal(pushes, 0)
   })
+  await context.test('partial retry archives and promotes retained successful jobs', async () => {
+    clearStore()
+    const originalUrl = bundle.run.workflow.url
+    bundle.run.attempt = 2
+    bundle.run.workflow.url = `https://github.com/${repository}/actions/runs/12345/attempts/2`
+    retryJobs = jobs.filter(job => ['eligibility', 'measure (0)', 'collect', 'Benchmark Status'].includes(job.name))
+    failedJob = true
+    writeJson(eventPath, { workflow_run: { ...notification.workflow_run, run_attempt: 2 } }, false)
+    try {
+      await main()
+      assert.equal(downloads, 1)
+      assert.equal(pushes, 2)
+      assert.deepEqual(readJson(resolve(temporary, 'data-store/index.json')), { schemaVersion: 1, runs: [{ id: '12345', attempt: 2 }] })
+      const stored = validateCompleteRun(readJson(resolve(temporary, 'data-store/runs/12345/2.json')))
+      assert.deepEqual(stored.measurements, bundle.measurements)
+      retryJobs = retryJobs.map(job => ({ ...job, conclusion: job.name === 'measure (0)' ? 'failure' : 'success' }))
+      await assert.rejects(main(), /successful measure/)
+      assert.equal(downloads, 1, 'A failed retry must not use an earlier successful job')
+      retryJobs = retryJobs.filter(job => job.name !== 'eligibility')
+      await assert.rejects(main(), /successful eligibility/)
+      assert.equal(downloads, 1)
+    } finally {
+      bundle.run.attempt = 1
+      bundle.run.workflow.url = originalUrl
+      retryJobs = []
+      failedJob = false
+      writeJson(eventPath, notification, false)
+      clearStore()
+    }
+  })
   await context.test('label changes prevent archival and promotion', async () => {
     pr.labels = [{ name: 'benchmarks: skip' }]
     await main()

scripts/publish-ci.ts

@@ -48,21 +48,28 @@ async function verifiedRun(runId: number, attempt: number) {
   if (workflow.encoding !== 'base64' || Buffer.from(workflow.content, 'base64').toString('utf8') !== readFileSync('.github/workflows/benchmark.yml', 'utf8')) {
     throw new Error('Benchmark workflow differs from the trusted workflow. Maintainer review and a fresh run are required.')
   }
-  const jobs: any[] = []
-  for (let page = 1; ; page++) {
-    const response = await github(`actions/runs/${runId}/attempts/${attempt}/jobs?per_page=100&page=${page}`)
-    jobs.push(...response.jobs)
-    if (jobs.length >= response.total_count) break
-    if (!response.jobs.length) throw new Error('Incomplete workflow job list')
+  const latestJobs = new Map<string, any>()
+  for (let currentAttempt = attempt; currentAttempt >= 1; currentAttempt--) {
+    let fetched = 0
+    for (let page = 1; ; page++) {
+      const response = await github(`actions/runs/${runId}/attempts/${currentAttempt}/jobs?per_page=100&page=${page}`)
+      for (const job of response.jobs) {
+        if (!latestJobs.has(job.name)) latestJobs.set(job.name, job)
+      }
+      fetched += response.jobs.length
+      if (fetched >= response.total_count) break
+      if (!response.jobs.length) throw new Error('Incomplete workflow job list')
+    }
   }
+  const jobs = [...latestJobs.values()]
   const expected = new Map([
     ['eligibility', 1], ['configure', 1], ['producer', 1], ['prepare', 2],
     ['measure', publicationPolicy.benchmark.expectedConfigurations.length], ['collect', 1], ['Benchmark Status', 1],
   ])
   for (const [name, count] of expected) {
     const matching = jobs.filter(job => job.name === name || job.name.startsWith(`${name} (`))
     if (matching.length !== count || matching.some(job => job.conclusion !== 'success')) {
-      throw new Error(`Attempt ${runId}.${attempt} requires ${count} successful ${name} jobs. Rerun all jobs.`)
+      throw new Error(`Run ${runId} through attempt ${attempt} requires ${count} successful ${name} jobs. Rerun the failed jobs.`)
     }
   }
   return run

shared/catalog.ts

@@ -50,7 +50,7 @@ export const benchmark = {
   id: 'http-redirect', version: 2,
   settings: {
     durationSeconds: 60, concurrency: 50, poolSize: 4, workerThreads: 'available-parallelism', workerReadiness: 'all-workers',
-    sandboxTimeoutMs: 1000, timeoutCheckIntervalMs: 10, clientTimeoutSeconds: 10,
+    sandboxTimeoutMs: 1000, timeoutCheckIntervalMs: 10, clientTimeoutSeconds: 30,
     memorySampleIntervalMs: 500, memorySampleAtShutdown: true, readinessRequests: 1, warmupRequests: 0,
     waitOngoingRequests: false, repetitions: 1, hyperlightAotMapping: 'linux-file-mapping-per-worker',
   },