ci: run consensus_sim test cases in parallel

.github/workflows/docker-pr-build.yml

@@ -11,6 +11,13 @@ on:
     # and a required check that never runs keeps the PR blocked forever. See
     # tests.yml for the same reasoning.
 
+# A new push to a PR makes the previous run's result irrelevant; cancel it instead
+# of letting a ~30 min run hold runners. Master pushes are never cancelled
+# (group includes the unique run id there), so every merged commit is tested.
+concurrency:
+  group: docker-pr-${{ github.event_name == 'pull_request' && github.event.pull_request.number || github.run_id }}
+  cancel-in-progress: true
+
 jobs:
   build:
     name: Build and push Docker image to Docker Hub

.github/workflows/tests.yml

@@ -9,6 +9,13 @@ on:
   # public-repo minutes are free.
   pull_request:
 
+# A new push to a PR makes the previous run's result irrelevant; cancel it instead
+# of letting a ~30 min run hold runners. Master pushes are never cancelled
+# (group includes the unique run id there), so every merged commit is tested.
+concurrency:
+  group: tests-${{ github.event_name == 'pull_request' && github.event.pull_request.number || github.run_id }}
+  cancel-in-progress: true
+
 jobs:
   consensus-and-pm:
     name: consensus_sim + pm unit tests (distro Boost)
@@ -98,14 +105,27 @@ jobs:
           # Pre-existing ASan/UBSan findings in chain code, see tests/consensus_sim/README.md
           ASAN_OPTIONS: new_delete_type_mismatch=0:detect_leaks=0
           UBSAN_OPTIONS: halt_on_error=0:print_stacktrace=0
-        run: cd build && ctest --output-on-failure
+        run: |
+          set -euo pipefail
+          cd build
+          ctest -E '^consensus_sim$' --output-on-failure
+          # consensus_sim is one Boost.Test binary that ran ~15 min on a single core.
+          # Run its cases as separate processes across all cores instead; same cases,
+          # same working directory as ctest (failure logs land in the same place).
+          cd tests/consensus_sim
+          # --min-cases: floor of enabled cases in a non-testnet build. Raise it when adding
+          # cases; a drop below it means a test file fell out of the build.
+          python3 ../../../tests/consensus_sim/run_parallel.py ./consensus_sim_tests -j"$(nproc)" --min-cases 118
 
       - name: Upload consensus_sim failure logs
         if: failure()
         uses: actions/upload-artifact@v4
         with:
           name: consensus-sim-failures
-          path: tests/consensus_sim/failures/
+          # The harness writes failures/ under its working directory, which is the
+          # BUILD tree (build/tests/consensus_sim); the source-tree dir only holds
+          # .gitkeep, so the old path uploaded an empty artifact.
+          path: build/tests/consensus_sim/failures/
           if-no-files-found: ignore
 
   # Mandatory on every code PR and master push: real two-signer BFT coverage
@@ -174,9 +194,11 @@ jobs:
       - name: Run the simulator
         run: |
           set -euo pipefail
-          cd build
+          cd build/tests/consensus_sim
+          # --min-cases: floor of enabled cases in a testnet build (checked before --filter).
+          run=(python3 ../../../tests/consensus_sim/run_parallel.py ./consensus_sim_tests -j"$(nproc)" --min-cases 173)
           if [ "${{ matrix.validators }}" = "2" ]; then
-            tests/consensus_sim/consensus_sim_tests --run_test=hf16_boundary_suite,hf16_production_safety_suite,validator_committed_state_suite --log_level=test_suite
+            "${run[@]}" --filter hf16_boundary_suite,hf16_production_safety_suite,validator_committed_state_suite
           else
-            ctest -R '^consensus_sim$' --output-on-failure
+            "${run[@]}"
           fi

tests/consensus_sim/README.md

@@ -22,9 +22,24 @@ Run a specific suite:
 
 ```
 ./tests/consensus_sim/consensus_sim_tests --run_test=equivocation_suite
-./tests/consensus_sim/consensus_sim_tests --run_test=equivocation_suite/seed_sweep_one_hundred
+./tests/consensus_sim/consensus_sim_tests --run_test=equivocation_suite/seed_sweep_one_hundred_all_fire     # all 100 seeds
+./tests/consensus_sim/consensus_sim_tests --run_test=equivocation_suite/seed_sweep_one_hundred_all_fire/_3  # seeds 30-39
 ```
 
+Run every case as its own process, N at a time (what CI does; prints the
+slowest cases at the end):
+
+```
+cd build/tests/consensus_sim
+python3 ../../../tests/consensus_sim/run_parallel.py ./consensus_sim_tests -j4
+python3 ../../../tests/consensus_sim/run_parallel.py ./consensus_sim_tests --filter equivocation_suite
+```
+
+The case list is read from two Boost.Test listings that must agree, and CI
+passes `--min-cases` (118 for a regular build, 173 with `BUILD_TESTNET=ON`) so
+a test file dropped from the build fails the job instead of shrinking it.
+Raise the floors in `.github/workflows/tests.yml` when you add cases.
+
 The chain code has pre-existing ASan/UBSan findings (a base-pointer delete in
 `evaluator_registry`, version/asset alignment) unrelated to the harness. Run
 with `ASAN_OPTIONS=new_delete_type_mismatch=0:detect_leaks=0

tests/consensus_sim/run_parallel.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""Run consensus_sim_tests one test case per process, N processes at a time.
+
+The suite is a single Boost.Test binary; run serially it takes ~15-20 min on a
+4-core CI runner, almost all of it in one core. Test cases are independent
+(each builds its own node in a unique temp dir, failure logs are named by seed
+and scenario), so they can run side by side.
+
+Usage:
+    run_parallel.py BINARY [-j N] [--filter SUITE[,SUITE...]] [--min-cases N]
+                    [-- EXTRA_ARGS...]
+
+The case list comes from the DOT listing (`--list_content=DOT`) and must match
+the indented `--list_content` listing exactly; any mismatch, or fewer enabled
+cases than --min-cases, aborts the run instead of quietly running a subset.
+Disabled cases are counted and listed.
+
+Exit status is 0 only if every case passed. The slowest cases are printed at
+the end so the split can be checked from the CI log.
+"""
+import argparse
+import re
+import os
+import subprocess
+import sys
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+
+def _list_output(binary, fmt=None):
+    arg = "--list_content" + ("=" + fmt if fmt else "")
+    # Boost.Test prints the tree to stderr on older versions and stdout on newer.
+    return subprocess.run([binary, arg], stdout=subprocess.PIPE,
+                          stderr=subprocess.STDOUT, text=True, check=True).stdout
+
+
+def list_cases_hrf(binary):
+    """Enabled leaf paths from the indented listing (a trailing '*' = enabled).
+
+    A row is a leaf when the next row is not indented deeper. An empty suite
+    also looks like a leaf, but it is never enabled, so the enabled list is
+    still exactly the enabled test cases.
+    """
+    out = _list_output(binary)
+    rows = []
+    for line in out.splitlines():
+        name = line.strip()
+        if not name:
+            continue
+        indent = len(line) - len(line.lstrip(" "))
+        rows.append((indent, name.rstrip("*").strip(), name.endswith("*")))
+    if not rows:
+        sys.exit("run_parallel: --list_content returned nothing:\n" + out)
+
+    step = min((i for i, _, _ in rows if i > 0), default=4)
+    cases, stack = [], []
+    for idx, (indent, name, enabled) in enumerate(rows):
+        del stack[indent // step:]
+        stack.append(name)
+        is_leaf = idx + 1 == len(rows) or rows[idx + 1][0] <= indent
+        if is_leaf and enabled:
+            cases.append("/".join(stack))
+    return cases
+
+
+# DOT listing: `tuN[...,color=green|yellow,label="name|file(line)"];` per node and
+# `tuP -> tuC;` per edge. Boost.Test gives test cases ids with bit 16 set
+# (TUT_CASE = 0x10000) and suites ids without it, so cases are identified by id,
+# not by layout. Enabled nodes are green, disabled ones yellow.
+_DOT_NODE = re.compile(r'^tu(\d+)\[.*color=(\w+),label="([^|"]*)')
+_DOT_EDGE = re.compile(r'^tu(\d+) -> tu(\d+);')
+_TUT_CASE = 0x10000
+
+
+def list_cases_dot(binary):
+    """(enabled, disabled) test-case paths from the DOT listing."""
+    nodes, parent, order = {}, {}, []
+    for line in _list_output(binary, "DOT").splitlines():
+        line = line.strip()
+        m = _DOT_NODE.match(line)
+        if m:
+            nid = int(m.group(1))
+            nodes[nid] = (m.group(3), m.group(2) == "green")
+            order.append(nid)
+            continue
+        m = _DOT_EDGE.match(line)
+        if m:
+            parent[int(m.group(2))] = int(m.group(1))
+
+    def path(nid):
+        parts = []
+        while nid in parent:          # the master suite has no parent and no name in paths
+            parts.append(nodes[nid][0])
+            nid = parent[nid]
+        return "/".join(reversed(parts))
+
+    enabled, disabled = [], []
+    for nid in order:
+        if nid & _TUT_CASE:
+            (enabled if nodes[nid][1] else disabled).append(path(nid))
+    return enabled, disabled
+
+
+def list_cases(binary):
+    """Enabled and disabled case paths, from two independent listings that must agree.
+
+    A Boost.Test format change that made one parser return a subset would
+    otherwise let CI go green having run fewer cases.
+    """
+    enabled, disabled = list_cases_dot(binary)
+    hrf = list_cases_hrf(binary)
+    if sorted(hrf) != sorted(enabled):
+        only_dot = sorted(set(enabled) - set(hrf))[:5]
+        only_hrf = sorted(set(hrf) - set(enabled))[:5]
+        sys.exit(f"run_parallel: case list mismatch: DOT listing has {len(enabled)} enabled "
+                 f"cases, indented listing {len(hrf)} (only in DOT: {only_dot}, "
+                 f"only in indented: {only_hrf}). Refusing to run a partial set.")
+    return enabled, disabled
+
+
+def run_case(binary, case, extra):
+    start = time.monotonic()
+    proc = subprocess.run([binary, "--run_test=" + case] + extra,
+                          stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    rc = proc.returncode
+    # A filter that matches nothing is a "Test setup error" with exit 0 on some
+    # Boost versions; that must not count as a pass.
+    if rc == 0 and "no test cases matching filter" in proc.stdout:
+        rc = 2
+    return case, rc, time.monotonic() - start, proc.stdout
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("binary")
+    ap.add_argument("-j", "--jobs", type=int, default=os.cpu_count() or 2)
+    ap.add_argument("--filter", help="comma-separated top-level suites to run")
+    ap.add_argument("--list", action="store_true", help="print the cases and exit")
+    ap.add_argument("--min-cases", type=int, default=0,
+                    help="fail if the binary has fewer enabled cases than this (before --filter)")
+    ap.add_argument("extra", nargs="*", help="extra args passed to every run (after --)")
+    args = ap.parse_args()
+
+    binary = os.path.abspath(args.binary)
+    cases, disabled = list_cases(binary)
+
+    if args.min_cases and len(cases) < args.min_cases:
+        sys.exit(f"run_parallel: only {len(cases)} enabled cases, expected at least "
+                 f"{args.min_cases}. Was a test file dropped from the build?")
+
+    if args.filter:
+        keep = set(args.filter.split(","))
+        missing = keep - {c.split("/")[0] for c in cases}
+        if missing:
+            sys.exit("run_parallel: --filter names no enabled cases: " + ", ".join(sorted(missing)))
+        cases = [c for c in cases if c.split("/")[0] in keep]
+        disabled = [c for c in disabled if c.split("/")[0] in keep]
+    if args.list:
+        print("\n".join(cases))
+        return 0
+    if not cases:
+        sys.exit("run_parallel: no test cases selected")
+
+    print(f"run_parallel: {len(cases)} cases, {args.jobs} at a time; "
+          f"skipped (disabled): {len(disabled)}", flush=True)
+    for case in disabled:
+        print(f"  disabled: {case}", flush=True)
+    t0 = time.monotonic()
+    results, failed = [], []
+    with ThreadPoolExecutor(max_workers=args.jobs) as pool:
+        futures = [pool.submit(run_case, binary, c, args.extra) for c in cases]
+        for fut in as_completed(futures):
+            case, rc, dt, out = fut.result()
+            results.append((dt, case))
+            status = "ok" if rc == 0 else f"FAILED (exit {rc})"
+            print(f"[{len(results)}/{len(cases)}] {case}: {status} {dt:.1f}s", flush=True)
+            if rc != 0:
+                failed.append((case, out))
+
+    for case, out in failed:
+        print(f"\n===== {case} =====\n{out}", flush=True)
+
+    print(f"\nrun_parallel: wall {time.monotonic() - t0:.0f}s, "
+          f"sum of cases {sum(d for d, _ in results):.0f}s")
+    print("slowest cases:")
+    for dt, case in sorted(results, reverse=True)[:10]:
+        print(f"  {dt:7.1f}s  {case}")
+
+    if failed:
+        print(f"\nrun_parallel: {len(failed)} of {len(cases)} cases failed:")
+        for case, _ in failed:
+            print(f"  {case}")
+        return 1
+    print(f"run_parallel: all {len(cases)} cases passed")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/consensus_sim/scenarios/test_equivocation.cpp

@@ -1,4 +1,6 @@
 #include <boost/test/unit_test.hpp>
+#include <boost/test/data/test_case.hpp>
+#include <boost/test/data/monomorphic.hpp>
 
 #include "fault_injector.hpp"
 #include "failure_log.hpp"
@@ -47,14 +49,26 @@ BOOST_AUTO_TEST_CASE(seed_deadbeef_fires_chains_consistent) {
     BOOST_CHECK_EQUAL(result->block_num, 2u);
 }
 
-BOOST_AUTO_TEST_CASE(seed_sweep_one_hundred_all_fire) {
-    constexpr uint64_t kSweepCount = 100;
+// 100 seeds, 9 s each under ASan, used to be one ~15 min test case. Split into
+// kSweepChunks data-driven cases (seed_sweep_one_hundred_all_fire/_0 .. /_9),
+// each covering a contiguous block of seeds, so run_parallel.py can spread
+// them across cores. The union is still seeds 0..kSweepCount-1.
+namespace {
+constexpr uint64_t kSweepCount = 100;
+constexpr uint64_t kSweepChunks = 10;
+static_assert(kSweepCount % kSweepChunks == 0, "chunks must cover every seed exactly once");
+}
+
+BOOST_DATA_TEST_CASE(seed_sweep_one_hundred_all_fire,
+                     boost::unit_test::data::xrange(kSweepChunks), chunk) {
+    constexpr uint64_t per_chunk = kSweepCount / kSweepChunks;
+    const uint64_t first = static_cast<uint64_t>(chunk) * per_chunk;
     uint32_t missed = 0;
 
     // Every seed should produce the same equivocation outcome: chains_consistent
     // fires at slot 2. The sweep proves the mechanism is robust across seeds,
     // and that the violation is reproducible per-seed.
-    for (uint64_t seed = 0; seed < kSweepCount; ++seed) {
+    for (uint64_t seed = first; seed < first + per_chunk; ++seed) {
         scenario_config cfg;
         cfg.seed = seed;
         cfg.num_witnesses = 7;