Merge pull request 'feat(ci): run L0/L1 offline checks, mutation suites and yamllint in bounded parallel pools' (#149) from feat/parallel-l0-checks into main

Reviewed-on: #149
Reviewed-by: Keith Jackson <[email protected]>

conformance/tests/l0-static/keycloak-client-secret-reconciler/mutations.py

@@ -25,15 +25,19 @@
 #
 #   python3 mutations.py <path-to-extracted-reconciler.py>
 #
-# Runtime is ~20s (one subprocess per mutation, each running the full selftest
-# against an in-process stub Keycloak; no network, no cluster).
+# Runtime is ~11s serial (one subprocess per mutation, each running the full selftest
+# against an in-process stub Keycloak; no network, no cluster). The selftests run in a
+# bounded pool (env MUTATION_JOBS, default min(8, CPUs); 1 = serial); output order and
+# verdicts are independent of the pool size.
 # =============================================================================
 from __future__ import annotations
 
+import os
 import re
 import subprocess
 import sys
 import tempfile
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 
@@ -136,43 +140,99 @@ MUTATIONS: list[tuple[str, str, str, str]] = [
 ]
 
 
+def mutation_jobs() -> int:
+    """Selftests to run at once: env MUTATION_JOBS, else min(8, usable CPUs).
+
+    conformance/lib/run_l1_offline.py exports a smaller MUTATION_JOBS when it is itself
+    running several checks in parallel, so the nested pools do not oversubscribe the host.
+    MUTATION_JOBS=1 reproduces the old serial loop exactly.
+    """
+    raw = os.environ.get("MUTATION_JOBS", "").strip()
+    if raw:
+        try:
+            n = int(raw)
+        except ValueError:
+            n = 0
+        if n < 1:
+            raise ValueError("MUTATION_JOBS must be a positive integer, got %r" % raw)
+        return n
+    try:
+        cpus = len(os.sched_getaffinity(0))
+    except (AttributeError, OSError):
+        cpus = os.cpu_count() or 1
+    return max(1, min(8, cpus))
+
+
+def run_selftest(mutated_src: str) -> subprocess.CompletedProcess:
+    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh:
+        fh.write(mutated_src)
+        path = fh.name
+    try:
+        return subprocess.run([sys.executable, path, "--selftest"],
+                              capture_output=True, text=True, timeout=240,
+                              stdin=subprocess.DEVNULL)
+    finally:
+        Path(path).unlink(missing_ok=True)
+
+
 def main() -> int:
     if len(sys.argv) != 2:
         print("usage: mutations.py <extracted-reconciler.py>", file=sys.stderr)
         return 2
     src = Path(sys.argv[1]).read_text()
 
-    problems: list[str] = []
-    killed = 0
+    try:
+        jobs = mutation_jobs()
+    except ValueError as e:
+        print("mutations.py: %s" % e, file=sys.stderr)
+        return 2
+
+    # Pass 1, serial and cheap: every anchor is checked BEFORE anything runs, exactly as the
+    # serial loop did, and each mutation's source is built here. None = anchor failure.
+    planned: list[tuple[str, str, str | None, str | None]] = []
     for name, why, old, new in MUTATIONS:
         old, new = dedent4(old), dedent4(new)
         # An anchor that no longer matches is a FAILURE, not a skip: it means the code
         # moved and this mutation silently stopped proving anything.
         n = src.count(old)
         if n != 1:
-            problems.append("%s: anchor matches %d times (expected exactly 1) -- the "
+            planned.append((name, why,
+                            "%s: anchor matches %d times (expected exactly 1) -- the "
                             "code moved and this mutation no longer proves anything. "
-                            "Re-anchor it against the current source." % (name, n))
+                            "Re-anchor it against the current source." % (name, n), None))
             continue
-        with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh:
-            fh.write(src.replace(old, new, 1))
-            path = fh.name
-        try:
-            proc = subprocess.run([sys.executable, path, "--selftest"],
-                                  capture_output=True, text=True, timeout=240)
-        finally:
-            Path(path).unlink(missing_ok=True)
-        if proc.returncode == 0:
-            problems.append("%s SURVIVED -- --selftest still passed with this regression "
-                            "applied, so nothing in the suite actually checks it. The "
-                            "mistake it models: %s" % (name, why))
-            continue
-        killed += 1
-        out = proc.stdout + proc.stderr
-        m = re.search(r"SELFTEST FAIL: (\d+) of \d+", out)
-        how = ("%s assertion(s) caught it" % m.group(1) if m
-               else "the selftest aborted (caught, less cleanly)")
-        print("  killed  %-42s %s" % (name, how))
+        planned.append((name, why, None, src.replace(old, new, 1)))
+
+    # Pass 2: the selftests, `jobs` at a time. Each child gets its own temp file, its own
+    # mkdtemp dirs and its own 127.0.0.1:0 stub port, so they share nothing. Results are
+    # consumed in MUTATIONS order, so every line below is printed in the same order, with the
+    # same verdict, as the serial loop produced -- whichever child finishes first.
+    problems: list[str] = []
+    killed = 0
+    pool = ThreadPoolExecutor(max_workers=jobs)
+    try:
+        futures = [pool.submit(run_selftest, mutated) if mutated is not None else None
+                   for (_n, _w, _a, mutated) in planned]
+        for (name, why, anchor_problem, _m), fut in zip(planned, futures):
+            if anchor_problem is not None:
+                problems.append(anchor_problem)
+                continue
+            proc = fut.result()
+            if proc.returncode == 0:
+                problems.append("%s SURVIVED -- --selftest still passed with this regression "
+                                "applied, so nothing in the suite actually checks it. The "
+                                "mistake it models: %s" % (name, why))
+                continue
+            killed += 1
+            out = proc.stdout + proc.stderr
+            m = re.search(r"SELFTEST FAIL: (\d+) of \d+", out)
+            how = ("%s assertion(s) caught it" % m.group(1) if m
+                   else "the selftest aborted (caught, less cleanly)")
+            print("  killed  %-42s %s" % (name, how))
+    finally:
+        # A timeout (or any other exception) still propagates and fails the check, as it did
+        # serially; queued mutations are cancelled rather than run for nothing.
+        pool.shutdown(wait=True, cancel_futures=True)
 
     print()
     if problems:

conformance/tests/l0-static/keycloak-passkey-reconciler/mutations.py

@@ -40,15 +40,19 @@
 #
 #   python3 mutations.py <path-to-extracted-reconciler.py>
 #
-# Runtime is ~60s (one subprocess per mutation, each running the full selftest
-# against an in-process stub Keycloak; no network, no cluster).
+# Runtime is ~85s serial (one subprocess per mutation, each running the full selftest
+# against an in-process stub Keycloak; no network, no cluster). The selftests run in a
+# bounded pool (env MUTATION_JOBS, default min(8, CPUs); 1 = serial); output order and
+# verdicts are independent of the pool size.
 # =============================================================================
 from __future__ import annotations
 
+import os
 import re
 import subprocess
 import sys
 import tempfile
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 # (name, the mistake being modelled, anchor, replacement)
@@ -403,42 +407,98 @@ MUTATIONS: list[tuple[str, str, str, str]] = [
 ]
 
 
+def mutation_jobs() -> int:
+    """Selftests to run at once: env MUTATION_JOBS, else min(8, usable CPUs).
+
+    conformance/lib/run_l1_offline.py exports a smaller MUTATION_JOBS when it is itself
+    running several checks in parallel, so the nested pools do not oversubscribe the host.
+    MUTATION_JOBS=1 reproduces the old serial loop exactly.
+    """
+    raw = os.environ.get("MUTATION_JOBS", "").strip()
+    if raw:
+        try:
+            n = int(raw)
+        except ValueError:
+            n = 0
+        if n < 1:
+            raise ValueError("MUTATION_JOBS must be a positive integer, got %r" % raw)
+        return n
+    try:
+        cpus = len(os.sched_getaffinity(0))
+    except (AttributeError, OSError):
+        cpus = os.cpu_count() or 1
+    return max(1, min(8, cpus))
+
+
+def run_selftest(mutated_src: str) -> subprocess.CompletedProcess:
+    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh:
+        fh.write(mutated_src)
+        path = fh.name
+    try:
+        return subprocess.run([sys.executable, path, "--selftest"],
+                              capture_output=True, text=True, timeout=300,
+                              stdin=subprocess.DEVNULL)
+    finally:
+        Path(path).unlink(missing_ok=True)
+
+
 def main() -> int:
     if len(sys.argv) != 2:
         print("usage: mutations.py <extracted-reconciler.py>", file=sys.stderr)
         return 2
     src = Path(sys.argv[1]).read_text()
 
-    problems: list[str] = []
-    killed = 0
+    try:
+        jobs = mutation_jobs()
+    except ValueError as e:
+        print("mutations.py: %s" % e, file=sys.stderr)
+        return 2
+
+    # Pass 1, serial and cheap: every anchor is checked BEFORE anything runs, exactly as the
+    # serial loop did, and each mutation's source is built here. None = anchor failure.
+    planned: list[tuple[str, str, str | None, str | None]] = []
     for name, why, old, new in MUTATIONS:
         # An anchor that no longer matches is a FAILURE, not a skip: it means the code
         # moved and this mutation silently stopped proving anything.
         n = src.count(old)
         if n != 1:
-            problems.append("%s: anchor matches %d times (expected exactly 1) -- the "
+            planned.append((name, why,
+                            "%s: anchor matches %d times (expected exactly 1) -- the "
                             "code moved and this mutation no longer proves anything. "
-                            "Re-anchor it against the current source." % (name, n))
+                            "Re-anchor it against the current source." % (name, n), None))
             continue
-        with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh:
-            fh.write(src.replace(old, new, 1))
-            path = fh.name
-        try:
-            proc = subprocess.run([sys.executable, path, "--selftest"],
-                                  capture_output=True, text=True, timeout=300)
-        finally:
-            Path(path).unlink(missing_ok=True)
-        if proc.returncode == 0:
-            problems.append("%s SURVIVED -- --selftest still passed with this regression "
-                            "applied, so nothing in the suite actually checks it. The "
-                            "mistake it models: %s" % (name, why))
-            continue
-        killed += 1
-        out = proc.stdout + proc.stderr
-        m = re.search(r"SELFTEST FAIL: (\d+) of \d+", out)
-        how = ("%s assertion(s) caught it" % m.group(1) if m
-               else "the selftest aborted (caught, less cleanly)")
-        print("  killed  %-54s %s" % (name, how))
+        planned.append((name, why, None, src.replace(old, new, 1)))
+
+    # Pass 2: the selftests, `jobs` at a time. Each child gets its own temp file, its own
+    # mkdtemp dirs and its own 127.0.0.1:0 stub port, so they share nothing. Results are
+    # consumed in MUTATIONS order, so every line below is printed in the same order, with the
+    # same verdict, as the serial loop produced -- whichever child finishes first.
+    problems: list[str] = []
+    killed = 0
+    pool = ThreadPoolExecutor(max_workers=jobs)
+    try:
+        futures = [pool.submit(run_selftest, mutated) if mutated is not None else None
+                   for (_n, _w, _a, mutated) in planned]
+        for (name, why, anchor_problem, _m), fut in zip(planned, futures):
+            if anchor_problem is not None:
+                problems.append(anchor_problem)
+                continue
+            proc = fut.result()
+            if proc.returncode == 0:
+                problems.append("%s SURVIVED -- --selftest still passed with this regression "
+                                "applied, so nothing in the suite actually checks it. The "
+                                "mistake it models: %s" % (name, why))
+                continue
+            killed += 1
+            out = proc.stdout + proc.stderr
+            m = re.search(r"SELFTEST FAIL: (\d+) of \d+", out)
+            how = ("%s assertion(s) caught it" % m.group(1) if m
+                   else "the selftest aborted (caught, less cleanly)")
+            print("  killed  %-54s %s" % (name, how))
+    finally:
+        # A timeout (or any other exception) still propagates and fails the check, as it did
+        # serially; queued mutations are cancelled rather than run for nothing.
+        pool.shutdown(wait=True, cancel_futures=True)
 
     print()
     if problems: