Speed up local Go test runs

.github/scripts/ci-test-sharding.md

@@ -55,6 +55,20 @@ estimated cost, place each on the currently-lightest shard.
   duplicated test). This invariant is unit-tested in `test_ci_shard.py`, which
   the `ci-policy` workflow runs — a broken partition can't merge.
 
+## Local full-suite runner
+
+From the repository root, run:
+
+```bash
+python3 .github/scripts/run_go_tests.py
+```
+
+The local runner uses the same partition controller and a CPU-aware local
+worker count. It enumerates the complete top-level test set, verifies that every
+test belongs to exactly one shard, and returns success only when every shard
+passes. Each shard remains serial inside its process. Unix CI keeps the
+unsharded `go test ./...` run as an independent correctness reference.
+
 ## Canonical location
 
 This repository owns `ci_shard.py` and the sharded runtime workflow. There is no

.github/scripts/run_go_tests.py

@@ -0,0 +1,203 @@
+#!/usr/bin/env python3
+"""Run the complete Boatstack Go suite in isolated, balanced local shards."""
+
+from __future__ import annotations
+
+import argparse
+import io
+import os
+import subprocess
+import sys
+import tempfile
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Sequence
+
+import ci_shard
+
+
+REPO = Path(__file__).resolve().parents[2]
+RUNTIME = REPO / "boatstack"
+MAX_DEFAULT_JOBS = 10
+
+
+class RunnerError(RuntimeError):
+    """A deterministic local-test precondition or worker failed."""
+
+
+@dataclass(frozen=True)
+class ShardResult:
+    index: int
+    test_count: int
+    returncode: int
+    elapsed_seconds: float
+    output: str
+
+
+def default_jobs() -> int:
+    cores = max(1, os.cpu_count() or 1)
+    # Keep headroom for the Git subprocesses created by almost every test.
+    cpu_aware_jobs = max(1, (cores * 5 + 3) // 7)
+    return min(MAX_DEFAULT_JOBS, cpu_aware_jobs)
+
+
+def list_top_level_tests(
+    runtime: Path = RUNTIME,
+    *,
+    run: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
+) -> list[str]:
+    completed = run(
+        ["go", "test", "-list", "^Test", "./..."],
+        cwd=runtime,
+        text=True,
+        capture_output=True,
+    )
+    if completed.returncode != 0:
+        detail = (completed.stdout + completed.stderr).strip()
+        raise RunnerError(f"test enumeration failed\n{detail}")
+    names = ci_shard.read_test_names(io.StringIO(completed.stdout))
+    if not names:
+        raise RunnerError("test enumeration returned no top-level tests")
+    return names
+
+
+def verified_partition(names: list[str], jobs: int) -> list[list[str]]:
+    if jobs < 1:
+        raise RunnerError("--jobs must be at least 1")
+    if not names:
+        raise RunnerError("cannot partition an empty test set")
+    shards = ci_shard.assign_shards(names, min(jobs, len(names)), {})
+    assigned = [name for shard in shards for name in shard]
+    if sorted(assigned) != sorted(names) or len(assigned) != len(set(assigned)):
+        raise RunnerError("shards do not partition the enumerated tests exactly")
+    return shards
+
+
+def stop_processes(processes: Sequence[subprocess.Popen], timeout: float = 2.0) -> None:
+    active = [process for process in processes if process.poll() is None]
+    for process in active:
+        process.terminate()
+    deadline = time.monotonic() + timeout
+    while active and time.monotonic() < deadline:
+        active = [process for process in active if process.poll() is None]
+        if active:
+            time.sleep(0.02)
+    for process in active:
+        process.kill()
+    for process in processes:
+        try:
+            process.wait(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait()
+
+
+def run_shards(
+    shards: list[list[str]],
+    runtime: Path = RUNTIME,
+    *,
+    popen: Callable[..., subprocess.Popen] = subprocess.Popen,
+) -> list[ShardResult]:
+    workers: list[tuple[int, list[str], subprocess.Popen, object, float]] = []
+    try:
+        for index, shard in enumerate(shards):
+            regex = ci_shard.shard_regex(shard)
+            if not regex:
+                raise RunnerError(f"refusing to run empty shard {index}")
+            output = tempfile.TemporaryFile(mode="w+t", encoding="utf-8")
+            started = time.monotonic()
+            try:
+                process = popen(
+                    ["go", "test", "-count=1", "-run", regex, "./..."],
+                    cwd=runtime,
+                    stdout=output,
+                    stderr=subprocess.STDOUT,
+                    text=True,
+                )
+            except Exception:
+                output.close()
+                raise
+            workers.append((index, shard, process, output, started))
+
+        while any(process.poll() is None for _, _, process, _, _ in workers):
+            time.sleep(0.05)
+    except BaseException:
+        stop_processes([process for _, _, process, _, _ in workers])
+        for _, _, _, output, _ in workers:
+            output.close()
+        raise
+
+    results: list[ShardResult] = []
+    for index, shard, process, output, started in workers:
+        process.wait()
+        output.seek(0)
+        value = output.read()
+        output.close()
+        results.append(
+            ShardResult(
+                index=index,
+                test_count=len(shard),
+                returncode=process.returncode,
+                elapsed_seconds=time.monotonic() - started,
+                output=value,
+            )
+        )
+    return results
+
+
+def positive_int(value: str) -> int:
+    parsed = int(value)
+    if parsed < 1:
+        raise argparse.ArgumentTypeError("must be at least 1")
+    return parsed
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--jobs",
+        type=positive_int,
+        default=default_jobs(),
+        help=f"isolated test processes (default: about seven of ten CPUs, up to {MAX_DEFAULT_JOBS})",
+    )
+    args = parser.parse_args(argv)
+
+    started = time.monotonic()
+    try:
+        names = list_top_level_tests()
+        shards = verified_partition(names, args.jobs)
+        print(
+            f"Running {len(names)} tests across {len(shards)} isolated shards.",
+            flush=True,
+        )
+        results = run_shards(shards)
+    except KeyboardInterrupt:
+        print("Interrupted; all local test shards were stopped.", file=sys.stderr)
+        return 130
+    except (OSError, RunnerError) as error:
+        print(f"ERROR: {error}", file=sys.stderr)
+        return 2
+
+    failed = False
+    for result in sorted(results, key=lambda item: item.index):
+        label = "PASS" if result.returncode == 0 else "FAIL"
+        print(
+            f"{label} shard {result.index + 1}/{len(results)} "
+            f"({result.test_count} tests, {result.elapsed_seconds:.1f}s)"
+        )
+        if result.returncode != 0:
+            failed = True
+            if result.output:
+                print(result.output.rstrip(), file=sys.stderr)
+
+    elapsed = time.monotonic() - started
+    if failed:
+        print(f"FAIL: one or more shards failed after {elapsed:.1f}s.", file=sys.stderr)
+        return 1
+    print(f"PASS: all {len(names)} tests passed in {elapsed:.1f}s.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/tests/test_run_go_tests.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+import io
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+
+SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
+sys.path.insert(0, str(SCRIPTS))
+
+import ci_shard  # noqa: E402
+import run_go_tests  # noqa: E402
+
+
+class Completed:
+    def __init__(self, returncode: int, stdout: str = "", stderr: str = ""):
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+class ImmediateProcess:
+    def __init__(self, command, *, stdout, returncode=0, **_kwargs):
+        self.command = command
+        self.returncode = returncode
+        stdout.write("worker output\n")
+        stdout.flush()
+
+    def poll(self):
+        return self.returncode
+
+    def wait(self, timeout=None):
+        return self.returncode
+
+    def terminate(self):
+        self.returncode = -15
+
+    def kill(self):
+        self.returncode = -9
+
+
+class ActiveProcess(ImmediateProcess):
+    def __init__(self):
+        self.returncode = None
+        self.terminated = False
+
+    def terminate(self):
+        self.terminated = True
+        self.returncode = -15
+
+
+class LocalGoTestBoundary(unittest.TestCase):
+    def test_help_is_renderable(self):
+        with self.assertRaises(SystemExit) as stopped:
+            run_go_tests.main(["--help"])
+        self.assertEqual(stopped.exception.code, 0)
+
+    def test_default_uses_ten_of_this_macs_fourteen_cores(self):
+        with mock.patch.object(run_go_tests.os, "cpu_count", return_value=14):
+            self.assertEqual(run_go_tests.default_jobs(), 10)
+
+    # control-law: complete-local-test-partition.
+    def test_partition_is_exact_and_has_no_overlap(self):
+        names = [f"Test{index:03d}" for index in range(31)]
+        shards = run_go_tests.verified_partition(names, 6)
+        assigned = [name for shard in shards for name in shard]
+        self.assertEqual(sorted(assigned), names)
+        self.assertEqual(len(assigned), len(set(assigned)))
+
+    # control-law: complete-local-test-partition.
+    def test_empty_enumeration_fails_closed(self):
+        with self.assertRaisesRegex(run_go_tests.RunnerError, "empty test set"):
+            run_go_tests.verified_partition([], 6)
+
+    # control-law: complete-local-test-partition.
+    def test_enumeration_failure_cannot_start_workers(self):
+        def failed_run(*_args, **_kwargs):
+            return Completed(1, stderr="compile failed")
+
+        with self.assertRaisesRegex(run_go_tests.RunnerError, "compile failed"):
+            run_go_tests.list_top_level_tests(run=failed_run)
+
+    # control-law: complete-local-test-partition.
+    def test_every_partition_reaches_one_isolated_worker(self):
+        commands = []
+
+        def launch(command, **kwargs):
+            commands.append(command)
+            return ImmediateProcess(command, **kwargs)
+
+        names = ["TestAlpha", "TestBeta", "TestGamma", "TestDelta"]
+        shards = run_go_tests.verified_partition(names, 3)
+        results = run_go_tests.run_shards(shards, popen=launch)
+        self.assertEqual(len(results), len(shards))
+        self.assertTrue(all(result.returncode == 0 for result in results))
+        selected = []
+        for command in commands:
+            self.assertEqual(command[:4], ["go", "test", "-count=1", "-run"])
+            regex = command[4]
+            for name in names:
+                if name in regex:
+                    selected.append(name)
+        self.assertEqual(sorted(selected), sorted(names))
+
+    # control-law: complete-local-test-partition.
+    def test_interrupt_stops_active_workers(self):
+        first, second = ActiveProcess(), ActiveProcess()
+        run_go_tests.stop_processes([first, second])
+        self.assertTrue(first.terminated)
+        self.assertTrue(second.terminated)
+
+    # control-law: complete-local-test-partition.
+    def test_any_worker_failure_fails_the_aggregate_gate(self):
+        names = ["TestAlpha", "TestBeta"]
+        results = [
+            run_go_tests.ShardResult(0, 1, 0, 0.1, "ok"),
+            run_go_tests.ShardResult(1, 1, 1, 0.1, "failed assertion"),
+        ]
+        with (
+            mock.patch.object(run_go_tests, "list_top_level_tests", return_value=names),
+            mock.patch.object(run_go_tests, "run_shards", return_value=results),
+            mock.patch("sys.stdout", new_callable=io.StringIO),
+            mock.patch("sys.stderr", new_callable=io.StringIO) as stderr,
+        ):
+            self.assertEqual(run_go_tests.main(["--jobs", "2"]), 1)
+        self.assertIn("failed assertion", stderr.getvalue())
+
+
+class ExistingShardControllerContract(unittest.TestCase):
+    def test_runner_reuses_the_reviewed_controller(self):
+        names = [f"Test{index:03d}" for index in range(12)]
+        self.assertEqual(
+            run_go_tests.verified_partition(names, 4),
+            ci_shard.assign_shards(names, 4, {}),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

CONTRIBUTING.md

@@ -4,6 +4,20 @@ Boatstack is developed directly in this repository. Propose runtime, workflow, d
 
 Every pull request must pass the cross-platform runtime checks and the repository contract. Review tests, adapter changes, public claims, and context-size changes with the product diff.
 
+## Local verification
+
+Run the complete Go suite from the repository root:
+
+```sh
+python3 .github/scripts/run_go_tests.py
+```
+
+The runner verifies the full test partition, then uses a CPU-aware number of
+isolated serial Go processes. This keeps the suite complete without racing the
+mutable test seams inside one process. Use a focused `go test -run` command while
+iterating, then use the complete runner before review. CI keeps a serial Unix
+run as the independent correctness reference.
+
 Repository-specific examples and outcome reports can be proposed here as new evidence. A failure becomes a durable move only after its mechanism and non-regression gate are documented.
 
 ## Public-facing changes