fix(ci): globally cap artifact download concurrency, fix duplicate-named lcov function merge

Addresses two Codex/Datadog Autotest review findings on PR #9197:

- The per-`downloadArtifacts`-call worker pool let each concurrently processed
  sibling workflow open its own 10-download burst, recreating the aggregate
  GitHub API pressure the cap was meant to prevent. A module-scoped Semaphore
  now bounds every in-flight call together.
- `mergeLcovRecord` keyed functions by name only, so two functions sharing a
  name at different lines (e.g. two closures both named `shared`) collapsed
  into one, undercounting FNF/FNH. FN/FNDA are now paired positionally and
  keyed by `line,name`.

Generated by Claude Code.

Author: rochdev

scripts/download-artifacts.mjs

@@ -15,9 +15,44 @@ const execFileAsync = promisify(execFile)
 // GitHub's connection/rate limits and fail every request for a run with a generic `fetch failed`
 // (undici's error for a dropped connection), even though a smaller burst succeeds fine. Capping
 // concurrency keeps the burst size sane; retrying absorbs the transient failures that still slip
-// through.
+// through. All Green processes more than one sibling workflow's downloads concurrently (see
+// `all-green.mjs`'s `scheduleProcessing`), so the cap has to hold across every in-flight
+// `downloadArtifacts` call, not just within one of them — a per-call limit still lets N concurrent
+// runs each open their own 10, recreating the exact burst this exists to prevent.
 const MAX_CONCURRENT_DOWNLOADS = 10
 
+/**
+ * Bounds how many downloads run at once across every concurrent `downloadArtifacts` call in this
+ * process, instead of per call.
+ */
+class Semaphore {
+  #permits
+  #queue = []
+
+  constructor (permits) {
+    this.#permits = permits
+  }
+
+  acquire () {
+    if (this.#permits > 0) {
+      this.#permits--
+      return Promise.resolve()
+    }
+    return new Promise(resolve => this.#queue.push(resolve))
+  }
+
+  release () {
+    const next = this.#queue.shift()
+    if (next) {
+      next()
+    } else {
+      this.#permits++
+    }
+  }
+}
+
+const downloadSemaphore = new Semaphore(MAX_CONCURRENT_DOWNLOADS)
+
 /**
  * Download and unzip a single artifact, retrying on failure with a backoff delay.
  *
@@ -60,32 +95,6 @@ async function downloadOne ({ runId, artifact, owner, repo, token, retries, dela
   }
 }
 
-/**
- * Pull tasks off the front of `tasks` one at a time until it's empty, so at most one task per
- * worker is ever in flight.
- *
- * @param {Array<() => Promise<void>>} tasks
- * @returns {Promise<void>}
- */
-async function worker (tasks) {
-  const task = tasks.shift()
-  if (!task) return
-  await task()
-  return worker(tasks)
-}
-
-/**
- * Run a bounded number of `tasks` at a time instead of firing every one at once.
- *
- * @param {Array<() => Promise<void>>} tasks
- * @param {number} limit
- * @returns {Promise<void>}
- */
-async function runWithConcurrencyLimit (tasks, limit) {
-  const queue = [...tasks]
-  await Promise.all(Array.from({ length: Math.min(limit, queue.length) }, () => worker(queue)))
-}
-
 /**
  * @param {import('octokit').Octokit} octokit
  * @param {object} opts
@@ -113,13 +122,19 @@ export async function downloadArtifacts (octokit, { owner, repo, token, runs, re
   )
 
   let failed = 0
-  await runWithConcurrencyLimit(
-    toDownload.map(({ runId, artifact }) => async () => {
-      const ok = await downloadOne({ runId, artifact, owner, repo, token, retries, delayMs })
-      if (!ok) failed++
-    }),
-    MAX_CONCURRENT_DOWNLOADS
+  await Promise.all(
+    toDownload.map(async ({ runId, artifact }) => {
+      await downloadSemaphore.acquire()
+      try {
+        const ok = await downloadOne({ runId, artifact, owner, repo, token, retries, delayMs })
+        if (!ok) failed++
+      } finally {
+        downloadSemaphore.release()
+      }
+    })
   )
 
   return { downloaded: toDownload.length - failed, failed }
 }
+
+export { Semaphore }

scripts/download-artifacts.spec.mjs

@@ -0,0 +1,61 @@
+import assert from 'node:assert/strict'
+import { setTimeout as sleep } from 'node:timers/promises'
+
+import { describe, it } from 'mocha'
+
+import { Semaphore } from './download-artifacts.mjs'
+
+// Acquires `semaphore`, records the peak number of concurrently active holders onto `peakTracker`,
+// holds briefly, then releases.
+async function acquireAndTrackPeak (semaphore, peakTracker) {
+  await semaphore.acquire()
+  peakTracker.active++
+  peakTracker.peak = Math.max(peakTracker.peak, peakTracker.active)
+  await sleep(5)
+  peakTracker.active--
+  semaphore.release()
+}
+
+describe('download-artifacts', () => {
+  describe('Semaphore', () => {
+    it('never lets more than the given number of permits run at once', async () => {
+      const semaphore = new Semaphore(2)
+      const peakTracker = { active: 0, peak: 0 }
+
+      await Promise.all(Array.from({ length: 5 }, () => acquireAndTrackPeak(semaphore, peakTracker)))
+
+      assert.equal(peakTracker.peak, 2)
+    })
+
+    it('holds the cap across independent batches acquiring concurrently', async () => {
+      // Regression test: All Green calls `downloadArtifacts` once per sibling workflow, and more
+      // than one call can be in flight at the same time (see `all-green.mjs`'s `scheduleProcessing`).
+      // A per-call limiter would let each batch open its own pool of permits; sharing one `Semaphore`
+      // module-wide is what keeps the aggregate bounded instead.
+      const semaphore = new Semaphore(3)
+      const peakTracker = { active: 0, peak: 0 }
+      const task = () => acquireAndTrackPeak(semaphore, peakTracker)
+
+      const batchA = Promise.all(Array.from({ length: 4 }, task))
+      const batchB = Promise.all(Array.from({ length: 4 }, task))
+      await Promise.all([batchA, batchB])
+
+      assert.equal(peakTracker.peak, 3)
+    })
+
+    it('runs every queued acquirer eventually', async () => {
+      const semaphore = new Semaphore(1)
+      const order = []
+
+      const task = async id => {
+        await semaphore.acquire()
+        order.push(id)
+        semaphore.release()
+      }
+
+      await Promise.all([task('a'), task('b'), task('c')])
+
+      assert.deepEqual(order.sort(), ['a', 'b', 'c'])
+    })
+  })
+})

scripts/group-coverage.mjs

@@ -104,8 +104,9 @@ function planCoverageGroups (files) {
 /**
  * @typedef {object} LcovFileRecord
  * @property {Map<string, number>} lines line number -> summed hit count
- * @property {Map<string, { line: string, count: number }>} functions function name -> declaration
- *   line and summed hit count
+ * @property {Map<string, { name: string, line: string, count: number }>} functions `line,name` ->
+ *   name, declaration line, and summed hit count — keyed by line as well as name because distinct
+ *   functions can share a name (e.g. two nested closures both named `shared`)
  * @property {Map<string, { hits: number, reached: boolean }>} branches `line,block,branch` -> summed
  *   hit count and whether any cell reported the enclosing block as reached (lcov uses `-` for a
  *   branch whose block was never reached, distinct from a reached block whose branch was never taken)
@@ -125,6 +126,9 @@ function splitLcovLine (line) {
 /**
  * Fold one `SF:`-delimited record's `DA:`/`FN:`/`FNDA:`/`BRDA:` lines into the running per-file
  * merge state, summing hit counts for lines/functions/branches already seen from earlier cells.
+ * `FNDA:` lines carry only a function name, not its declaration line, so a same-named function
+ * declared at two different lines can't be told apart by name alone; lcov writers emit `FNDA:` lines
+ * in the same order as their `FN:` declarations, so they're paired positionally instead.
  *
  * @param {string[]} recordLines
  * @param {Map<string, LcovFileRecord>} files source file path -> merge state
@@ -142,19 +146,23 @@ function mergeLcovRecord (recordLines, files, order) {
   }
   const record = files.get(path)
 
+  const declaredFunctionKeys = []
+  let functionDeclarationIndex = 0
   for (const line of recordLines) {
     const [tag, rest] = splitLcovLine(line)
     if (tag === 'DA') {
       const [lineNumber, count] = rest.split(',')
       record.lines.set(lineNumber, (record.lines.get(lineNumber) ?? 0) + Number(count))
     } else if (tag === 'FN') {
       const [lineNumber, name] = rest.split(',')
-      if (!record.functions.has(name)) record.functions.set(name, { line: lineNumber, count: 0 })
+      const key = `${lineNumber},${name}`
+      declaredFunctionKeys.push(key)
+      if (!record.functions.has(key)) record.functions.set(key, { name, line: lineNumber, count: 0 })
     } else if (tag === 'FNDA') {
-      const [count, name] = rest.split(',')
-      const fn = record.functions.get(name) ?? { line: '0', count: 0 }
-      fn.count += Number(count)
-      record.functions.set(name, fn)
+      const [count] = rest.split(',')
+      const key = declaredFunctionKeys[functionDeclarationIndex++]
+      if (key === undefined) continue
+      record.functions.get(key).count += Number(count)
     } else if (tag === 'BRDA') {
       const [lineNumber, block, branch, count] = rest.split(',')
       const key = `${lineNumber},${block},${branch}`
@@ -179,8 +187,8 @@ function mergeLcovRecord (recordLines, files, order) {
 function serializeLcovRecord (path, record) {
   const lines = [`SF:${path}`]
 
-  for (const [name, fn] of record.functions) lines.push(`FN:${fn.line},${name}`)
-  for (const [name, fn] of record.functions) lines.push(`FNDA:${fn.count},${name}`)
+  for (const fn of record.functions.values()) lines.push(`FN:${fn.line},${fn.name}`)
+  for (const fn of record.functions.values()) lines.push(`FNDA:${fn.count},${fn.name}`)
   if (record.functions.size > 0) {
     const hit = [...record.functions.values()].filter(fn => fn.count > 0).length
     lines.push(`FNF:${record.functions.size}`, `FNH:${hit}`)

scripts/group-coverage.spec.mjs

@@ -112,6 +112,18 @@ describe('group-coverage', () => {
       )
     })
 
+    it('keeps same-named functions declared at different lines distinct', () => {
+      const a = join(dir, 'a.info')
+      writeFileSync(
+        a,
+        'SF:shared.js\nFN:4,shared\nFN:9,shared\nFNF:2\nFNDA:2,shared\nFNDA:0,shared\nFNH:1\nend_of_record\n'
+      )
+      assert.equal(
+        mergeLcov([a]),
+        'SF:shared.js\nFN:4,shared\nFN:9,shared\nFNDA:2,shared\nFNDA:0,shared\nFNF:2\nFNH:1\nend_of_record\n'
+      )
+    })
+
     it('sums BRDA hit counts for the same branch, treating "-" as an unreached block', () => {
       const a = join(dir, 'a.info')
       const b = join(dir, 'b.info')