mirror of
https://github.com/getpaseo/paseo.git
synced 2026-07-29 12:01:31 +00:00
perf(cli): run CLI E2E tests in parallel (#708)
* perf(cli): run E2E tests in parallel via worker pool The custom CLI test runner ran 35 tsx test files sequentially, making the cli-tests CI job the longest in main CI (~17 minutes). Each test file already isolates its own daemon (ephemeral port + tmp PASEO_HOME), so parallelism was just gated by the runner. Replace the sequential recursion with a fixed-size worker pool (default concurrency=4 to match GitHub Actions standard runners; override via PASEO_CLI_TEST_CONCURRENCY). Buffer per-test stdout/stderr and flush as a contiguous block on completion so concurrent output stays readable, and report per-test wall clock plus the five slowest tests. Local wall clock drops from ~12-15 minutes serial to ~2:49 with concurrency=4. The slowest single test (05-agent-run, 49s) is now the floor; CI should land near 4-5 minutes. * fix(cli-tests): deflake 30-chat under load and shard CI across 3 runners `chat wait` reads the latest message id and then subscribes for newer messages. Under CI load the subprocess takes >1s to bootstrap, so the old test's 250ms head start before posting "second message" raced against that read. When the post landed first, the subprocess saw the second message as latest and timed out waiting for a newer one. Replace the brittle delay with a post-and-race loop: every iteration posts a fresh "second message" and races a 250ms tick against the wait promise, so whichever message lands after the snapshot wakes wait deterministically. CI was also slower than local (10.3min vs 2.8min). On 4-vCPU GHA runners, 35 sequential-CPU-time of ~1850s caps wall clock around 8 min even at concurrency=4. Shard the suite across 3 GHA runners via matrix strategy. The runner now reads PASEO_CLI_TEST_SHARD/SHARD_TOTAL and distributes files into buckets — known long-pole tests (05/06/11/13/14-...) round-robin first so they spread across shards instead of clustering by their numeric prefixes, then the remainder fills in the reverse direction to balance light load. Local run is unchanged (single shard by default). Expected per-shard wall on CI: ~2:30 + setup ≈ ~4:30 total.
This commit is contained in:
Mohamed Boudra
7
.github/workflows/ci.yml
vendored
7
.github/workflows/ci.yml
vendored
@@ -252,7 +252,12 @@ jobs:
|
||||
run: npm run test --workspace=@getpaseo/relay
|
||||
|
||||
cli-tests:
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
shard: [1, 2, 3]
|
||||
runs-on: ubuntu-latest
|
||||
name: cli-tests (shard ${{ matrix.shard }}/3)
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
@@ -276,3 +281,5 @@ jobs:
|
||||
PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD: "0"
|
||||
PASEO_DICTATION_ENABLED: "0"
|
||||
PASEO_VOICE_MODE_ENABLED: "0"
|
||||
PASEO_CLI_TEST_SHARD: ${{ matrix.shard }}
|
||||
PASEO_CLI_TEST_SHARD_TOTAL: "3"
|
||||
|
||||
@@ -44,10 +44,25 @@ try {
|
||||
const readPayload = JSON.parse(readJson.stdout);
|
||||
assert.strictEqual(readPayload[0]?.author, "00000000-0000-4000-8000-000000000111");
|
||||
|
||||
const waitPromise = ctx.paseo(["chat", "wait", "coord-room", "--timeout", "5s"]);
|
||||
await new Promise((resolve) => setTimeout(resolve, 250));
|
||||
const secondPost = await ctx.paseo(["chat", "post", "coord-room", "second message"]);
|
||||
assert.strictEqual(secondPost.exitCode, 0, secondPost.stderr);
|
||||
// `chat wait` reads "latest message id" then subscribes for newer ones.
|
||||
// Under CI load the subprocess can take >1s to bootstrap, so a single
|
||||
// delayed post races against the read. Post repeatedly and race against
|
||||
// wait — every post is newer than the snapshot wait took, so one of them
|
||||
// always wakes it up.
|
||||
const waitPromise = ctx.paseo(["chat", "wait", "coord-room", "--timeout", "30s"]);
|
||||
const waitSentinel = waitPromise.then(() => "settled" as const);
|
||||
let postedSecond = false;
|
||||
for (let attempt = 0; attempt < 60; attempt++) {
|
||||
const post = await ctx.paseo(["chat", "post", "coord-room", "second message"]);
|
||||
assert.strictEqual(post.exitCode, 0, post.stderr);
|
||||
const tick = new Promise<"tick">((resolve) => setTimeout(() => resolve("tick"), 250));
|
||||
const result = await Promise.race([waitSentinel, tick]);
|
||||
if (result === "settled") {
|
||||
postedSecond = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert(postedSecond, "chat wait did not return after repeated posts");
|
||||
|
||||
const waited = await waitPromise;
|
||||
assert.strictEqual(waited.exitCode, 0, waited.stderr);
|
||||
|
||||
@@ -3,10 +3,12 @@
|
||||
/**
|
||||
* Test runner for Paseo CLI E2E tests
|
||||
*
|
||||
* Runs all test phases in sequence and reports results.
|
||||
* Each test is a separate .ts file that can also be run independently.
|
||||
* Runs all test phases as separate subprocesses with a bounded worker pool
|
||||
* so independent tests run concurrently. Each test file already isolates
|
||||
* its own daemon (ephemeral port + tmp PASEO_HOME), so parallelism is safe.
|
||||
*/
|
||||
|
||||
import { spawn } from "child_process";
|
||||
import { $ } from "zx";
|
||||
import { readdir, writeFile } from "fs/promises";
|
||||
import { join, dirname } from "path";
|
||||
@@ -20,6 +22,29 @@ const testEnvDefaults = {
|
||||
PASEO_VOICE_MODE_ENABLED: process.env.PASEO_VOICE_MODE_ENABLED ?? "0",
|
||||
};
|
||||
|
||||
const DEFAULT_CONCURRENCY = 4;
|
||||
const concurrencyEnv = process.env.PASEO_CLI_TEST_CONCURRENCY;
|
||||
const parsedConcurrency = concurrencyEnv ? Number.parseInt(concurrencyEnv, 10) : NaN;
|
||||
const concurrency =
|
||||
Number.isFinite(parsedConcurrency) && parsedConcurrency > 0
|
||||
? parsedConcurrency
|
||||
: DEFAULT_CONCURRENCY;
|
||||
|
||||
function parsePositiveInt(value: string | undefined, fallback: number): number {
|
||||
if (!value) return fallback;
|
||||
const parsed = Number.parseInt(value, 10);
|
||||
return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
|
||||
}
|
||||
|
||||
const shardTotal = parsePositiveInt(process.env.PASEO_CLI_TEST_SHARD_TOTAL, 1);
|
||||
const shardIndexRaw = parsePositiveInt(process.env.PASEO_CLI_TEST_SHARD, 1);
|
||||
if (shardIndexRaw < 1 || shardIndexRaw > shardTotal) {
|
||||
throw new Error(
|
||||
`PASEO_CLI_TEST_SHARD=${shardIndexRaw} out of range for SHARD_TOTAL=${shardTotal}`,
|
||||
);
|
||||
}
|
||||
const shardIndex = shardIndexRaw - 1;
|
||||
|
||||
let jsonOutputPath: string | null = null;
|
||||
for (let i = 0; i < args.length; i++) {
|
||||
const arg = args[i];
|
||||
@@ -96,15 +121,47 @@ console.log("=".repeat(50));
|
||||
|
||||
// Discover all test files
|
||||
const files = await readdir(__dirname);
|
||||
const testFiles = files.filter((f) => f.match(/^\d{2}-.*\.test\.ts$/)).sort();
|
||||
const allTestFiles = files.filter((f) => f.match(/^\d{2}-.*\.test\.ts$/)).sort();
|
||||
|
||||
if (testFiles.length === 0) {
|
||||
// Naive `index % shardTotal` round-robin clusters slow tests by accident
|
||||
// because their numeric prefixes (05, 06, 11, 13, 14) align with the stride.
|
||||
// Hand off the known long-pole tests round-robin first, then fill the
|
||||
// remainder in the reverse direction so shards heavy on slow tests get
|
||||
// fewer light tests. Update KNOWN_HEAVY_TESTS from the runner's "Slowest
|
||||
// tests" report when timings shift materially.
|
||||
const KNOWN_HEAVY_TESTS = new Set([
|
||||
"05-agent-run.test.ts",
|
||||
"06-agent-send.test.ts",
|
||||
"11-agent-wait.test.ts",
|
||||
"13-permit-allow-deny.test.ts",
|
||||
"14-worktree.test.ts",
|
||||
]);
|
||||
const heavyFiles = allTestFiles.filter((f) => KNOWN_HEAVY_TESTS.has(f));
|
||||
const otherFiles = allTestFiles.filter((f) => !KNOWN_HEAVY_TESTS.has(f));
|
||||
const shardBuckets: string[][] = Array.from({ length: shardTotal }, () => []);
|
||||
heavyFiles.forEach((f, i) => {
|
||||
shardBuckets[i % shardTotal].push(f);
|
||||
});
|
||||
otherFiles.forEach((f, i) => {
|
||||
shardBuckets[shardTotal - 1 - (i % shardTotal)].push(f);
|
||||
});
|
||||
const testFiles = shardBuckets[shardIndex];
|
||||
|
||||
if (allTestFiles.length === 0) {
|
||||
console.log("❌ No test files found");
|
||||
await writeJsonSummary({ passed: 0, failed: 0, failures: [] });
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
console.log(`Found ${testFiles.length} test file(s):\n`);
|
||||
if (testFiles.length === 0) {
|
||||
console.log(`❌ No test files for shard ${shardIndex + 1}/${shardTotal}`);
|
||||
await writeJsonSummary({ passed: 0, failed: 0, failures: [] });
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
console.log(
|
||||
`Shard ${shardIndex + 1}/${shardTotal}: ${testFiles.length} of ${allTestFiles.length} test file(s):\n`,
|
||||
);
|
||||
for (const file of testFiles) {
|
||||
console.log(` - ${file}`);
|
||||
}
|
||||
@@ -118,52 +175,119 @@ await runCommand("Building relay", "npm run build --workspace=@getpaseo/relay");
|
||||
await runCommand("Building server", "npm run build --workspace=@getpaseo/server");
|
||||
await runCommand("Building CLI", "npm run build --workspace=@getpaseo/cli");
|
||||
|
||||
type TestOutcome = { status: "passed" } | { status: "failed"; failure: Failure };
|
||||
type TestOutcome =
|
||||
| { status: "passed"; durationMs: number }
|
||||
| { status: "failed"; durationMs: number; failure: Failure };
|
||||
|
||||
function formatDuration(ms: number): string {
|
||||
if (ms < 1000) return `${ms}ms`;
|
||||
return `${(ms / 1000).toFixed(1)}s`;
|
||||
}
|
||||
|
||||
async function runSingleTest(testFile: string): Promise<TestOutcome> {
|
||||
const testPath = join(__dirname, testFile);
|
||||
const testName = testFile.replace(/\.test\.ts$/, "");
|
||||
const startedAt = Date.now();
|
||||
|
||||
console.log(`\n${"─".repeat(50)}`);
|
||||
console.log(`📋 Running ${testName}...`);
|
||||
console.log("─".repeat(50));
|
||||
return new Promise<TestOutcome>((resolve) => {
|
||||
const proc = spawn("npx", ["tsx", testPath], {
|
||||
env: {
|
||||
...process.env,
|
||||
PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD: testEnvDefaults.PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD,
|
||||
PASEO_DICTATION_ENABLED: testEnvDefaults.PASEO_DICTATION_ENABLED,
|
||||
PASEO_VOICE_MODE_ENABLED: testEnvDefaults.PASEO_VOICE_MODE_ENABLED,
|
||||
},
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
|
||||
try {
|
||||
const result =
|
||||
await $`PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD=${testEnvDefaults.PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD} PASEO_DICTATION_ENABLED=${testEnvDefaults.PASEO_DICTATION_ENABLED} PASEO_VOICE_MODE_ENABLED=${testEnvDefaults.PASEO_VOICE_MODE_ENABLED} npx tsx ${testPath}`.nothrow();
|
||||
if (result.exitCode === 0) {
|
||||
console.log(`\n✅ ${testName} PASSED`);
|
||||
return { status: "passed" };
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
proc.stdout.on("data", (chunk: Buffer) => {
|
||||
stdout += chunk.toString();
|
||||
});
|
||||
proc.stderr.on("data", (chunk: Buffer) => {
|
||||
stderr += chunk.toString();
|
||||
});
|
||||
|
||||
proc.on("error", (err) => {
|
||||
const durationMs = Date.now() - startedAt;
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
flushTestBlock(testName, durationMs, false, stdout, stderr || message);
|
||||
resolve({
|
||||
status: "failed",
|
||||
durationMs,
|
||||
failure: { test: testName, error: message },
|
||||
});
|
||||
});
|
||||
|
||||
proc.on("exit", (code) => {
|
||||
const durationMs = Date.now() - startedAt;
|
||||
const exitCode = code ?? 1;
|
||||
if (exitCode === 0) {
|
||||
flushTestBlock(testName, durationMs, true, stdout, stderr);
|
||||
resolve({ status: "passed", durationMs });
|
||||
return;
|
||||
}
|
||||
flushTestBlock(testName, durationMs, false, stdout, stderr);
|
||||
resolve({
|
||||
status: "failed",
|
||||
durationMs,
|
||||
failure: { test: testName, error: stderr || `Exit code: ${exitCode}` },
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function flushTestBlock(
|
||||
testName: string,
|
||||
durationMs: number,
|
||||
success: boolean,
|
||||
stdout: string,
|
||||
stderr: string,
|
||||
): void {
|
||||
const icon = success ? "✅" : "❌";
|
||||
const status = success ? "PASSED" : "FAILED";
|
||||
const lines: string[] = [];
|
||||
lines.push("─".repeat(50));
|
||||
lines.push(`📋 ${testName} (${formatDuration(durationMs)})`);
|
||||
lines.push("─".repeat(50));
|
||||
if (stdout) lines.push(stdout.trimEnd());
|
||||
if (!success && stderr) {
|
||||
lines.push("stderr:");
|
||||
lines.push(stderr.trimEnd());
|
||||
}
|
||||
lines.push(`${icon} ${testName} ${status}`);
|
||||
process.stdout.write(`${lines.join("\n")}\n\n`);
|
||||
}
|
||||
|
||||
console.log(
|
||||
`\nRunning tests with concurrency=${concurrency}, shard=${shardIndex + 1}/${shardTotal}\n`,
|
||||
);
|
||||
|
||||
const totalStart = Date.now();
|
||||
const queue = [...testFiles];
|
||||
const timings: { test: string; durationMs: number; status: "passed" | "failed" }[] = [];
|
||||
|
||||
async function worker(): Promise<void> {
|
||||
while (queue.length > 0) {
|
||||
const testFile = queue.shift();
|
||||
if (!testFile) return;
|
||||
const outcome = await runSingleTest(testFile);
|
||||
const test = testFile.replace(/\.test\.ts$/, "");
|
||||
timings.push({ test, durationMs: outcome.durationMs, status: outcome.status });
|
||||
if (outcome.status === "passed") {
|
||||
passed++;
|
||||
} else {
|
||||
failed++;
|
||||
failures.push(outcome.failure);
|
||||
}
|
||||
console.log(`\n❌ ${testName} FAILED (exit code: ${result.exitCode})`);
|
||||
if (result.stderr) {
|
||||
console.log("stderr:", result.stderr);
|
||||
}
|
||||
return {
|
||||
status: "failed",
|
||||
failure: { test: testName, error: result.stderr || `Exit code: ${result.exitCode}` },
|
||||
};
|
||||
} catch (e) {
|
||||
const error = e instanceof Error ? e.message : String(e);
|
||||
console.log(`\n❌ ${testName} FAILED`);
|
||||
console.log("Error:", error);
|
||||
return { status: "failed", failure: { test: testName, error } };
|
||||
}
|
||||
}
|
||||
|
||||
async function runRemainingTests(index: number): Promise<void> {
|
||||
if (index >= testFiles.length) return;
|
||||
const testFile = testFiles[index];
|
||||
const outcome = await runSingleTest(testFile);
|
||||
if (outcome.status === "passed") {
|
||||
passed++;
|
||||
return runRemainingTests(index + 1);
|
||||
}
|
||||
failed++;
|
||||
failures.push(outcome.failure);
|
||||
}
|
||||
const workerCount = Math.min(concurrency, testFiles.length);
|
||||
await Promise.all(Array.from({ length: workerCount }, () => worker()));
|
||||
|
||||
await runRemainingTests(0);
|
||||
const totalDurationMs = Date.now() - totalStart;
|
||||
|
||||
// Summary
|
||||
console.log("\n" + "=".repeat(50));
|
||||
@@ -172,6 +296,15 @@ console.log("=".repeat(50));
|
||||
console.log(` ✅ Passed: ${passed}`);
|
||||
console.log(` ❌ Failed: ${failed}`);
|
||||
console.log(` 📝 Total: ${passed + failed}`);
|
||||
console.log(` ⏱ Wall: ${formatDuration(totalDurationMs)} (concurrency=${concurrency})`);
|
||||
|
||||
const slowest = [...timings].sort((a, b) => b.durationMs - a.durationMs).slice(0, 5);
|
||||
if (slowest.length > 0) {
|
||||
console.log("\n🐢 Slowest tests:");
|
||||
for (const t of slowest) {
|
||||
console.log(` - ${t.test} (${formatDuration(t.durationMs)})`);
|
||||
}
|
||||
}
|
||||
|
||||
if (failures.length > 0) {
|
||||
console.log("\n❌ Failed tests:");
|
||||
|
||||
Reference in New Issue
Block a user