perf(ci): run Tier A coverage concurrently with isolatedModules ts-jest (#1701)

Tier A coverage runs projects a few at a time (largest first, bounded Jest workers, buffered output, fail-fast kept) and ts-jest transpiles with isolatedModules instead of type-checking per process; five type re-exports become export type, two decorated inputs use import type. Unit Tests and Typechecks job: 26 min -> 9 min (Tier A step 23 min -> 6.5 min).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tools/coverage/coverage-run-pool.mjs

@@ -0,0 +1,153 @@
+/**
+ * Scheduling helpers for tools/coverage/run-tier-a-coverage.mjs.
+ *
+ * Tier A used to run its ~33 projects one after another, each as its own
+ * `pnpm nx run <project>:test` process: 23 minutes in CI, most of it Jest and
+ * Nx start-up, ts-jest cache warm-up and idle workers on small projects. The
+ * runner now keeps a few projects in flight at once and gives each Jest a
+ * bounded worker count, so the runner's total CPU budget stays close to the
+ * machine's core count instead of multiplying with it.
+ */
+import { readdirSync } from 'node:fs';
+import path from 'node:path';
+
+const SPEC_FILE = /\.(spec|test)\.ts$/;
+
+/** Counts spec files under a directory; used to start the big projects first. */
+export function countSpecFiles(directory) {
+    let count = 0;
+    let entries;
+    try {
+        entries = readdirSync(directory, { withFileTypes: true });
+    } catch {
+        return 0;
+    }
+    for (const entry of entries) {
+        const fullPath = path.join(directory, entry.name);
+        if (entry.isDirectory()) {
+            count += countSpecFiles(fullPath);
+        } else if (entry.isFile() && SPEC_FILE.test(entry.name)) {
+            count += 1;
+        }
+    }
+    return count;
+}
+
+/**
+ * Longest-first ordering: with a bounded pool, a big project started last
+ * would run alone at the end while the other slots sit idle. Ties keep the
+ * policy order so the output stays stable.
+ */
+export function orderLongestFirst(projects, weightOf) {
+    return projects
+        .map((project, index) => ({ project, index, weight: weightOf(project) }))
+        .sort((a, b) => b.weight - a.weight || a.index - b.index)
+        .map((entry) => entry.project);
+}
+
+/**
+ * How many projects to keep in flight. Defaults to one less than the core
+ * count, capped at three: beyond that the per-process start-up cost is paid
+ * anyway and the Jest workers of the concurrent runs starve each other.
+ */
+export function resolveConcurrency({ requested, cpuCount }) {
+    if (Number.isInteger(requested) && requested > 0) return requested;
+    return Math.max(1, Math.min(3, cpuCount - 1));
+}
+
+/**
+ * Jest workers per project, so that concurrency × workers stays near the core
+ * count. Small projects never use them all, which is what leaves room for the
+ * other slots.
+ */
+export function resolveWorkersPerProject({ requested, concurrency, cpuCount }) {
+    if (Number.isInteger(requested) && requested > 0) return requested;
+    return Math.max(1, Math.ceil(cpuCount / concurrency));
+}
+
+/**
+ * Runs `tasks` (functions returning a promise of `{ status }`) with at most
+ * `concurrency` in flight. Fail-fast: after the first non-zero status no new
+ * task starts, but the ones already running are awaited so their output and
+ * coverage files are complete. Resolves with every started task's result in
+ * start order plus the names that were never started.
+ */
+export async function runWithConcurrency(tasks, { concurrency, onSettled }) {
+    const results = [];
+    const skipped = [];
+    let nextIndex = 0;
+    let failed = false;
+
+    async function worker() {
+        while (nextIndex < tasks.length) {
+            const index = nextIndex++;
+            const task = tasks[index];
+            if (failed) {
+                skipped.push(task.name);
+                continue;
+            }
+            const startedAt = Date.now();
+            let result;
+            try {
+                result = await task.run();
+            } catch (error) {
+                result = { status: 1, error };
+            }
+            const settled = {
+                name: task.name,
+                status: result.status,
+                error: result.error,
+                durationMs: Date.now() - startedAt,
+            };
+            results[index] = settled;
+            if (settled.status !== 0) failed = true;
+            onSettled?.(settled);
+        }
+    }
+
+    const workers = [];
+    for (let slot = 0; slot < Math.max(1, concurrency); slot += 1) {
+        workers.push(worker());
+    }
+    await Promise.all(workers);
+
+    return {
+        failed,
+        results: results.filter(Boolean),
+        skipped,
+    };
+}
+
+export function formatDuration(ms) {
+    const seconds = Math.round(ms / 1000);
+    const minutes = Math.floor(seconds / 60);
+    return minutes > 0 ? `${minutes}m ${String(seconds % 60).padStart(2, '0')}s` : `${seconds}s`;
+}
+
+/**
+ * Whole positive integers only: `3oops` and `2.5` are rejected rather than
+ * truncated, so a typo cannot silently apply a different resource budget.
+ */
+function parsePositiveInteger(raw, what) {
+    if (!/^\d+$/.test(raw.trim()) || Number.parseInt(raw, 10) < 1) {
+        throw new Error(
+            `${what} expects a positive integer, received "${raw}".`
+        );
+    }
+    return Number.parseInt(raw, 10);
+}
+
+/** Reads a positive integer from the environment; unset or empty means absent. */
+export function integerEnv(env, name) {
+    const raw = env[name];
+    if (raw === undefined || raw.trim() === '') return undefined;
+    return parsePositiveInteger(raw, name);
+}
+
+/** Parses `--flag=value` style integers; returns undefined when absent. */
+export function integerFlag(argv, name) {
+    const prefix = `--${name}=`;
+    const raw = argv.find((argument) => argument.startsWith(prefix));
+    if (raw === undefined) return undefined;
+    return parsePositiveInteger(raw.slice(prefix.length), prefix);
+}

tools/coverage/coverage-run-pool.test.mjs

@@ -0,0 +1,139 @@
+import assert from 'node:assert/strict';
+import { mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import { after, before, test } from 'node:test';
+
+import {
+    countSpecFiles,
+    formatDuration,
+    integerEnv,
+    integerFlag,
+    orderLongestFirst,
+    resolveConcurrency,
+    resolveWorkersPerProject,
+    runWithConcurrency,
+} from './coverage-run-pool.mjs';
+
+let workDir;
+before(async () => {
+    workDir = await mkdtemp(path.join(os.tmpdir(), 'coverage-run-pool-'));
+});
+after(async () => {
+    await rm(workDir, { recursive: true, force: true });
+});
+
+test('counts spec and test files recursively and ignores sources', async () => {
+    const root = path.join(workDir, 'lib');
+    await mkdir(path.join(root, 'nested', 'deeper'), { recursive: true });
+    await writeFile(path.join(root, 'a.spec.ts'), '');
+    await writeFile(path.join(root, 'a.ts'), '');
+    await writeFile(path.join(root, 'nested', 'b.test.ts'), '');
+    await writeFile(path.join(root, 'nested', 'deeper', 'c.spec.ts'), '');
+    await writeFile(path.join(root, 'nested', 'deeper', 'c.spec.ts.snap'), '');
+    assert.equal(countSpecFiles(root), 3);
+    assert.equal(countSpecFiles(path.join(workDir, 'missing')), 0);
+});
+
+test('orders longest first and keeps policy order for ties', () => {
+    const projects = [
+        { name: 'small' },
+        { name: 'big' },
+        { name: 'medium' },
+        { name: 'also-small' },
+    ];
+    const weights = { small: 2, big: 50, medium: 10, 'also-small': 2 };
+    assert.deepEqual(
+        orderLongestFirst(projects, (project) => weights[project.name]).map((p) => p.name),
+        ['big', 'medium', 'small', 'also-small']
+    );
+});
+
+test('derives concurrency and workers from the core count unless overridden', () => {
+    assert.equal(resolveConcurrency({ requested: undefined, cpuCount: 4 }), 3);
+    assert.equal(resolveConcurrency({ requested: undefined, cpuCount: 2 }), 1);
+    assert.equal(resolveConcurrency({ requested: undefined, cpuCount: 16 }), 3);
+    assert.equal(resolveConcurrency({ requested: 5, cpuCount: 2 }), 5);
+    assert.equal(resolveWorkersPerProject({ requested: undefined, concurrency: 3, cpuCount: 4 }), 2);
+    assert.equal(resolveWorkersPerProject({ requested: undefined, concurrency: 1, cpuCount: 4 }), 4);
+    assert.equal(resolveWorkersPerProject({ requested: 1, concurrency: 3, cpuCount: 16 }), 1);
+});
+
+function task(name, { delay = 0, status = 0, log }) {
+    return {
+        name,
+        run: () =>
+            new Promise((resolve) => {
+                log.push(`start ${name}`);
+                setTimeout(() => {
+                    log.push(`end ${name}`);
+                    resolve({ status });
+                }, delay);
+            }),
+    };
+}
+
+test('keeps at most `concurrency` tasks in flight and reports results in start order', async () => {
+    const log = [];
+    const tasks = [
+        task('a', { delay: 30, log }),
+        task('b', { delay: 10, log }),
+        task('c', { delay: 10, log }),
+        task('d', { delay: 5, log }),
+    ];
+    const settled = [];
+    const outcome = await runWithConcurrency(tasks, {
+        concurrency: 2,
+        onSettled: (result) => settled.push(result.name),
+    });
+    assert.equal(outcome.failed, false);
+    assert.deepEqual(outcome.skipped, []);
+    assert.deepEqual(outcome.results.map((r) => r.name), ['a', 'b', 'c', 'd']);
+    // b finishes before a, so the third task starts before a ends.
+    assert.ok(log.indexOf('start c') < log.indexOf('end a'));
+    assert.ok(log.indexOf('start c') > log.indexOf('end b'));
+    assert.ok(outcome.results.every((r) => r.status === 0 && r.durationMs >= 0));
+    assert.equal(settled.length, 4);
+});
+
+test('fails fast: a failure stops new tasks but lets running ones finish', async () => {
+    const log = [];
+    const tasks = [
+        task('a', { delay: 40, log }),
+        task('b', { delay: 5, status: 1, log }),
+        task('c', { delay: 5, log }),
+        task('d', { delay: 5, log }),
+    ];
+    const outcome = await runWithConcurrency(tasks, { concurrency: 2 });
+    assert.equal(outcome.failed, true);
+    assert.deepEqual(outcome.skipped, ['c', 'd']);
+    assert.deepEqual(outcome.results.map((r) => [r.name, r.status]), [['a', 0], ['b', 1]]);
+    assert.ok(log.includes('end a'), 'the running task was awaited');
+});
+
+test('a task that throws counts as a failure with the error attached', async () => {
+    const outcome = await runWithConcurrency(
+        [{ name: 'boom', run: () => Promise.reject(new Error('spawn failed')) }],
+        { concurrency: 1 }
+    );
+    assert.equal(outcome.failed, true);
+    assert.match(outcome.results[0].error.message, /spawn failed/);
+});
+
+test('formats durations and parses integer flags', () => {
+    assert.equal(formatDuration(4200), '4s');
+    assert.equal(formatDuration(125000), '2m 05s');
+    assert.equal(integerFlag(['--concurrency=3'], 'concurrency'), 3);
+    assert.equal(integerFlag(['--projects=a'], 'concurrency'), undefined);
+    assert.throws(() => integerFlag(['--concurrency=0'], 'concurrency'), /positive integer/);
+    assert.throws(() => integerFlag(['--max-workers=two'], 'max-workers'), /positive integer/);
+    assert.equal(integerEnv({}, 'TIER_A_CONCURRENCY'), undefined);
+    assert.equal(integerEnv({ TIER_A_CONCURRENCY: '' }, 'TIER_A_CONCURRENCY'), undefined);
+    assert.equal(integerEnv({ TIER_A_CONCURRENCY: '2' }, 'TIER_A_CONCURRENCY'), 2);
+    assert.throws(() => integerEnv({ TIER_A_CONCURRENCY: 'x' }, 'TIER_A_CONCURRENCY'), /positive integer/);    // Prefixes and fractions are rejected, not truncated.
+    assert.throws(() => integerFlag(['--concurrency=3oops'], 'concurrency'), /positive integer/);
+    assert.throws(() => integerFlag(['--concurrency=2.5'], 'concurrency'), /positive integer/);
+    assert.throws(() => integerEnv({ TIER_A_MAX_WORKERS: '2.5' }, 'TIER_A_MAX_WORKERS'), /positive integer/);
+    assert.throws(() => integerEnv({ TIER_A_MAX_WORKERS: '-1' }, 'TIER_A_MAX_WORKERS'), /positive integer/);
+    assert.equal(integerEnv({ TIER_A_MAX_WORKERS: ' 4 ' }, 'TIER_A_MAX_WORKERS'), 4);
+});

tools/coverage/run-tier-a-coverage.mjs

@@ -2,6 +2,7 @@
 
 import { spawn } from 'node:child_process';
 import { existsSync, readFileSync, rmSync } from 'node:fs';
+import os from 'node:os';
 import path from 'node:path';
 import process from 'node:process';
 
@@ -9,6 +10,16 @@ import {
     createCoverageOutputScanner,
     validateProjectCoverage,
 } from './coverage-integrity.mjs';
+import {
+    countSpecFiles,
+    formatDuration,
+    integerEnv,
+    integerFlag,
+    orderLongestFirst,
+    resolveConcurrency,
+    resolveWorkersPerProject,
+    runWithConcurrency,
+} from './coverage-run-pool.mjs';
 
 const workspaceRoot = process.cwd();
 const policyPath = path.join(workspaceRoot, 'tools/coverage/coverage-policy.json');
@@ -29,6 +40,24 @@ const tierAProjects = policy.unitCoverage.tierA.filter(
     (project) => requestedProjects.size === 0 || requestedProjects.has(project.name)
 );
 
+// Projects run a few at a time (see coverage-run-pool.mjs). Override with
+// --concurrency=N / --max-workers=N or TIER_A_CONCURRENCY / TIER_A_MAX_WORKERS
+// when a machine has more or less room than the defaults assume.
+const cpuCount = os.availableParallelism?.() ?? os.cpus().length;
+const concurrency = resolveConcurrency({
+    requested:
+        integerFlag(process.argv.slice(2), 'concurrency') ??
+        integerEnv(process.env, 'TIER_A_CONCURRENCY'),
+    cpuCount,
+});
+const workersPerProject = resolveWorkersPerProject({
+    requested:
+        integerFlag(process.argv.slice(2), 'max-workers') ??
+        integerEnv(process.env, 'TIER_A_MAX_WORKERS'),
+    concurrency,
+    cpuCount,
+});
+
 if (tierAProjects.length === 0) {
     console.error('No Tier A coverage projects matched the requested filters.');
     process.exit(1);
@@ -103,6 +132,7 @@ function buildNxArgs(project) {
             '--configuration=ci',
             '--codeCoverage',
             `--coverageDirectory=${coverageDirFor(project, 'workspace')}`,
+            `--maxWorkers=${workersPerProject}`,
             '--output-style=static',
         ];
     }
@@ -116,6 +146,7 @@ function buildNxArgs(project) {
             `${project.name}:test`,
             '--output-style=static',
             '--',
+            `--maxWorkers=${workersPerProject}`,
             ...collectCoverageArgs(project, jestRootMode),
         ];
     }
@@ -125,7 +156,12 @@ function buildNxArgs(project) {
     );
 }
 
-function spawnCoverage(args, scanner) {
+/**
+ * Output is buffered per project and written in one piece when the project
+ * finishes: with several Jest processes in flight, interleaved lines would be
+ * unreadable and the coverage-failure scanner would see other projects' text.
+ */
+function spawnCoverage(args, scanner, output) {
     return new Promise((resolve, reject) => {
         const child = spawn('pnpm', args, {
             cwd: workspaceRoot,
@@ -134,16 +170,16 @@ function spawnCoverage(args, scanner) {
                 CI: process.env.CI ?? 'true',
                 NX_TASKS_RUNNER_DYNAMIC_OUTPUT: 'false',
             },
-            stdio: ['inherit', 'pipe', 'pipe'],
+            stdio: ['ignore', 'pipe', 'pipe'],
         });
 
         child.stdout.on('data', (chunk) => {
             scanner.push(chunk);
-            process.stdout.write(chunk);
+            output.push(chunk);
         });
         child.stderr.on('data', (chunk) => {
             scanner.push(chunk);
-            process.stderr.write(chunk);
+            output.push(chunk);
         });
         child.once('error', reject);
         child.once('close', (code, signal) => {
@@ -152,35 +188,48 @@ function spawnCoverage(args, scanner) {
     });
 }
 
-async function collectProjectCoverage(project) {
+async function collectProjectCoverage(project, specCount) {
     const args = buildNxArgs(project);
-    console.log(`\n==> Collecting coverage for ${project.name}`);
-    console.log(`pnpm ${args.join(' ')}`);
+    const output = [];
+    // The start line goes out immediately so a stalled project is visible in
+    // the log before the job times out; its full output follows on completion.
+    console.log(
+        `==> Started ${project.name} (${specCount} spec files): pnpm ${args.join(' ')}`
+    );
+    const lines = [`\n==> Coverage for ${project.name}`, `pnpm ${args.join(' ')}`];
 
     const scanner = createCoverageOutputScanner();
-    const result = await spawnCoverage(args, scanner);
+    const result = await spawnCoverage(args, scanner, output);
     let failed = result.code !== 0 || result.signal !== null;
 
+    const flush = () => {
+        process.stdout.write(`${lines.join('\n')}\n`);
+        for (const chunk of output) process.stdout.write(chunk);
+        const last = output.at(-1);
+        if (last && !last.toString().endsWith('\n')) process.stdout.write('\n');
+    };
+
     if (scanner.collectionFailed) {
-        console.error(
-            `Coverage collection failed while testing ${project.name}.`
-        );
-        failed = true;
+        flush();
+        console.error(`Coverage collection failed while testing ${project.name}.`);
+        return { status: 1 };
     }
 
     if (failed) {
-        return result.code && result.code !== 0 ? result.code : 1;
+        flush();
+        return { status: result.code && result.code !== 0 ? result.code : 1 };
     }
 
     const validation = validateProjectCoverage({
         project,
         workspaceRoot,
     });
+    flush();
     for (const error of validation.errors) {
         console.error(`Error: ${error}`);
     }
 
-    return validation.errors.length === 0 ? 0 : 1;
+    return { status: validation.errors.length === 0 ? 0 : 1 };
 }
 
 for (const project of tierAProjects) {
@@ -198,9 +247,48 @@ if (requestedProjects.size === 0) {
     }
 }
 
-for (const project of tierAProjects) {
-    const status = await collectProjectCoverage(project);
-    if (status !== 0) {
-        process.exit(status);
+const specCounts = new Map(
+    tierAProjects.map((project) => [
+        project.name,
+        countSpecFiles(path.join(workspaceRoot, project.sourceRoot)),
+    ])
+);
+const ordered = orderLongestFirst(tierAProjects, (project) =>
+    specCounts.get(project.name)
+);
+console.log(
+    `Tier A coverage: ${ordered.length} projects, ${concurrency} in flight, ${workersPerProject} Jest workers each (${cpuCount} cores).`
+);
+const startedAt = Date.now();
+const outcome = await runWithConcurrency(
+    ordered.map((project) => ({
+        name: project.name,
+        run: () =>
+            collectProjectCoverage(project, specCounts.get(project.name)),
+    })),
+    {
+        concurrency,
+        onSettled: (settled) => {
+            console.log(
+                `<== ${settled.name} ${settled.status === 0 ? 'ok' : `failed (${settled.status})`} in ${formatDuration(settled.durationMs)}`
+            );
+            if (settled.error) console.error(settled.error);
+        },
     }
+);
+
+const longest = [...outcome.results].sort((a, b) => b.durationMs - a.durationMs);
+console.log(`\nTier A coverage finished in ${formatDuration(Date.now() - startedAt)} wall-clock; longest projects:`);
+for (const entry of longest.slice(0, 8)) {
+    console.log(`  ${formatDuration(entry.durationMs).padStart(7)}  ${entry.name}`);
+}
+if (outcome.skipped.length > 0) {
+    console.error(`Not started after the first failure: ${outcome.skipped.join(', ')}`);
+}
+if (outcome.failed) {
+    const first = outcome.results.find((entry) => entry.status !== 0);
+    // Set the exit code instead of calling process.exit(): the failing
+    // project's buffered output may still be queued on a stdout pipe, and an
+    // immediate exit would truncate exactly the log that explains the failure.
+    process.exitCode = first?.status && first.status !== 0 ? first.status : 1;
 }