From 3fbe8c8ca650927fad7584e18b42c9276b7464d2 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Oct 2026 01:33:19 +0000 Subject: [PATCH 1/4] Print per-target test timings from CI shards ci-test-shard.py now echoes cargo's merged output and records each test binary's `finished in` time, then prints one `ci-test-shard: timings` JSON line at the end of the shard, with the build time up to the first test binary. The log tail is all the log API returns, so this is how a Linux coverage runtime sample can be read back: the only sample in the repo is from Windows, which models Linux coverage badly (#1768). Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011f9aeQpFXjqvowuU1RFbdQ --- scripts/ci-test-shard.py | 44 ++++++++++++++++++++- scripts/tests/test_ci_test_shard.py | 60 +++++++++++++++++++++++------ 2 files changed, 92 insertions(+), 12 deletions(-) diff --git a/scripts/ci-test-shard.py b/scripts/ci-test-shard.py index 079861c6b..04a4611df 100644 --- a/scripts/ci-test-shard.py +++ b/scripts/ci-test-shard.py @@ -19,6 +19,12 @@ no doctests, which cargo-llvm-cov leaves out without `--doctests`. A later `cargo llvm-cov report` exports the shard's profiles. +Each test binary's `finished in` time is collected from cargo's output and +printed at the end as one `ci-test-shard: timings {...}` JSON line, along +with how long cargo took to reach its first test binary. The log tail is +what log viewers and the log API return, so a fresh runtime sample can be +read from any shard's log. + Extra arguments after `SHARD COUNT` go to every `cargo` invocation. Each invocation runs with `--no-fail-fast`; the exit status is non-zero if any of them failed. @@ -30,13 +36,22 @@ import json import os from pathlib import Path +import re import subprocess import sys +import time # Shard 1's share of the integration targets, relative to every other # shard's 1.0 (it also carries the unit tests and doctests). FIRST_SHARD_WEIGHT = 0.5 +# cargo's `Running tests/x.rs (target/debug/deps/x-0123456789abcdef)` and +# `test result: ... finished in 1.23s` lines. The binary name, not the source +# path, is the target name (tests/apply/main.rs is `apply`); unit-test +# binaries are recorded as `unit:`. +RUNNING = re.compile(r"^\s*Running (unittests )?\S+ \(.*?([^/\\]+)-[0-9a-f]{16}(?:\.exe)?\)\s*$") +RESULT = re.compile(r"^test result: .*finished in ([0-9.]+)s\s*$") + def integration_targets(metadata): """Sorted, de-duplicated names of the workspace's integration tests.""" @@ -90,6 +105,24 @@ def invocations(shard, count, names, extra=(), timings=None, coverage=False): return runs +def run(args, timings): + """Run `args`, echoing its merged output, and add each test binary's + `finished in` seconds to `timings`. Returns the exit status.""" + current = None + with subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, errors="replace", bufsize=1) as proc: + for line in proc.stdout: + sys.stdout.write(line) + sys.stdout.flush() + if m := RUNNING.match(line): + current = ("unit:" if m[1] else "") + m[2] + timings.setdefault("_first_binary", time.monotonic()) + elif current and (m := RESULT.match(line)): + timings[current] = timings.get(current, 0.0) + float(m[1]) + current = None + return proc.returncode + + def main(argv): if len(argv) < 2: print(__doc__, file=sys.stderr) @@ -105,10 +138,19 @@ def main(argv): print(f"ci-test-shard: shard {shard}/{count}: " f"{len(partition(names, count, timings)[shard - 1])} of {len(names)} integration targets", flush=True) status = 0 + measured = {} for args in invocations(shard, count, names, argv[2:], timings, coverage): print("+ " + " ".join(args), flush=True) - if subprocess.run(args).returncode != 0: + start = time.monotonic() + measured.pop("_first_binary", None) + if run(args, measured) != 0: status = 1 + # Build time of this invocation: start to its first test binary. + first = measured.pop("_first_binary", None) + if first is not None: + measured["_build_seconds"] = round(measured.get("_build_seconds", 0.0) + first - start, 2) + order = sorted(measured.items(), key=lambda kv: (not kv[0].startswith("_"), -kv[1], kv[0])) + print("ci-test-shard: timings " + json.dumps({k: round(v, 2) for k, v in order}), flush=True) return status diff --git a/scripts/tests/test_ci_test_shard.py b/scripts/tests/test_ci_test_shard.py index 6ffd95ffa..17762d8be 100644 --- a/scripts/tests/test_ci_test_shard.py +++ b/scripts/tests/test_ci_test_shard.py @@ -1,11 +1,13 @@ """ci-test-shard.py: the debug/release shards preserve the workspace tests.""" import importlib.util +import io import json import os import re import shutil import subprocess +import sys import tempfile import unittest from pathlib import Path @@ -62,13 +64,11 @@ def test_release_profile_is_kept_for_integration_unit_and_doc_tests(self): def test_failed_unit_or_integration_run_still_runs_docs_and_fails_the_shard(self): metadata = {"workspace_members": ["a"], "packages": [ {"id": "a", "targets": [{"name": "integration", "kind": ["test"]}]}]} - with patch.object(shard.subprocess, "run", side_effect=[ - subprocess.CompletedProcess([], 0, stdout=json.dumps(metadata)), - subprocess.CompletedProcess([], 1), - subprocess.CompletedProcess([], 0), - ]) as run: + with patch.object(shard.subprocess, "run", return_value=subprocess.CompletedProcess( + [], 0, stdout=json.dumps(metadata))), \ + patch.object(shard, "run", side_effect=[1, 0]) as run: self.assertEqual(shard.main(["1", "3", "--locked", "--profile", "ci-release"]), 1) - self.assertEqual(run.call_count, 3) + self.assertEqual(run.call_count, 2) self.assertIn("--doc", run.call_args_list[-1].args[0]) def test_coverage_shards_run_llvm_cov_without_doctests(self): @@ -88,14 +88,52 @@ def test_coverage_mode_comes_from_the_environment(self): {"id": "a", "targets": [{"name": "integration", "kind": ["test"]}]}]} env = {k: v for k, v in shard.os.environ.items() if k != "CI_TEST_TIMINGS"} with patch.dict(shard.os.environ, {**env, "CI_TEST_COVERAGE": "1"}, clear=True), \ - patch.object(shard.subprocess, "run", side_effect=[ - subprocess.CompletedProcess([], 0, stdout=json.dumps(metadata)), - subprocess.CompletedProcess([], 0), - ]) as run: + patch.object(shard.subprocess, "run", return_value=subprocess.CompletedProcess( + [], 0, stdout=json.dumps(metadata))), \ + patch.object(shard, "run", return_value=0) as run: self.assertEqual(shard.main(["1", "2"]), 0) - self.assertEqual(run.call_count, 2) + self.assertEqual(run.call_count, 1) self.assertEqual(run.call_args_list[-1].args[0][:3], ["cargo", "llvm-cov", "--no-report"]) + def test_run_echoes_output_and_records_each_binary_time(self): + script = ( + "import sys\n" + "print(' Running unittests src/lib.rs (target/debug/deps/socket_patch_core-0123456789abcdef)'," + " file=sys.stderr, flush=True)\n" + "print('test result: ok. 3 passed; 0 failed; finished in 30.12s', flush=True)\n" + "print(r' Running tests\\apply\\main.rs (target\\debug\\deps\\apply-fedcba9876543210.exe)'," + " file=sys.stderr, flush=True)\n" + "print('test result: FAILED. 1 passed; 1 failed; finished in 1.50s', flush=True)\n" + "print(' Doc-tests socket_patch_core', file=sys.stderr, flush=True)\n" + "print('test result: ok. 9 passed; 0 failed; finished in 4.00s', flush=True)\n" + "sys.exit(3)\n") + timings = {} + with patch.object(shard.sys, "stdout", new_callable=io.StringIO) as out: + self.assertEqual(shard.run([sys.executable, "-c", script], timings), 3) + timings.pop("_first_binary") + self.assertEqual(timings, {"unit:socket_patch_core": 30.12, "apply": 1.5}) + self.assertIn("Doc-tests socket_patch_core", out.getvalue()) + self.assertIn("finished in 4.00s", out.getvalue()) + + def test_main_prints_the_timing_sample_last(self): + metadata = {"workspace_members": ["a"], "packages": [ + {"id": "a", "targets": [{"name": "integration", "kind": ["test"]}]}]} + + def fake_run(args, timings): + timings["_first_binary"] = shard.time.monotonic() + timings["integration"] = 2.0 + return 0 + with patch.object(shard.subprocess, "run", return_value=subprocess.CompletedProcess( + [], 0, stdout=json.dumps(metadata))), \ + patch.object(shard, "run", side_effect=fake_run), \ + patch.object(shard.sys, "stdout", new_callable=io.StringIO) as out: + self.assertEqual(shard.main(["2", "2"]), 0) + last = out.getvalue().splitlines()[-1] + self.assertTrue(last.startswith("ci-test-shard: timings "), last) + sample = json.loads(last.removeprefix("ci-test-shard: timings ")) + self.assertEqual(list(sample), ["_build_seconds", "integration"]) + self.assertEqual(sample["integration"], 2.0) + def test_negative_bad_shard(self): with self.assertRaises(ValueError): shard.invocations(3, 2, self.NAMES) From d7bfacc4f9e361fc72cecd4928069d3aa9e1f53c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Oct 2026 01:42:43 +0000 Subject: [PATCH 2/4] Balance coverage shards on a Linux sample coverage (2) sets the merge-queue critical path: with the count split its test step ran 248 s against 173 s for shard 1 on this PR's first run, and 307-335 s against 176-179 s in recent merge_group runs. #1768 balanced on the Windows sample and saved nothing because Windows times don't model instrumented Linux targets (e2e_vendor_pypi_build: 35-72 s on Linux, 0 s in the Windows sample). ci-coverage-durations.json is this PR's own first coverage run, read back from the new timings line. unit_seconds adds ~34 s to the unit runtime for building the unit-test harnesses, and compile_seconds is 0.3 s, both fitted from the two shards' build and report times. Coverage uses it unless CI_TEST_TIMINGS is set. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011f9aeQpFXjqvowuU1RFbdQ --- scripts/ci-coverage-durations.json | 246 ++++++++++++++++++++++++++++ scripts/ci-test-shard.py | 12 +- scripts/tests/test_ci_test_shard.py | 39 +++++ 3 files changed, 294 insertions(+), 3 deletions(-) create mode 100644 scripts/ci-coverage-durations.json diff --git a/scripts/ci-coverage-durations.json b/scripts/ci-coverage-durations.json new file mode 100644 index 000000000..cb2ab11a8 --- /dev/null +++ b/scripts/ci-coverage-durations.json @@ -0,0 +1,246 @@ +{ + "source_run": 40801155956, + "source_jobs": [ + 123209371302, + 123209371281 + ], + "unit_seconds": 63.68, + "compile_seconds": 0.3, + "default_seconds": 1.0, + "targets": { + "api_retry_e2e": 0.12, + "api_timeout_e2e": 4.04, + "apply": 2.5, + "binary_fetch_error_classification_e2e": 0.0, + "blob_fetcher_edges_e2e": 0.06, + "child_deadline_guard": 0.08, + "cli": 4.7, + "cli_apply_silent": 0.01, + "cli_argv_non_utf8": 0.0, + "cli_config_fallback": 0.09, + "cli_get_silent": 0.01, + "cli_global_args": 0.12, + "cli_parse_apply": 0.0, + "cli_parse_get": 0.02, + "cli_parse_list": 0.04, + "cli_parse_main": 0.03, + "cli_parse_remove": 0.01, + "cli_parse_repair": 0.01, + "cli_parse_rollback": 0.0, + "cli_parse_scan": 0.05, + "cli_parse_vendor": 0.02, + "cli_parse_vex": 0.01, + "cli_path_flags_validated": 0.13, + "cli_remove_silent": 0.03, + "cli_scan_silent": 0.05, + "cli_sigpipe": 0.01, + "command_module_layering": 0.03, + "contract_gradle_codes": 0.36, + "contract_paid_required": 0.0, + "crawl_fd_limit_e2e": 0.15, + "crawler_cargo_e2e": 0.01, + "crawler_composer_e2e": 0.27, + "crawler_deno_e2e": 0.0, + "crawler_go_e2e": 0.01, + "crawler_gradle_e2e": 0.18, + "crawler_maven_e2e": 0.01, + "crawler_monorepo_gaps": 0.01, + "crawler_npm_e2e": 7.78, + "crawler_nuget_e2e": 0.04, + "crawler_python_e2e": 0.34, + "crawler_ruby_e2e": 0.28, + "crawlers_empty_paths_e2e": 0.02, + "docker_e2e_cargo": 0.0, + "docker_e2e_composer": 0.0, + "docker_e2e_deno": 0.0, + "docker_e2e_gem": 0.0, + "docker_e2e_golang": 0.0, + "docker_e2e_maven": 0.0, + "docker_e2e_npm": 0.0, + "docker_e2e_nuget": 0.0, + "docker_e2e_pypi": 0.0, + "docker_e2e_sbt": 0.0, + "docker_e2e_vendor_composer": 0.0, + "docker_e2e_vendor_gem": 0.0, + "docker_e2e_vendor_maven": 0.0, + "docker_e2e_vendor_nuget": 0.0, + "docker_e2e_vendor_pypi_pm": 0.0, + "e2e_bun_lockb": 0.01, + "e2e_cargo": 0.09, + "e2e_composer": 0.03, + "e2e_composer_vendor_dir_hints": 0.02, + "e2e_composer_version_identity": 0.04, + "e2e_embedded_vex": 0.04, + "e2e_gem": 0.02, + "e2e_golang": 0.02, + "e2e_golang_build": 1.7, + "e2e_golang_hosted_build": 3.26, + "e2e_golang_hosted_state": 0.09, + "e2e_golang_workspace_build": 3.12, + "e2e_gradle_agent_build": 0.01, + "e2e_gradle_discovery_build": 0.07, + "e2e_hosted_production": 0.0, + "e2e_maven": 0.02, + "e2e_npm": 0.0, + "e2e_nuget": 0.02, + "e2e_nuget_dotnet_build": 0.0, + "e2e_pypi": 0.0, + "e2e_pypi_bytecode": 0.15, + "e2e_pypi_multi_copy": 0.05, + "e2e_redirect_bun_build": 0.01, + "e2e_redirect_cargo_build": 0.66, + "e2e_redirect_cargo_shapes": 2.85, + "e2e_redirect_composer_build": 0.0, + "e2e_redirect_gem_build": 0.0, + "e2e_redirect_gem_stale_install": 0.71, + "e2e_redirect_gradle_build": 0.02, + "e2e_redirect_maven_build": 0.0, + "e2e_redirect_npm_build": 0.0, + "e2e_redirect_pnpm_build": 0.03, + "e2e_redirect_rush_sim": 0.0, + "e2e_redirect_uv_build": 0.0, + "e2e_redirect_vlt_build": 0.0, + "e2e_redirect_yarn_berry_build": 14.37, + "e2e_redirect_yarn_classic_build": 6.44, + "e2e_safety_cargo_build": 0.02, + "e2e_safety_cow": 0.08, + "e2e_safety_internals": 0.0, + "e2e_safety_lock": 1.01, + "e2e_safety_pnpm": 0.0, + "e2e_safety_vlt": 0.01, + "e2e_safety_yarn_pnp": 0.08, + "e2e_sbt": 0.16, + "e2e_sbt_build": 0.0, + "e2e_sbt_hosted": 0.1, + "e2e_sbt_vendor": 0.48, + "e2e_sbt_vendor_build": 0.0, + "e2e_scala_cli_vendor": 0.17, + "e2e_scan": 0.0, + "e2e_socket_yml_policy": 0.87, + "e2e_vendor_bun_build": 0.02, + "e2e_vendor_cargo_build": 0.74, + "e2e_vendor_composer_build": 0.01, + "e2e_vendor_composer_crlf": 0.05, + "e2e_vendor_gem_build": 0.0, + "e2e_vendor_golang_build": 1.03, + "e2e_vendor_gradle_build": 0.02, + "e2e_vendor_jvm_build": 0.03, + "e2e_vendor_maven_build": 0.0, + "e2e_vendor_npm_build": 1.24, + "e2e_vendor_pnpm_build": 12.61, + "e2e_vendor_pypi_build": 34.89, + "e2e_vendor_vlt_build": 0.0, + "e2e_vendor_yarn_berry_build": 2.22, + "e2e_vendor_yarn_classic_build": 0.95, + "e2e_vendor_yarn_classic_dev_flow": 1.81, + "e2e_vendored_production": 0.0, + "e2e_vex": 0.15, + "e2e_vex_build": 0.0, + "e2e_vex_lockfile": 11.21, + "e2e_vex_redirect": 0.41, + "e2e_vex_vendor": 0.14, + "e2e_vlt": 0.56, + "e2e_yarn4_pnpm_linker_build": 2.57, + "e2e_yarn4_workspaces_build": 2.08, + "e2e_yarn_legacy_cachekey_refusal_build": 0.71, + "ecosystem_dispatch_e2e": 0.21, + "fuzzy_match_e2e": 0.0, + "get": 0.68, + "global_probe_spawn_e2e": 0.02, + "global_scope_project_state": 0.24, + "gradle_agent_cli": 0.29, + "help_text_hygiene": 0.11, + "hosted_inventory": 0.06, + "hosted_memory_engine": 0.78, + "hosted_memory_parity": 0.26, + "hosted_memory_rollout": 0.22, + "hosted_superseding_pypi": 0.07, + "in_process_agent_reapply": 0.14, + "in_process_alternate_installers": 4.58, + "in_process_cargo_apply": 0.82, + "in_process_edge_cases": 0.17, + "in_process_gem_apply": 0.55, + "in_process_gem_multi_platform": 0.26, + "in_process_get": 0.11, + "in_process_get_corrupt_manifest": 0.0, + "in_process_get_hosted_ecosystems": 0.29, + "in_process_get_manifest_path": 0.06, + "in_process_get_modes": 0.13, + "in_process_get_update_count": 0.01, + "in_process_get_uuid_fallback": 0.01, + "in_process_pypi_apply": 8.43, + "in_process_pypi_multi_release": 9.6, + "in_process_python_envs": 0.22, + "in_process_redirect": 4.06, + "in_process_redirect_pdm": 0.47, + "in_process_redirect_pipenv": 0.67, + "in_process_redirect_pnpm": 1.81, + "in_process_redirect_poetry": 0.27, + "in_process_remote_ecosystems_apply": 1.04, + "in_process_remove_repair_lifecycle": 0.56, + "in_process_rollback_all_ecosystems": 0.57, + "in_process_rollback_hosted": 1.96, + "in_process_rollback_vendored": 0.19, + "in_process_scan": 0.19, + "in_process_scan_withdrawn": 0.35, + "in_process_target_ambiguity": 0.51, + "in_process_vendor": 5.58, + "in_process_vendor_bun_takeover": 0.42, + "in_process_vendor_npm_v1_takeover": 0.12, + "in_process_vendor_pnpm_parent_child": 0.17, + "in_process_vendor_pnpm_takeover": 0.19, + "in_process_vendor_pypi_takeover": 0.1, + "json_error_shape": 0.03, + "manifest_load_error_matrix": 0.16, + "manifest_path_ledger_root": 0.09, + "maven_sidecar_cli": 0.05, + "mode_migration_bun": 0.0, + "mode_migration_cargo": 0.78, + "mode_migration_composer": 0.31, + "mode_migration_npm": 3.45, + "mode_migration_pypi": 1.06, + "mode_migration_vlt": 0.01, + "output_helpers_e2e": 0.0, + "package_e2e": 0.01, + "pnpm_hosted": 0.02, + "poetry_hosted": 0.02, + "policy_pypi_names": 0.07, + "proxy_batch_e2e": 0.42, + "redirect_golden": 0.11, + "redirect_sbt_golden": 0.09, + "remedy_commands_parse": 0.17, + "remove": 0.78, + "remove_rollback_api_overrides": 0.01, + "repair": 1.54, + "rollback": 0.72, + "rollback_new_file_e2e": 0.01, + "scan": 4.26, + "scan_api_retry_e2e": 2.52, + "scan_get_mode_from_project_state": 0.17, + "scan_pnpm_relocated_store_cwd_e2e": 0.02, + "scan_requirements_lock_only": 0.66, + "scan_rollout_e2e": 0.56, + "scan_vendor_e2e": 0.49, + "scan_vendor_requirements_unwired": 0.07, + "self_update_e2e": 4.02, + "self_update_failures_e2e": 4.51, + "spawn_env_hygiene": 0.07, + "telemetry_helpers_e2e": 0.0, + "update": 4.75, + "update_state_env": 0.0, + "upstream_restore_golden": 1.45, + "user_patches": 0.15, + "uv_hosted": 0.0, + "vendor": 1.93, + "vendor_crash_safety_e2e": 0.09, + "vendor_eject": 0.14, + "vendor_eject_bun_lockb": 0.32, + "vendor_eject_fresh_checkout": 0.16, + "vendor_group_commit_e2e": 1.83, + "vendor_jvm_cli": 0.86, + "vendor_ledger_schema_e2e": 0.73, + "vendor_partial_staging_e2e": 0.09, + "vex": 0.06, + "vlt_locks": 3.16 + } +} diff --git a/scripts/ci-test-shard.py b/scripts/ci-test-shard.py index 04a4611df..c72e1e76a 100644 --- a/scripts/ci-test-shard.py +++ b/scripts/ci-test-shard.py @@ -12,7 +12,9 @@ Windows debug CI sets CI_TEST_TIMINGS to the checked-in runtime sample. Those shards balance measured test time plus linking work, with shard 1's -unit tests reserved up front. Other runners retain the count-based split. +unit tests reserved up front. Coverage shards do the same on their own +Linux sample (scripts/ci-coverage-durations.json) unless CI_TEST_TIMINGS +is set; an empty CI_TEST_TIMINGS, as on macOS, keeps the count split. CI_TEST_COVERAGE=1 runs the shard under cargo-llvm-cov for ci.yml's `coverage` job: `cargo llvm-cov --no-report` in place of `cargo test`, and @@ -41,6 +43,10 @@ import sys import time +# The Linux runtime sample coverage shards balance on (the Windows one, +# scripts/ci-test-durations.json, doesn't model them: #1768). +COVERAGE_TIMINGS = Path(__file__).with_name("ci-coverage-durations.json") + # Shard 1's share of the integration targets, relative to every other # shard's 1.0 (it also carries the unit tests and doctests). FIRST_SHARD_WEIGHT = 0.5 @@ -132,9 +138,9 @@ def main(argv): ["cargo", "metadata", "--no-deps", "--format-version", "1"], check=True, capture_output=True, text=True).stdout) names = integration_targets(metadata) - timing_path = os.environ.get("CI_TEST_TIMINGS") - timings = json.loads(Path(timing_path).read_text(encoding="utf-8")) if timing_path else None coverage = os.environ.get("CI_TEST_COVERAGE") == "1" + timing_path = os.environ.get("CI_TEST_TIMINGS", COVERAGE_TIMINGS if coverage else None) + timings = json.loads(Path(timing_path).read_text(encoding="utf-8")) if timing_path else None print(f"ci-test-shard: shard {shard}/{count}: " f"{len(partition(names, count, timings)[shard - 1])} of {len(names)} integration targets", flush=True) status = 0 diff --git a/scripts/tests/test_ci_test_shard.py b/scripts/tests/test_ci_test_shard.py index 17762d8be..e828c3206 100644 --- a/scripts/tests/test_ci_test_shard.py +++ b/scripts/tests/test_ci_test_shard.py @@ -134,6 +134,45 @@ def fake_run(args, timings): self.assertEqual(list(sample), ["_build_seconds", "integration"]) self.assertEqual(sample["integration"], 2.0) + def test_coverage_shards_default_to_the_linux_sample(self): + # With the sample, the longest target gets shard 1 to itself; the + # count split gives shard 1 only `b`, by its smaller weight. + metadata = {"workspace_members": ["x"], "packages": [ + {"id": "x", "targets": [{"name": n, "kind": ["test"]} for n in ("a", "b", "c")]}]} + sample = {"unit_seconds": 0.0, "compile_seconds": 0.0, "default_seconds": 1.0, + "targets": {"a": 1.0, "b": 2.0, "c": 100.0}} + env = {k: v for k, v in shard.os.environ.items() if k not in ("CI_TEST_TIMINGS", "CI_TEST_COVERAGE")} + cases = [({"CI_TEST_COVERAGE": "1"}, ["c"]), + ({"CI_TEST_COVERAGE": "1", "CI_TEST_TIMINGS": ""}, ["b"]), + ({}, ["b"])] + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "sample.json" + path.write_text(json.dumps(sample), encoding="utf-8") + for extra, expected in cases: + with self.subTest(env=extra), \ + patch.dict(shard.os.environ, {**env, **extra}, clear=True), \ + patch.object(shard, "COVERAGE_TIMINGS", path), \ + patch.object(shard.subprocess, "run", return_value=subprocess.CompletedProcess( + [], 0, stdout=json.dumps(metadata))), \ + patch.object(shard, "run", return_value=0) as run: + shard.main(["1", "2"]) + self.assertEqual(selected([run.call_args_list[0].args[0]]), expected) + + def test_the_linux_coverage_sample_covers_the_checkout(self): + sample = json.loads(shard.COVERAGE_TIMINGS.read_text(encoding="utf-8")) + for key in ("unit_seconds", "compile_seconds", "default_seconds", "targets"): + self.assertIn(key, sample) + try: + out = subprocess.run(["cargo", "metadata", "--no-deps", "--format-version", "1"], + cwd=ROOT, check=True, capture_output=True, text=True).stdout + except (OSError, subprocess.CalledProcessError): + self.skipTest("cargo not available") + names = shard.integration_targets(json.loads(out)) + # New targets fall back to default_seconds; a mostly-stale sample + # would quietly degrade into an unbalanced split. + missing = [n for n in names if n not in sample["targets"]] + self.assertLess(len(missing), len(names) // 10, missing) + def test_negative_bad_shard(self): with self.assertRaises(ValueError): shard.invocations(3, 2, self.NAMES) From f453afd241cbbc6481cab2ce2ed221ab03fdc76a Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Oct 2026 01:48:34 +0000 Subject: [PATCH 3/4] Harden the shard runner and retune the sample Review fixes for the timings capture: forward cargo's output as raw bytes (Windows decoded UTF-8 as cp1252 and could raise on write), stop reading 10 s after cargo exits so a daemon a test left holding the pipe can't hang the shard, strip ANSI colour before matching, and return the first-binary time from run() instead of a sentinel key. The first balanced run overshot: shard 1's test step went 173 -> 229 s and shard 2's 248 -> 202 s. unit_seconds goes up by that 27 s gap. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011f9aeQpFXjqvowuU1RFbdQ --- scripts/ci-coverage-durations.json | 2 +- scripts/ci-test-shard.py | 56 +++++++++++++++++++---------- scripts/tests/test_ci_test_shard.py | 46 ++++++++++++++++-------- 3 files changed, 71 insertions(+), 33 deletions(-) diff --git a/scripts/ci-coverage-durations.json b/scripts/ci-coverage-durations.json index cb2ab11a8..6d64a8836 100644 --- a/scripts/ci-coverage-durations.json +++ b/scripts/ci-coverage-durations.json @@ -4,7 +4,7 @@ 123209371302, 123209371281 ], - "unit_seconds": 63.68, + "unit_seconds": 90.7, "compile_seconds": 0.3, "default_seconds": 1.0, "targets": { diff --git a/scripts/ci-test-shard.py b/scripts/ci-test-shard.py index c72e1e76a..e75b51e97 100644 --- a/scripts/ci-test-shard.py +++ b/scripts/ci-test-shard.py @@ -14,7 +14,7 @@ Those shards balance measured test time plus linking work, with shard 1's unit tests reserved up front. Coverage shards do the same on their own Linux sample (scripts/ci-coverage-durations.json) unless CI_TEST_TIMINGS -is set; an empty CI_TEST_TIMINGS, as on macOS, keeps the count split. +is set; setting it empty on the coverage step keeps the count split. CI_TEST_COVERAGE=1 runs the shard under cargo-llvm-cov for ci.yml's `coverage` job: `cargo llvm-cov --no-report` in place of `cargo test`, and @@ -41,6 +41,7 @@ import re import subprocess import sys +import threading import time # The Linux runtime sample coverage shards balance on (the Windows one, @@ -54,10 +55,14 @@ # cargo's `Running tests/x.rs (target/debug/deps/x-0123456789abcdef)` and # `test result: ... finished in 1.23s` lines. The binary name, not the source # path, is the target name (tests/apply/main.rs is `apply`); unit-test -# binaries are recorded as `unit:`. +# binaries are recorded as `unit:`. ANSI colour is stripped first. +ANSI = re.compile(r"\x1b\[[0-9;]*m") RUNNING = re.compile(r"^\s*Running (unittests )?\S+ \(.*?([^/\\]+)-[0-9a-f]{16}(?:\.exe)?\)\s*$") RESULT = re.compile(r"^test result: .*finished in ([0-9.]+)s\s*$") +# How long run() keeps reading cargo's pipe after cargo exits. +DRAIN_SECONDS = 10 + def integration_targets(metadata): """Sorted, de-duplicated names of the workspace's integration tests.""" @@ -112,21 +117,35 @@ def invocations(shard, count, names, extra=(), timings=None, coverage=False): def run(args, timings): - """Run `args`, echoing its merged output, and add each test binary's - `finished in` seconds to `timings`. Returns the exit status.""" - current = None - with subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, errors="replace", bufsize=1) as proc: - for line in proc.stdout: - sys.stdout.write(line) - sys.stdout.flush() + """Run `args`, echoing its merged output byte for byte, and add each test + binary's `finished in` seconds to `timings`. Returns the exit status and + when the first test binary started (None if none did).""" + first = None + + def forward(pipe): + nonlocal first + current = None + for raw in pipe: + sys.stdout.buffer.write(raw) + sys.stdout.buffer.flush() + line = ANSI.sub("", raw.decode("utf-8", "replace")) if m := RUNNING.match(line): current = ("unit:" if m[1] else "") + m[2] - timings.setdefault("_first_binary", time.monotonic()) + first = first or time.monotonic() elif current and (m := RESULT.match(line)): timings[current] = timings.get(current, 0.0) + float(m[1]) current = None - return proc.returncode + + proc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + reader = threading.Thread(target=forward, args=(proc.stdout,), daemon=True) + reader.start() + status = proc.wait() + # A process a test left behind (a build daemon, say) can hold the pipe + # open after cargo exits; don't wait for it past a short drain. + reader.join(timeout=DRAIN_SECONDS) + if not reader.is_alive(): + proc.stdout.close() + return status, first def main(argv): @@ -145,18 +164,19 @@ def main(argv): f"{len(partition(names, count, timings)[shard - 1])} of {len(names)} integration targets", flush=True) status = 0 measured = {} + build = 0.0 for args in invocations(shard, count, names, argv[2:], timings, coverage): print("+ " + " ".join(args), flush=True) start = time.monotonic() - measured.pop("_first_binary", None) - if run(args, measured) != 0: + code, first = run(args, measured) + if code != 0: status = 1 # Build time of this invocation: start to its first test binary. - first = measured.pop("_first_binary", None) + # (The doctest run prints no `Running` line and isn't counted.) if first is not None: - measured["_build_seconds"] = round(measured.get("_build_seconds", 0.0) + first - start, 2) - order = sorted(measured.items(), key=lambda kv: (not kv[0].startswith("_"), -kv[1], kv[0])) - print("ci-test-shard: timings " + json.dumps({k: round(v, 2) for k, v in order}), flush=True) + build += first - start + measured = {"_build_seconds": build, **dict(sorted(measured.items(), key=lambda kv: (-kv[1], kv[0])))} + print("ci-test-shard: timings " + json.dumps({k: round(v, 2) for k, v in measured.items()}), flush=True) return status diff --git a/scripts/tests/test_ci_test_shard.py b/scripts/tests/test_ci_test_shard.py index e828c3206..6f6a1bdd7 100644 --- a/scripts/tests/test_ci_test_shard.py +++ b/scripts/tests/test_ci_test_shard.py @@ -9,6 +9,7 @@ import subprocess import sys import tempfile +import time import unittest from pathlib import Path from unittest.mock import patch @@ -19,6 +20,11 @@ spec.loader.exec_module(shard) +def quiet(): + """Capture the script's stdout (text and raw bytes) instead of printing it.""" + return patch.object(shard.sys, "stdout", io.TextIOWrapper(io.BytesIO(), encoding="utf-8", write_through=True)) + + def selected(runs): names = [] for args in runs: @@ -66,7 +72,7 @@ def test_failed_unit_or_integration_run_still_runs_docs_and_fails_the_shard(self {"id": "a", "targets": [{"name": "integration", "kind": ["test"]}]}]} with patch.object(shard.subprocess, "run", return_value=subprocess.CompletedProcess( [], 0, stdout=json.dumps(metadata))), \ - patch.object(shard, "run", side_effect=[1, 0]) as run: + patch.object(shard, "run", side_effect=[(1, None), (0, None)]) as run, quiet(): self.assertEqual(shard.main(["1", "3", "--locked", "--profile", "ci-release"]), 1) self.assertEqual(run.call_count, 2) self.assertIn("--doc", run.call_args_list[-1].args[0]) @@ -90,7 +96,7 @@ def test_coverage_mode_comes_from_the_environment(self): with patch.dict(shard.os.environ, {**env, "CI_TEST_COVERAGE": "1"}, clear=True), \ patch.object(shard.subprocess, "run", return_value=subprocess.CompletedProcess( [], 0, stdout=json.dumps(metadata))), \ - patch.object(shard, "run", return_value=0) as run: + patch.object(shard, "run", return_value=(0, None)) as run, quiet(): self.assertEqual(shard.main(["1", "2"]), 0) self.assertEqual(run.call_count, 1) self.assertEqual(run.call_args_list[-1].args[0][:3], ["cargo", "llvm-cov", "--no-report"]) @@ -98,7 +104,7 @@ def test_coverage_mode_comes_from_the_environment(self): def test_run_echoes_output_and_records_each_binary_time(self): script = ( "import sys\n" - "print(' Running unittests src/lib.rs (target/debug/deps/socket_patch_core-0123456789abcdef)'," + "print('\\x1b[1m\\x1b[92m Running\\x1b[0m unittests src/lib.rs (target/debug/deps/socket_patch_core-0123456789abcdef)'," " file=sys.stderr, flush=True)\n" "print('test result: ok. 3 passed; 0 failed; finished in 30.12s', flush=True)\n" "print(r' Running tests\\apply\\main.rs (target\\debug\\deps\\apply-fedcba9876543210.exe)'," @@ -106,29 +112,41 @@ def test_run_echoes_output_and_records_each_binary_time(self): "print('test result: FAILED. 1 passed; 1 failed; finished in 1.50s', flush=True)\n" "print(' Doc-tests socket_patch_core', file=sys.stderr, flush=True)\n" "print('test result: ok. 9 passed; 0 failed; finished in 4.00s', flush=True)\n" + "sys.stdout.buffer.write('\\u2550\\u274c\\n'.encode('utf-8'))\n" "sys.exit(3)\n") timings = {} - with patch.object(shard.sys, "stdout", new_callable=io.StringIO) as out: - self.assertEqual(shard.run([sys.executable, "-c", script], timings), 3) - timings.pop("_first_binary") + with quiet() as out: + status, first = shard.run([sys.executable, "-c", script], timings) + self.assertEqual(status, 3) + self.assertIsNotNone(first) self.assertEqual(timings, {"unit:socket_patch_core": 30.12, "apply": 1.5}) - self.assertIn("Doc-tests socket_patch_core", out.getvalue()) - self.assertIn("finished in 4.00s", out.getvalue()) + text = out.buffer.getvalue().decode("utf-8") + self.assertIn("Doc-tests socket_patch_core", text) + self.assertIn("finished in 4.00s", text) + self.assertIn("\u2550\u274c", text) + + def test_run_does_not_wait_for_a_leftover_process_holding_the_pipe(self): + # A test can leave a daemon behind with cargo's stdout inherited; the + # old subprocess.run(args) returned when cargo exited, and so must this. + script = ("import subprocess, sys\n" + "subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'])\n") + start = time.monotonic() + with quiet(), patch.object(shard, "DRAIN_SECONDS", 0.5): + self.assertEqual(shard.run([sys.executable, "-c", script], {})[0], 0) + self.assertLess(time.monotonic() - start, 15) def test_main_prints_the_timing_sample_last(self): metadata = {"workspace_members": ["a"], "packages": [ {"id": "a", "targets": [{"name": "integration", "kind": ["test"]}]}]} def fake_run(args, timings): - timings["_first_binary"] = shard.time.monotonic() timings["integration"] = 2.0 - return 0 + return 0, shard.time.monotonic() with patch.object(shard.subprocess, "run", return_value=subprocess.CompletedProcess( [], 0, stdout=json.dumps(metadata))), \ - patch.object(shard, "run", side_effect=fake_run), \ - patch.object(shard.sys, "stdout", new_callable=io.StringIO) as out: + patch.object(shard, "run", side_effect=fake_run), quiet() as out: self.assertEqual(shard.main(["2", "2"]), 0) - last = out.getvalue().splitlines()[-1] + last = out.buffer.getvalue().decode("utf-8").splitlines()[-1] self.assertTrue(last.startswith("ci-test-shard: timings "), last) sample = json.loads(last.removeprefix("ci-test-shard: timings ")) self.assertEqual(list(sample), ["_build_seconds", "integration"]) @@ -154,7 +172,7 @@ def test_coverage_shards_default_to_the_linux_sample(self): patch.object(shard, "COVERAGE_TIMINGS", path), \ patch.object(shard.subprocess, "run", return_value=subprocess.CompletedProcess( [], 0, stdout=json.dumps(metadata))), \ - patch.object(shard, "run", return_value=0) as run: + patch.object(shard, "run", return_value=(0, None)) as run, quiet(): shard.main(["1", "2"]) self.assertEqual(selected([run.call_args_list[0].args[0]]), expected) From 489db940742c4b9e77269d1361ebb39b223419d9 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Oct 2026 02:07:07 +0000 Subject: [PATCH 4/4] Keep draining cargo's pipe after a write error If echoing to stdout failed (BrokenPipeError when tee dies), the reader thread exited and cargo blocked on a full pipe, so proc.wait() never returned and the shard sat until the job timeout. The reader now stops echoing but keeps reading and recording timings. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_011f9aeQpFXjqvowuU1RFbdQ --- scripts/ci-test-shard.py | 11 +++++++++-- scripts/tests/test_ci_test_shard.py | 20 ++++++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/scripts/ci-test-shard.py b/scripts/ci-test-shard.py index e75b51e97..cc0ddcf02 100644 --- a/scripts/ci-test-shard.py +++ b/scripts/ci-test-shard.py @@ -125,9 +125,16 @@ def run(args, timings): def forward(pipe): nonlocal first current = None + echo = True for raw in pipe: - sys.stdout.buffer.write(raw) - sys.stdout.buffer.flush() + # Keep draining after a failed write (say `tee` died): a reader + # that stops would leave cargo blocked on a full pipe. + if echo: + try: + sys.stdout.buffer.write(raw) + sys.stdout.buffer.flush() + except (OSError, ValueError): + echo = False line = ANSI.sub("", raw.decode("utf-8", "replace")) if m := RUNNING.match(line): current = ("unit:" if m[1] else "") + m[2] diff --git a/scripts/tests/test_ci_test_shard.py b/scripts/tests/test_ci_test_shard.py index 6f6a1bdd7..bbfbbde4c 100644 --- a/scripts/tests/test_ci_test_shard.py +++ b/scripts/tests/test_ci_test_shard.py @@ -135,6 +135,26 @@ def test_run_does_not_wait_for_a_leftover_process_holding_the_pipe(self): self.assertEqual(shard.run([sys.executable, "-c", script], {})[0], 0) self.assertLess(time.monotonic() - start, 15) + def test_run_keeps_draining_when_its_own_stdout_breaks(self): + # 1 MB is far past a pipe buffer: if the reader stopped at the first + # failed write, the child would block and run() would never return. + script = ("import sys\n" + "for _ in range(1024): sys.stdout.write('x' * 1023 + '\\n')\n" + "print(' Running tests/late.rs (target/debug/deps/late-0123456789abcdef)', flush=True)\n" + "print('test result: ok. 1 passed; finished in 2.00s', flush=True)\n") + + class Broken(io.RawIOBase): + def writable(self): + return True + + def write(self, b): + raise BrokenPipeError + + timings = {} + with patch.object(shard.sys, "stdout", io.TextIOWrapper(io.BufferedWriter(Broken()))): + self.assertEqual(shard.run([sys.executable, "-c", script], timings)[0], 0) + self.assertEqual(timings, {"late": 2.0}) + def test_main_prints_the_timing_sample_last(self): metadata = {"workspace_members": ["a"], "packages": [ {"id": "a", "targets": [{"name": "integration", "kind": ["test"]}]}]}