bsp: fix compile-server self-deadlock, cut ECJ and footprint overhead, keep CI telemetry

.github/scripts/summarise-bsp-metrics.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+"""Print what the compile server did during a CI run, from its metrics.jsonl.
+
+The point is that the common questions should not require downloading an artifact. Two of them
+recur often enough to be worth answering on every run:
+
+  "did each compile get slower, or did fewer of them run at once?"
+      Those have different causes and different fixes, and the distinction is invisible in a wall
+      clock. Per-project compile durations answer the first; machine occupancy answers the second.
+
+  "is bleep leaving the machine idle while work is queued?"
+      Idle cores with a non-empty queue means admission is refusing work it could take — a
+      scheduling bug rather than a slow compiler.
+
+Reads every metrics.jsonl under the directory given, tolerates truncated or malformed lines (a
+server killed mid-write leaves one), and never fails the build: a diagnostic that breaks the run it
+is diagnosing is worse than no diagnostic.
+"""
+
+from __future__ import annotations
+
+import json
+import pathlib
+import sys
+from collections import Counter, defaultdict
+
+
+def load(path: pathlib.Path) -> list[dict]:
+    rows = []
+    with path.open(encoding="utf-8", errors="replace") as fh:
+        for line in fh:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                rows.append(json.loads(line))
+            except ValueError:
+                continue  # truncated final line from a killed server
+    return rows
+
+
+def fmt_ms(ms: float) -> str:
+    return f"{ms / 1000:.1f}s" if ms < 60_000 else f"{ms / 60_000:.1f}m"
+
+
+def summarise(path: pathlib.Path, rows: list[dict]) -> None:
+    print(f"\n=== {path.parent.name} — {len(rows)} events ===")
+    by_type = Counter(r.get("type") for r in rows)
+    print("  events: " + ", ".join(f"{k}={v}" for k, v in by_type.most_common()))
+
+    compiles = [r for r in rows if r.get("type") == "compile_end"]
+    if compiles:
+        total = sum(r.get("duration_ms", 0) for r in compiles)
+        failed = sum(1 for r in compiles if not r.get("success", True))
+        print(f"\n  compiles: {len(compiles)} ({failed} failed), total task time {fmt_ms(total)}")
+        worst = sorted(compiles, key=lambda r: -r.get("duration_ms", 0))[:8]
+        print("  slowest:")
+        for r in worst:
+            print(f"    {fmt_ms(r.get('duration_ms', 0)):>8}  {r.get('project', '?')}")
+
+    allocs = [r for r in rows if r.get("type") == "compile_allocation"]
+    if allocs:
+        per = defaultdict(int)
+        for r in allocs:
+            per[r.get("project", "?")] += r.get("allocated_mb", 0)
+        print(f"\n  allocation attributed for {len(allocs)} compiles; heaviest:")
+        for proj, mb in sorted(per.items(), key=lambda kv: -kv[1])[:5]:
+            print(f"    {mb:>8} MB  {proj}")
+
+    machine = [r for r in rows if r.get("type") == "machine"]
+    if machine:
+        samples = len(machine)
+        starved = sum(1 for r in machine if r.get("waiting", 0) > 0 and r.get("used_cpu", 0) < r.get("total_cpu", 0))
+        saturated = sum(1 for r in machine if r.get("used_cpu", 0) >= r.get("total_cpu", 0))
+        idle = sum(1 for r in machine if r.get("used_cpu", 0) == 0 and r.get("waiting", 0) == 0)
+        peak_wait = max((r.get("waiting", 0) for r in machine), default=0)
+        print(f"\n  machine: {samples} samples at 15s")
+        print(f"    saturated (all cpu in use):            {saturated:>4}  ({100 * saturated / samples:.0f}%)")
+        print(f"    idle (nothing running, nothing queued): {idle:>4}  ({100 * idle / samples:.0f}%)")
+        print(f"    STARVED (queue non-empty, cpu free):    {starved:>4}  ({100 * starved / samples:.0f}%)   <- admission refusing work it could take")
+        print(f"    deepest queue: {peak_wait}")
+
+    cache = [r for r in rows if r.get("type") == "analysis_cache"]
+    if cache:
+        last = cache[-1]
+        hits, misses = last.get("intern_hits", 0), last.get("intern_misses", 0)
+        looked_up = hits + misses
+        rate = (100 * hits / looked_up) if looked_up else 0.0
+        print(f"\n  analysis cache: {last.get('entries', 0)} analyses over {last.get('workspaces', 0)} workspace(s), {last.get('file_bytes', 0) // (1024 * 1024)}MB of files")
+        print(f"    interning: {hits} hits / {misses} misses ({rate:.1f}% hit rate), sharing factor {last.get('sharing_factor', 0)}")
+        if looked_up and rate < 5:
+            print("    NOTE: interning is paying its full cost and returning almost nothing here.")
+
+    oom = [r for r in rows if r.get("type") in ("oom_pressure", "oom_crash")]
+    if oom:
+        print(f"\n  heap events: {len(oom)}")
+        for r in oom[-3:]:
+            print(f"    {r.get('type')}: used={r.get('heap_used_mb')}MB live={r.get('heap_live_mb', '?')}MB max={r.get('heap_max_mb')}MB compiles={r.get('concurrent_compiles')}")
+
+
+def main() -> int:
+    root = pathlib.Path(sys.argv[1] if len(sys.argv) > 1 else ".")
+    found = sorted(root.rglob("metrics.jsonl"))
+    if not found:
+        print(f"no metrics.jsonl under {root}")
+        return 0
+    for path in found:
+        try:
+            summarise(path, load(path))
+        except Exception as e:  # noqa: BLE001 - a broken summary must not break the build
+            print(f"could not summarise {path}: {e!r}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

.github/workflows/build.yml

@@ -65,6 +65,36 @@ jobs:
           CI: true
         run: ./bleep-cli.sh --dev test
 
+      # Always, including on failure and timeout — a run that went wrong is the one whose telemetry
+      # is worth having. CI is also the only environment with a realistic shape: 4 cores and 16GB,
+      # where memory bounds actually bind. A developer machine cannot reproduce that.
+      - name: Summarise compile-server telemetry
+        if: always()
+        shell: bash
+        run: |
+          set -uo pipefail
+          shopt -s nullglob
+          dirs=(~/.cache/bleep/socket/*/)
+          if [ ${#dirs[@]} -eq 0 ]; then echo "no compile-server socket dirs — nothing to summarise"; exit 0; fi
+          mkdir -p bsp-diagnostics
+          for d in "${dirs[@]}"; do
+            name=$(basename "$d")
+            mkdir -p "bsp-diagnostics/$name"
+            for f in "$d"metrics.jsonl "$d"output "$d"output.1 "$d"output.2; do
+              [ -f "$f" ] && cp "$f" "bsp-diagnostics/$name/"
+            done
+          done
+          python3 .github/scripts/summarise-bsp-metrics.py bsp-diagnostics
+
+      - name: Upload compile-server telemetry
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: bsp-diagnostics
+          path: bsp-diagnostics
+          retention-days: 14
+          if-no-files-found: ignore
+
   yaml-ls-check:
     runs-on: ubuntu-latest
     steps:

bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala

@@ -78,12 +78,26 @@ class ForkCostModelTest extends AnyFunSuite with Matchers {
     prog.timeout(10.seconds).unsafeRunSync()
   }
 
-  test("macOS footprint output parses, including units other than MB") {
-    ProcessMemory.MacOs.parseSize("    phys_footprint: 542 MB") shouldBe Some(542L)
-    ProcessMemory.MacOs.parseSize("phys_footprint_peak: 1645 MB") shouldBe Some(1645L)
-    ProcessMemory.MacOs.parseSize("phys_footprint: 1.6 GB") shouldBe Some(1638L)
-    ProcessMemory.MacOs.parseSize("phys_footprint: 8192 KB") shouldBe Some(8L)
-    ProcessMemory.MacOs.parseSize("phys_footprint: not-a-number") shouldBe None
+  test("a pid that no longer exists is declined, not guessed at") {
+    // The tree sweep lists pids and then measures them, so forks exit mid-sweep as a matter of course.
+    // proc_pid_rusage answers ESRCH for those; the contract is None rather than a bogus zero.
+    val goneForever = ProcessHandle.current().pid() + 100000000L
+    ProcessMemory.system.footprintMb(goneForever) shouldBe None
+  }
+
+  test("the peak is at least the current footprint, and both are plausible") {
+    // Guards the struct offsets: ri_phys_footprint and ri_lifetime_max_phys_footprint are read from
+    // fixed positions in rusage_info_v4, so a layout that moved would show up as nonsense here.
+    val self = ProcessHandle.current().pid()
+    (ProcessMemory.system.footprintMb(self), ProcessMemory.system.peakFootprintMb(self)) match {
+      case (Some(current), Some(peak)) =>
+        current should be > 0L
+        peak should be >= current
+        peak should be < 200000L
+      case _ =>
+        // Only macOS answers both: Linux tracks no high-water Pss, and Windows answers neither.
+        ProcessMemory.system should not be ProcessMemory.MacOs
+    }
   }
 
   test("the platform reader measures a real process, or honestly declines to") {

bleep-bsp-tests/src/scala/bleep/analysis/LinkDagIntegrationTest.scala

@@ -246,6 +246,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (_, _) => IO.pure(TaskResult.Success),
         link = (lt, _) => {
           linkCalled = true
@@ -294,6 +295,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (_, _) => IO.pure(TaskResult.Success),
         link = (_, _) =>
           IO.pure(
@@ -345,6 +347,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (_, _) => IO.pure(TaskResult.Success),
         link = (_, _) => IO.pure((TaskResult.Failure("Link error", List.empty), LinkResult.Failure("Link error", List.empty))),
         discover = (_, _) => IO.pure((TaskResult.Success, List.empty)),

bleep-bsp-tests/src/scala/bleep/analysis/SourcegenDagIntegrationTest.scala

@@ -297,6 +297,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (t, _) => IO(order.add(s"compile:${t.project.value}"): Unit).as(TaskResult.Success),
         link = (_, _) => sys.error("LinkTask should not appear here"),
         discover = (_, _) => sys.error("DiscoverTask should not appear here"),
@@ -343,6 +344,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (t, _) =>
           if (t.project == scriptsProject) IO.pure(TaskResult.Failure("compile error", Nil))
           else if (t.project == target) IO(targetCompileCalled.set(true)).as(TaskResult.Success)
@@ -393,6 +395,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (t, _) =>
           if (t.project == target) IO(targetCompileCalled.set(true)).as(TaskResult.Success)
           else IO.pure(TaskResult.Success),
@@ -441,6 +444,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (t, _) =>
           if (t.project == target) IO(targetCompileCalled.set(true)).as(TaskResult.Success)
           else IO.pure(TaskResult.Success),
@@ -489,6 +493,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (_, _) => IO.pure(TaskResult.Success),
         link = (_, _) => sys.error("LinkTask should not appear here"),
         discover = (_, _) => sys.error("DiscoverTask should not appear here"),
@@ -541,6 +546,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (_, _) => IO.pure(TaskResult.Success),
         link = (_, _) => sys.error("LinkTask should not appear here"),
         discover = (_, _) => sys.error("DiscoverTask should not appear here"),
@@ -589,6 +595,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (t, _) =>
           if (t.project == target) IO(targetCompileCalled.set(true)).as(TaskResult.Success)
           else IO.pure(TaskResult.Success),
@@ -638,6 +645,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (t, _) => record(s"compile:${t.project.value}").as(TaskResult.Success),
         link = (_, _) => sys.error("LinkTask should not appear here"),
         discover = (_, _) => sys.error("DiscoverTask should not appear here"),
@@ -684,6 +692,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (_, _) => IO.raiseError(new RuntimeException("bleep-test-runner resolution returned no jars")),
         link = (_, _) => sys.error("LinkTask should not appear here"),
         discover = (_, _) => sys.error("DiscoverTask should not appear here"),
@@ -736,6 +745,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers {
 
     val executor = TaskDag.executor(
       Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (_, _) => IO.pure(TaskResult.Success),
         link = (_, _) => sys.error("LinkTask should not appear here"),
         discover = (_, _) => sys.error("DiscoverTask should not appear here"),

bleep-bsp-tests/src/scala/bleep/analysis/SymbolProcessorDagIntegrationTest.scala

@@ -106,6 +106,7 @@ class SymbolProcessorDagIntegrationTest extends AnyFunSuite with Matchers {
       eventQueue <- Queue.bounded[IO, Option[TaskDag.DagEvent]](1024)
       killSignal <- Deferred[IO, KillReason]
       handlers = Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (ct, _) => IO { timeline.add(s"compile:${ct.project.value}"); TaskResult.Success },
         link = (_, _) => sys.error("LinkTask should not appear here"),
         discover = (_, _) => sys.error("DiscoverTask should not appear here"),
@@ -138,6 +139,7 @@ class SymbolProcessorDagIntegrationTest extends AnyFunSuite with Matchers {
       eventQueue <- Queue.bounded[IO, Option[TaskDag.DagEvent]](1024)
       killSignal <- Deferred[IO, KillReason]
       handlers = Handlers(
+        mayAdmitCompile = _ => IO.pure(true),
         compile = (ct, _) => IO { compileInvoked.set(true); finishedTasks.add(ct.project.value -> TaskResult.Success); TaskResult.Success },
         link = (_, _) => sys.error("LinkTask should not appear here"),
         discover = (_, _) => sys.error("DiscoverTask should not appear here"),

bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala

@@ -30,6 +30,23 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits {
   val mediumTimeout = Span(10, Seconds)
   val parallelTimeout = Span(20, Seconds)
 
+  /** Upper bound for "cancellation short-circuited the work", used by the cancellation tests below.
+    *
+    * What those tests prove is that a run which was told to stop does NOT sit through its workload — each one starts a job that would take 30s or 60s, so
+    * returning in any fraction of that is the evidence. The bound is deliberately generous rather than tight: a tight wall-clock number does not measure the
+    * cancellation path, it measures how loaded the machine is, and these run alongside the rest of the suite.
+    *
+    * They previously used 5s, which held only because the suite was effectively serialised — a double-charged CPU reservation capped test parallelism at ~1.4x
+    * (fixed in e3ad7bf0). At the ~11x the suite now really runs at, a Windows runner took 6074ms on a path whose behaviour was correct: the kill was honoured,
+    * `terminationReason` was `Killed`, and only the clock assertion failed. Raising the bound keeps the invariant that matters and drops the one that was
+    * silently asserting "nothing else is running".
+    *
+    * 15s is picked against the evidence rather than for round-number comfort: healthy is sub-second (all eight tests here finish in ~1.6s together), the worst
+    * seen on a loaded runner is 6074ms, and the shortest workload being ruled out is 30s. So it clears the worst observation by ~2.5x and still fails long
+    * before a cancellation that did not happen at all.
+    */
+  val cancellationShortCircuitMs = 15000L
+
   def createTempDir(prefix: String): Path =
     Files.createTempDirectory(prefix)
 
@@ -83,7 +100,7 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits {
         val duration = System.currentTimeMillis() - startTime
 
         result.terminationReason shouldBe a[TestRunnerTypes.TerminationReason.Killed]
-        duration should be < 5000L
+        duration should be < cancellationShortCircuitMs
       } finally deleteRecursively(tempDir)
     }
   }
@@ -124,7 +141,7 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits {
         val duration = System.currentTimeMillis() - startTime
 
         result.terminationReason shouldBe a[TestRunnerTypes.TerminationReason.Killed]
-        duration should be < 10000L
+        duration should be < cancellationShortCircuitMs
       } finally deleteRecursively(tempDir)
     }
   }
@@ -163,7 +180,7 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits {
         val duration = System.currentTimeMillis() - startTime
 
         result.terminationReason shouldBe a[TestRunnerTypes.TerminationReason.Killed]
-        duration should be < 5000L
+        duration should be < cancellationShortCircuitMs
       } finally deleteRecursively(tempDir)
     }
   }
@@ -312,7 +329,7 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits {
         val duration = System.currentTimeMillis() - startTime
 
         result.terminationReason shouldBe a[TestRunnerTypes.TerminationReason.Killed]
-        duration should be < 5000L
+        duration should be < cancellationShortCircuitMs
       } finally deleteRecursively(tempDir)
     }
   }