ci: shard Windows go test into 6 balanced parallel jobs (~15m → <5m)

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

@@ -0,0 +1,70 @@
+# Windows CI test sharding
+
+## The problem
+
+The Boatstack Go suite (`boatstack/`) is ~330 tests in one package. Almost every
+test builds a real on-disk git repo, spawning several `git` subprocesses — on the
+order of 1,000–2,000 process spawns across the suite, run **serially**
+(`t.Parallel()` is not used anywhere).
+
+On Linux/macOS this is ~1–2 min. On Windows each `CreateProcess` (plus Microsoft
+Defender scanning the freshly written compile/link output) costs ~10× the Linux
+`fork+exec`, so the serial suite took **~15 min** — the process-spawn *latency*,
+not CPU, is the entire gap. Inline Defender exclusions in `ci.yml` were already
+applied and are not enough on their own.
+
+In-process `t.Parallel()` is **not** a safe fix here: the package swaps ~14
+mutable package-global function-seams (`runGitCommand`, `operationNow`,
+`hookDiagnosticRunner`, `fetchLatestRelease`, …) and uses many `t.Setenv` sites.
+Parallel tests within one process would race on that shared global state.
+
+## The fix: job-level sharding
+
+Run the Windows suite as **N separate runner processes**, each executing a
+disjoint, balanced subset of tests serially. Globals are per-process, so each
+shard keeps today's exact serial semantics; wall-clock drops ~N×. We use N=6,
+targeting a slowest-shard time well under 5 min.
+
+`.github/workflows/ci.yml` runs Unix (`test` job) as the full, unsharded
+correctness reference and Windows (`test-windows` job) as a
+`matrix: { shard: [0..5] }`. Each shard (working-directory `boatstack`):
+
+```bash
+regex=$(go test -list '^Test' ./... | python .github/scripts/ci_shard.py --total 6 --index <shard>)
+go test -run "$regex" ./...
+```
+
+`go test -list` and `go test -run` share the warm GOCACHE within a job, so the
+second compile is a cache hit.
+
+## The controller: `ci_shard.py`
+
+Assigning tests to shards to minimize the slowest shard is the classic
+multiprocessor-scheduling / makespan-minimization problem (P || Cmax, NP-hard).
+`ci_shard.py` uses the standard **LPT (Longest-Processing-Time-first) greedy**
+approximation (a 4/3 bound on optimal makespan): sort tests by descending
+estimated cost, place each on the currently-lightest shard.
+
+- **Default weight** is 1 per test (count-balanced) — already good because heavy
+  tests are spread across many names.
+- **Calibration (optional):** pass `--timings <json>` (a `{test-name: seconds}`
+  map from a prior `go test -json` run) to weight by measured runtime and close
+  the loop against the real cost envelope. No profile is committed yet; the
+  controller degrades gracefully to count-balancing without one.
+- `--verify` asserts the shards partition the input exactly (no dropped or
+  duplicated test). This invariant is unit-tested in `test_ci_shard.py`, which
+  the `ci-policy` workflow runs — a broken partition can't merge.
+
+## Keep the two copies in sync
+
+`.github/` is control-plane and is **not** projected by labkit, so the upstream
+`operatorstack/intelligence-flow` monorepo carries its own copy of `ci_shard.py`
+and its own sharded `runtime-windows` job in
+`.github/workflows/boatstack-lab.yml`. When you change the controller here,
+mirror it there (and vice versa).
+
+## Operational note
+
+Sharding renames the Windows status check (`test (windows-latest)` →
+`test-windows (shard 0..5)`). Update the branch-protection required-status-checks
+after merging, or PRs will wait on a check that no longer runs.

.github/scripts/ci_shard.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+"""Balanced test-shard selector — a makespan-minimization controller for CI.
+
+Why this exists
+---------------
+The Boatstack Go suite (~325 tests in one package) runs strictly serially, and
+almost every test spawns several `git` subprocesses. On Windows each process
+spawn is ~10x costlier than on Linux, so the serial suite takes ~15 min there
+(vs ~1 min on Linux) — process-spawn *latency*, not CPU, is the bottleneck.
+In-process `t.Parallel()` is unsafe here because the package swaps ~14 mutable
+global function-seams; parallel tests would race. The safe lever is job-level
+sharding: run N shards as N separate processes (globals are per-process), each
+running a disjoint subset of tests serially, so wall-clock drops ~Nx.
+
+Control-theory framing
+----------------------
+Assigning tests to shards to minimize the slowest shard is the classic
+multiprocessor-scheduling / makespan-minimization problem (P || Cmax), which is
+NP-hard. We use the standard **LPT (Longest-Processing-Time-first) greedy**
+approximation: sort tests by descending estimated cost, then place each on the
+currently-lightest shard. LPT is a 4/3-approximation of optimal makespan.
+
+Cost is a per-test weight. With no data every test weighs 1 (count-balanced,
+which is already good because the heavy tests are spread across many names).
+Passing `--timings <json>` (a map of test-name -> measured seconds) closes the
+loop with the measured runtime envelope, matching the calibration pattern used
+elsewhere in the repo.
+
+Usage
+-----
+    go test -list '^Test' ./... | \
+        python .github/scripts/ci_shard.py --total 6 --index 0
+        # -> prints a `go test -run` regex: ^(TestA|TestB|...)$
+
+    ... | python .github/scripts/ci_shard.py --total 6 --verify
+        # -> exits non-zero if the 6 shards don't partition the input exactly
+
+The workflow captures the printed regex and runs `go test -run "<regex>" ./...`.
+An empty selection prints nothing (exit 0); the caller must treat empty as
+"skip this shard", never as `go test -run ''` (which would run everything).
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+
+# `go test -list` prints one test name per line plus a trailing "ok  <pkg>  <t>"
+# summary line (and possibly blank lines). Real test names are Go identifiers
+# beginning with "Test"; keep only those.
+TEST_NAME = re.compile(r"^Test[A-Za-z0-9_]*$")
+
+
+def read_test_names(stream) -> list[str]:
+    """Parse `go test -list` output from a stream into a sorted, de-duped list."""
+    names = set()
+    for line in stream:
+        name = line.strip()
+        if TEST_NAME.match(name):
+            names.add(name)
+    return sorted(names)
+
+
+def assign_shards(names: list[str], total: int, timings: dict[str, float]) -> list[list[str]]:
+    """Partition `names` into `total` shards via LPT greedy on estimated cost.
+
+    Returns a list of `total` shard lists. Deterministic: tests are ordered by
+    (descending weight, name) before placement, and ties in shard load are
+    broken by lowest shard index, so the same input always yields the same
+    partition regardless of platform or run.
+    """
+    if total < 1:
+        raise ValueError("--total must be >= 1")
+
+    # Descending weight, then name, for a stable ordering.
+    ordered = sorted(names, key=lambda n: (-float(timings.get(n, 1.0)), n))
+
+    shards: list[list[str]] = [[] for _ in range(total)]
+    loads = [0.0] * total
+    for name in ordered:
+        # Lightest shard wins; ties -> lowest index (min is stable on first).
+        target = min(range(total), key=lambda i: (loads[i], i))
+        shards[target].append(name)
+        loads[target] += float(timings.get(name, 1.0))
+
+    # Emit each shard sorted by name for readable, stable regexes.
+    return [sorted(shard) for shard in shards]
+
+
+def shard_regex(shard: list[str]) -> str:
+    """Build an anchored `go test -run` alternation for one shard.
+
+    Go test names are identifiers, but escape defensively so a stray character
+    can never turn into an unintended regex. Empty shard -> empty string.
+    """
+    if not shard:
+        return ""
+    alternation = "|".join(re.escape(name) for name in shard)
+    return f"^({alternation})$"
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--total", type=int, required=True, help="Number of shards.")
+    parser.add_argument("--index", type=int, help="Shard index to emit (0-based).")
+    parser.add_argument(
+        "--timings",
+        help="Optional JSON file mapping test name -> measured seconds (LPT weights).",
+    )
+    parser.add_argument(
+        "--verify",
+        action="store_true",
+        help="Assert the shards partition the input exactly; print a summary; no regex.",
+    )
+    args = parser.parse_args(argv)
+
+    if args.total < 1:
+        print("error: --total must be >= 1", file=sys.stderr)
+        return 2
+    if not args.verify and args.index is None:
+        print("error: --index is required unless --verify is set", file=sys.stderr)
+        return 2
+    if args.index is not None and not (0 <= args.index < args.total):
+        print(f"error: --index must be in [0, {args.total})", file=sys.stderr)
+        return 2
+
+    timings: dict[str, float] = {}
+    if args.timings:
+        with open(args.timings, "r", encoding="utf-8") as fh:
+            timings = {str(k): float(v) for k, v in json.load(fh).items()}
+
+    names = read_test_names(sys.stdin)
+    shards = assign_shards(names, args.total, timings)
+
+    if args.verify:
+        assigned = [n for shard in shards for n in shard]
+        if sorted(assigned) != names or len(assigned) != len(names):
+            print(
+                "error: shards do not partition the input exactly "
+                f"(input={len(names)}, assigned={len(assigned)})",
+                file=sys.stderr,
+            )
+            return 1
+        sizes = ", ".join(f"#{i}={len(s)}" for i, s in enumerate(shards))
+        print(f"partition OK: {len(names)} tests across {args.total} shards ({sizes})")
+        return 0
+
+    print(shard_regex(shards[args.index]))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/scripts/test_ci_shard.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""Unit tests for the balanced test-shard selector (ci_shard.py).
+
+Run: python .github/scripts/test_ci_shard.py
+The CI-speed-policy workflow runs these so the sharding controller is itself
+gated — a broken partition would silently drop tests from every shard.
+"""
+from __future__ import annotations
+
+import io
+import unittest
+
+import ci_shard
+
+
+def names(n: int) -> list[str]:
+    return [f"Test{i:03d}" for i in range(n)]
+
+
+class ReadTestNames(unittest.TestCase):
+    def test_filters_non_test_lines(self):
+        raw = "TestAlpha\nTestBeta\nok  \texample/pkg\t1.2s\n\nhelperFunc\n"
+        got = ci_shard.read_test_names(io.StringIO(raw))
+        self.assertEqual(got, ["TestAlpha", "TestBeta"])
+
+    def test_dedupes_and_sorts(self):
+        raw = "TestB\nTestA\nTestB\n"
+        self.assertEqual(ci_shard.read_test_names(io.StringIO(raw)), ["TestA", "TestB"])
+
+
+class Partition(unittest.TestCase):
+    def test_union_is_input_no_overlap(self):
+        for total in (1, 2, 3, 6, 7):
+            for count in (0, 1, 5, 50, 325):
+                ns = names(count)
+                shards = ci_shard.assign_shards(ns, total, {})
+                self.assertEqual(len(shards), total)
+                flat = [n for s in shards for n in s]
+                self.assertEqual(sorted(flat), ns, (total, count))
+                self.assertEqual(len(flat), len(set(flat)), (total, count))
+
+    def test_deterministic(self):
+        ns = names(97)
+        a = ci_shard.assign_shards(ns, 6, {})
+        b = ci_shard.assign_shards(list(reversed(ns)), 6, {})
+        self.assertEqual(a, b)
+
+    def test_count_balanced_without_timings(self):
+        shards = ci_shard.assign_shards(names(300), 6, {})
+        sizes = [len(s) for s in shards]
+        self.assertLessEqual(max(sizes) - min(sizes), 1)
+
+    def test_lpt_balances_weighted_load(self):
+        # One very heavy test plus many light ones: LPT must isolate the heavy
+        # one and spread the rest so the makespan stays near optimal.
+        ns = names(20)
+        timings = {ns[0]: 100.0}
+        for n in ns[1:]:
+            timings[n] = 1.0
+        shards = ci_shard.assign_shards(ns, 4, timings)
+        loads = [sum(timings[n] for n in s) for s in shards]
+        # Optimal makespan is dominated by the 100s test; LPT must not exceed it
+        # by more than one light unit of slack per the 4/3 bound on this input.
+        self.assertLessEqual(max(loads), 106.0)
+
+    def test_more_shards_than_tests_leaves_empties(self):
+        shards = ci_shard.assign_shards(names(2), 6, {})
+        non_empty = [s for s in shards if s]
+        self.assertEqual(sum(len(s) for s in shards), 2)
+        self.assertEqual(len(non_empty), 2)
+
+
+class Regex(unittest.TestCase):
+    def test_anchored_alternation(self):
+        self.assertEqual(ci_shard.shard_regex(["TestA", "TestB"]), "^(TestA|TestB)$")
+
+    def test_empty_shard_is_empty_string(self):
+        self.assertEqual(ci_shard.shard_regex([]), "")
+
+    def test_escapes_metacharacters(self):
+        # Defensive: a name with regex metacharacters must be escaped, not
+        # interpreted, so it can never widen the selection.
+        self.assertEqual(ci_shard.shard_regex(["Test.A+"]), r"^(Test\.A\+)$")
+
+
+class Cli(unittest.TestCase):
+    def _run(self, argv, stdin_text):
+        import contextlib
+
+        out, err = io.StringIO(), io.StringIO()
+        stdin = io.StringIO(stdin_text)
+        with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+            import sys
+
+            saved = sys.stdin
+            sys.stdin = stdin
+            try:
+                code = ci_shard.main(argv)
+            finally:
+                sys.stdin = saved
+        return code, out.getvalue(), err.getvalue()
+
+    def test_index_prints_regex(self):
+        code, out, _ = self._run(["--total", "2", "--index", "0"], "TestA\nTestB\n")
+        self.assertEqual(code, 0)
+        self.assertTrue(out.strip().startswith("^("))
+
+    def test_verify_ok(self):
+        stdin = "".join(f"{n}\n" for n in names(50))
+        code, out, _ = self._run(["--total", "6", "--verify"], stdin)
+        self.assertEqual(code, 0)
+        self.assertIn("partition OK", out)
+
+    def test_index_out_of_range(self):
+        code, _, err = self._run(["--total", "2", "--index", "5"], "TestA\n")
+        self.assertEqual(code, 2)
+        self.assertIn("--index", err)
+
+    def test_missing_index_without_verify(self):
+        code, _, err = self._run(["--total", "2"], "TestA\n")
+        self.assertEqual(code, 2)
+        self.assertIn("--index is required", err)
+
+    def test_empty_shard_prints_nothing(self):
+        # 2 tests, 6 shards: shards 2..5 are empty -> empty stdout, exit 0.
+        code, out, _ = self._run(["--total", "6", "--index", "5"], "TestA\nTestB\n")
+        self.assertEqual(code, 0)
+        self.assertEqual(out.strip(), "")
+
+
+if __name__ == "__main__":
+    unittest.main()