fix(opal-server): memory leak — purge GitPolicyFetcher caches on scope delete + webhook task cleanup (PR2)

app-tests/git-leak/README.md

@@ -51,6 +51,33 @@ Gate-coverage matrix (what each flagship test actually does):
 | `test_boot_loads_all_scopes` | **baseline → gate (PR4)** | PASSES with the loose default target; set `BOOT_TARGET_SECONDS` low (plan: 120 @ 50) on PR4 to gate the parallel-boot fix |
 | `test_repeat_sync_rss_stays_bounded` | **RSS guard** | PASSES; an RSS-budget guard against per-sync allocation leaks (the cache *count* can't grow for any impl, so there is no count assertion — see below) |
 | `test_server_recovers_after_postgres_bounce` | **guard (PER-15065 + gap publishes)** | PASSES on this branch (which has #915); guards the in-place broadcaster reconnect and that a scope PUT *during* the outage is buffered/replayed, not dropped |
+| `test_delete_recreate_storm` | **guard (lock re-mint)** | PASSES — rapid delete/re-create of the same source serializes on the repo lock and ends with clean caches; guards 89e090be |
+| `test_randomized_churn_holds_invariants` | **guard (seeded churn)** | PASSES — seeded random put/refresh/settled-delete churn holds invariants at every settle point (replay a failure with `CHURN_SEED=<seed>`); repoints and delete-vs-inflight-sync races are deliberately excluded, both lifted when PR3's fleet purge lands |
+| `test_delete_during_hung_fetch_no_crash` | **guard (use-after-free)** | PASSES — deleting a scope whose clone is hung never crashes a worker; guards the use-after-free class 89e090be fixed |
+| `test_delete_during_hung_fetch_returns_bounded` | **gate (PR3)** | FAILS without the PR3 fetch timeout — the purge waits on the repo lock and a hung clone holds that lock indefinitely, so the DELETE never returns in bounded time |
+| `test_repoint_during_inflight_fetch_drains_old_source` | **gate (PR3, update path)** | Green half passes today — repointing while the old source's clone is hung still serves the new source; red half FAILS without PR3's update-path purge — the old source's cache entries never drain |
+| `test_multiworker_churn_drains_every_worker` | **gate (PR3, broadcast)** | FAILS without PR3's broadcast purge — cache purges are process-local, so a worker whose caches were populated by something other than the DELETE it served (e.g. the leader's watcher syncs) leaks permanently; the HIGH finding from the PR2 review, as a gate |
+| `test_warm_boot_reuses_clones` | **guard (S2)** | PASSES — a restart with intact clones must serve without re-cloning |
+| `test_corrupt_clone_recovers_without_clone_loop` | **guard (S3/T7)** | PASSES — emptying a clone's object store in place while the server holds a warm cached handle is detected as invalid and recovers through the invalid-repo branch with exactly one re-clone, no serve-500s wedge and no re-clone loop; verifies the gutted-object-store detection fix |
+| `test_orphan_clone_dir_is_reclaimed` | **gate (orphan sweep, unowned)** | FAILS — a clone dir with no live scope is never reclaimed; no orphan sweep exists yet (PR3+, currently unowned) |
+| `test_redis_wiped_boot_reclaims_clones` | **gate (orphan sweep, unowned)** | FAILS — after a scope-store wipe, on-disk clones referencing nothing are never reclaimed; same missing-sweep class as the orphan-dir gate |
+| `test_boot_with_unreachable_remotes_still_serves_healthy` | **gate (PR3) — watch this flip** | FAILS without the PR3 fetch timeout — unreachable remotes present at boot hang the preload/first-sync clones and starve the executor, so a healthy scope can't serve; boot-time cousin of the offline gate |
+| `test_shard_reconfig_still_serves_but_orphans_old_clones` | **half-gate (S5, orphan sweep)** | Green half PASSES — serving survives a `SCOPES_REPO_CLONES_SHARDS` reconfig (re-clone under new ids); red half FAILS — the old-shard dirs are orphaned until the orphan sweep lands |
+| `test_force_push_rewrite_recovers` | **characterization** | PASSES — a force-pushed (rewritten) head is picked up on refresh, pinning today's behavior (pygit2's forced default fetch refspec plus `set_target` moving the local ref) |
+| `test_deleted_branch_keeps_serving_last_head` | **characterization** | PASSES — deleting the tracked branch upstream doesn't crash anything; fetch doesn't prune, so OPAL silently keeps serving the last known head (documented, not necessarily desirable, behavior) |
+
+## Invariants
+`invariants.py` defines quiescence invariants I1-I6, checked at every `opal`-fixture
+teardown (I1-I4 and I6 via `check_invariants`; I5 lives directly in `conftest.py`,
+since it needs the worker pids captured at fixture setup): I1 no orphan clone dir on
+disk, I2 no cached repo handle with no dir on disk, I3 no `repos_last_fetched` key
+with no live scope, I4 no `repo_locks` key with no live scope, I5 the worker pid set
+is unchanged across the test (proving no crash/respawn), I6 the scopes API still
+answers 200. Two markers let a test opt out deliberately instead of by accident:
+`@pytest.mark.invariant_exempt("I1", ...)` for red-gate tests that knowingly leave a
+violation behind (named per test, not a blanket skip), and
+`@pytest.mark.allow_worker_restart` for tests that restart `opal_server` on purpose,
+which would otherwise trip I5.
 
 Notes on the guards:
 - `test_repeat_sync_rss_stays_bounded` — clone paths are keyed by the repo URL,

app-tests/git-leak/conftest.py

@@ -10,6 +10,7 @@
     list_seeded_repos,
     worker_pids,
 )
+from invariants import check_invariants
 
 
 def pytest_addoption(parser):
@@ -67,7 +68,7 @@ def stack(request, repo_count):
 
 
 @pytest.fixture()
-def opal(stack) -> OpalServerClient:
+def opal(stack, request) -> OpalServerClient:
     # The compose stack is session-scoped (one server for the whole run), but
     # scopes must not leak between tests: clone paths are keyed by repo URL, so
     # a scope left behind by one test shares a cache entry with any later test
@@ -84,8 +85,18 @@ def opal(stack) -> OpalServerClient:
     assert (
         len(worker_pids()) == 1
     ), f"expected a single-worker stack, found workers {sorted(worker_pids())}"
+    pids_before = worker_pids()
     yield stack
     stack.delete_all_scopes()
+    exempt_marker = request.node.get_closest_marker("invariant_exempt")
+    exempt = set(exempt_marker.args) if exempt_marker else set()
+    if request.node.get_closest_marker("allow_worker_restart") is None:
+        assert worker_pids() == pids_before, (
+            f"I5: worker pid-set changed during the test "
+            f"({sorted(pids_before)} -> {sorted(worker_pids())}) — a crash/"
+            f"respawn resets the caches and can fake a clean drain"
+        )
+    check_invariants(stack, exempt=exempt)
 
 
 @pytest.fixture()

app-tests/git-leak/docker-compose.yml

@@ -114,6 +114,10 @@ services:
       # (references/debug-pubsub.md §3-4) — a single worker can't tell #915's
       # in-place broadcaster reconnect from a plain worker respawn.
       UVICORN_NUM_WORKERS: "${OPAL_TEST_WORKERS:-1}"
+      # Shard-count knob for the reshard boot test (test_boot_states.py):
+      # changing SCOPES_REPO_CLONES_SHARDS between boots moves every source_id,
+      # orphaning all previous clone dirs — a real ops event worth gating.
+      OPAL_SCOPES_REPO_CLONES_SHARDS: "${OPAL_TEST_SHARDS:-1}"
       OPAL_SCOPES: "1"
       OPAL_REDIS_URL: "redis://redis:6379"
       OPAL_BROADCAST_URI: "postgres://opal:opal@postgres:5432/opal"

app-tests/git-leak/helpers.py

@@ -2,7 +2,7 @@
 import subprocess
 import time
 from pathlib import Path
-from typing import Dict, List
+from typing import Any, Dict, List
 
 import requests
 
@@ -50,7 +50,7 @@ def wait_healthy(self, timeout: int = 180) -> None:
             time.sleep(2)
         raise RuntimeError(f"opal-server not healthy in {timeout}s (last: {last})")
 
-    def stats(self, samples: int = 3, interval: float = 0.1) -> Dict[str, int]:
+    def stats(self, samples: int = 3, interval: float = 0.1) -> Dict[str, Any]:
         """Read the git-fetcher cache stats, merged across a few reads.
 
         The stack runs a single uvicorn worker (see docker-compose.yml), so the
@@ -59,15 +59,29 @@ def stats(self, samples: int = 3, interval: float = 0.1) -> Dict[str, int]:
         the ``max`` per key only smooths over a read that races an in-flight
         mutation; it is not relied on to paper over multi-worker nondeterminism
         (which the single-worker setup removes outright).
+
+        Merge semantics: numeric counts take the max across samples (peak
+        smoothing); ``pid`` and the ``*_keys`` lists are last-wins. The
+        count fields and their paired ``*_keys`` lists are therefore only
+        guaranteed mutually consistent at ``samples=1`` — which is what
+        every consistency-sensitive consumer (the invariant checker,
+        per-pid sampling) uses. Do not assert
+        ``len(stats["repos_keys"]) == stats["repos"]`` on a multi-sample
+        merge.
         """
-        merged: Dict[str, int] = {}
+        merged: Dict[str, Any] = {}
         for i in range(max(1, samples)):
             resp = requests.get(
                 f"{self.base_url}/internal/git-fetcher-cache-stats", timeout=10
             )
             resp.raise_for_status()
             for key, value in resp.json().items():
-                merged[key] = max(merged.get(key, 0), value)
+                if key != "pid" and isinstance(value, (int, float)):
+                    merged[key] = max(merged.get(key, 0), value)
+                else:
+                    # pid and the *_keys lists: last-wins (single-worker stack
+                    # makes every read hit the same worker anyway)
+                    merged[key] = value
             if i < samples - 1:
                 time.sleep(interval)
         return merged
@@ -234,7 +248,12 @@ def create_repo(self, name: str) -> None:
             return
         resp = requests.post(
             f"{self.base_url}/api/v1/user/repos",
-            json={"name": name, "private": False, "auto_init": True},
+            json={
+                "name": name,
+                "private": False,
+                "auto_init": True,
+                "default_branch": "main",
+            },
             auth=self._auth,
             timeout=10,
         )
@@ -250,6 +269,39 @@ def delete_repo(self, name: str) -> None:
             resp.raise_for_status()
 
 
+class RepoMutator:
+    """Host-side git mutations against a bed Gitea repo (force-push, branch
+    ops) — the remote-transition tests' hands.
+
+    Uses GitPython over the published host port with admin basic-auth.
+    """
+
+    def __init__(self, name: str, workdir: Path):
+        import git as gitpython
+
+        self._url = (
+            f"http://{GITEA_USER}:{GITEA_PASSWORD}@localhost:13000/"
+            f"{GITEA_USER}/{name}.git"
+        )
+        self._clone = gitpython.Repo.clone_from(self._url, str(workdir / name))
+        with self._clone.config_writer() as cw:
+            cw.set_value("user", "name", "bed-mutator")
+            cw.set_value("user", "email", "bed@test.local")
+
+    def force_push_rewrite(self, branch: str = "main") -> None:
+        self._clone.git.checkout(branch)
+        self._clone.git.commit("--amend", "--allow-empty", "-m", "rewritten history")
+        self._clone.git.push("--force", "origin", branch)
+
+    def push_new_branch(self, branch: str) -> None:
+        self._clone.git.checkout("-b", branch)
+        self._clone.git.commit("--allow-empty", "-m", f"seed {branch}")
+        self._clone.git.push("origin", branch)
+
+    def delete_remote_branch(self, branch: str) -> None:
+        self._clone.git.push("origin", f":{branch}")
+
+
 def gitea_repo_url(name: str) -> str:
     # url reachable from inside the opal_server container
     return f"{GITEA_INTERNAL_URL}/{GITEA_USER}/{name}.git"
@@ -396,6 +448,47 @@ def bounce_postgres(down_seconds: int = 5, during=None) -> None:
         compose("up", "-d", "--wait", "--no-recreate", "postgres")
 
 
+def wait_until(predicate, timeout: float, interval: float = 2.0) -> bool:
+    """Poll ``predicate()`` until truthy or ``timeout`` elapses.
+
+    Swallows transient exceptions from the predicate (a stats read
+    racing a restart is not a verdict) — only the final state decides.
+    """
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        try:
+            if predicate():
+                return True
+        except Exception:
+            pass
+        time.sleep(interval)
+    try:
+        return bool(predicate())
+    except Exception:
+        return False
+
+
+def stats_by_pid(opal, min_pids: int = 2, attempts: int = 200, interval: float = 0.05):
+    """Sample the stats endpoint repeatedly; keep the LATEST snapshot per pid.
+
+    Requests land on arbitrary workers, so repeated single-sample reads
+    eventually observe each worker. Returns {pid: latest_stats} once min_pids
+    distinct pids are seen (plus one grace sample for freshness), or after
+    `attempts` samples; the caller decides whether the count is sufficient.
+    """
+    seen = {}
+    grace_sweep_done = False
+    for _ in range(attempts):
+        snap = opal.stats(samples=1)
+        seen[snap["pid"]] = snap
+        if len(seen) >= min_pids:
+            if grace_sweep_done:
+                break
+            grace_sweep_done = True
+        time.sleep(interval)
+    return seen
+
+
 def list_seeded_repos(count: int) -> List[str]:
     return [f"policy-repo-{i:04d}" for i in range(count)]
 

app-tests/git-leak/invariants.py

@@ -0,0 +1,89 @@
+"""Quiescence invariants I1-I6 for the git-leak bed.
+
+Asserted at every test's teardown (see the ``opal`` fixture) after
+``delete_all_scopes()``, and callable mid-test. Red-gate tests that
+knowingly leave violations exempt specific IDs via
+``@pytest.mark.invariant_exempt("I1", ...)``.
+
+Deviation from the design spec: I1's "none missing" direction is NOT
+asserted — scopes clone lazily, so a just-created scope legitimately has
+no dir yet. Only the orphan direction (dir with no live scope) signals a
+leak.
+"""
+import hashlib
+
+import requests
+from helpers import compose
+
+BASE_DIR_IN_CONTAINER = "/opal/git_sources"
+
+
+def source_id(url: str, branch: str = "main", shards: int = 1) -> str:
+    """Host-side mirror of GitPolicyFetcher.source_id (sha256(url) + shard)."""
+    base = hashlib.sha256(url.encode("utf-8")).hexdigest()
+    index = hashlib.sha256(branch.encode("utf-8")).digest()[0] % shards
+    return f"{base}-{index}"
+
+
+def clone_dirs(service: str = "opal_server") -> set:
+    out = compose(
+        "exec",
+        "-T",
+        service,
+        "sh",
+        "-c",
+        f"ls -1 {BASE_DIR_IN_CONTAINER} 2>/dev/null || true",
+    ).stdout
+    return {line.strip() for line in out.splitlines() if line.strip()}
+
+
+def live_source_ids(opal, shards: int = 1) -> set:
+    resp = requests.get(f"{opal.base_url}/scopes", timeout=30)
+    resp.raise_for_status()
+    return {
+        source_id(s["policy"]["url"], s["policy"].get("branch", "main"), shards)
+        for s in resp.json()
+    }
+
+
+def check_invariants(opal, exempt=frozenset(), shards: int = 1) -> None:
+    exempt = set(exempt)
+    failures = []
+    live = live_source_ids(opal, shards)
+    disk = clone_dirs()
+    stats = opal.stats(samples=1)
+
+    if "I1" not in exempt:
+        orphans = disk - live
+        if orphans:
+            failures.append(
+                f"I1: {len(orphans)} orphan clone dir(s) on disk, e.g. {sorted(orphans)[:3]}"
+            )
+    if "I2" not in exempt:
+        cached_dirs = {k.rsplit("/", 1)[-1] for k in stats.get("repos_keys", [])}
+        stray = cached_dirs - disk
+        if stray:
+            failures.append(
+                f"I2: cached repo handle(s) with no dir on disk: {sorted(stray)[:3]}"
+            )
+    if "I3" not in exempt:
+        stray = set(stats.get("repos_last_fetched_keys", [])) - live
+        if stray:
+            failures.append(
+                f"I3: repos_last_fetched key(s) with no live scope: {sorted(stray)[:3]}"
+            )
+    if "I4" not in exempt:
+        stray = set(stats.get("repo_locks_keys", [])) - live
+        if stray:
+            failures.append(
+                f"I4: repo_locks key(s) with no live scope: {sorted(stray)[:3]}"
+            )
+    if "I6" not in exempt:
+        if requests.get(f"{opal.base_url}/scopes", timeout=10).status_code != 200:
+            failures.append("I6: scopes API not answering 200")
+
+    assert not failures, (
+        "invariant violations:\n  "
+        + "\n  ".join(failures)
+        + f"\nstats={stats}\ndisk(sample)={sorted(disk)[:10]}"
+    )

app-tests/git-leak/pytest.ini

@@ -9,3 +9,7 @@
 # marker makes that guarantee explicit — it pins the rootdir to this directory
 # when run in place, independent of the root config — while the root run (from
 # the repo root) still never descends here.
+
+markers =
+    invariant_exempt(ids): exempt named invariants (I1..I6) from the opal fixture's teardown check for this (red-gate) test
+    allow_worker_restart: this test restarts opal_server on purpose; skip the I5 pid-stability teardown check