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

packages/opal-server/opal_server/git_fetcher.py

@@ -369,6 +369,26 @@ def base_dir(base_dir: Path) -> Path:
     def repo_clone_path(base_dir: Path, source: GitPolicyScopeSource) -> Path:
         return GitPolicyFetcher.base_dir(base_dir) / GitPolicyFetcher.source_id(source)
 
+    @staticmethod
+    def forget_repo(path: str) -> None:
+        """Drop the cached repository for a clone path and release its handles.
+
+        The cached ``pygit2.Repository`` keeps OS file descriptors and mmapped
+        pack indexes open; without this, a deleted scope's repo pins memory and
+        inodes for the lifetime of the process even after the clone is removed.
+        ``Repository.free()`` is called only when available (it is not part of
+        every pygit2 release); otherwise the dropped reference is reclaimed by GC.
+        """
+        repo = GitPolicyFetcher.repos.pop(path, None)
+        if repo is None:
+            return
+        free = getattr(repo, "free", None)
+        if callable(free):
+            try:
+                free()
+            except Exception:
+                logger.debug("pygit2 Repository.free() failed; relying on GC")
+
 
 class GitCallback(RemoteCallbacks):
     def __init__(self, source: GitPolicyScopeSource):

packages/opal-server/opal_server/policy/watcher/task.py

@@ -28,11 +28,10 @@ async def __aexit__(self, exc_type, exc, tb):
 
     async def _on_webhook(self, topic: Topic, data: Any):
         logger.info(f"Webhook listener triggered ({len(self._webhook_tasks)})")
-        for task in self._webhook_tasks:
-            if task.done():
-                # Clean references to finished tasks
-                self._webhook_tasks.remove(task)
-
+        # Rebuild rather than remove-while-iterating: list.remove() inside a
+        # `for t in self._webhook_tasks` loop skips the element after each removal,
+        # so finished tasks accumulate.
+        self._webhook_tasks = [t for t in self._webhook_tasks if not t.done()]
         self._webhook_tasks.append(asyncio.create_task(self.trigger(topic, data)))
 
     async def _listen_to_webhook_notifications(self):

packages/opal-server/opal_server/scopes/api.py

@@ -44,6 +44,7 @@
 from opal_server.data.data_update_publisher import DataUpdatePublisher
 from opal_server.git_fetcher import GitPolicyFetcher
 from opal_server.scopes.scope_repository import ScopeNotFoundError, ScopeRepository
+from opal_server.scopes.service import ScopesService
 
 
 def verify_private_key(private_key: str, key_format: EncryptionKeyFormat) -> bool:
@@ -80,6 +81,7 @@ def init_scope_router(
     scopes: ScopeRepository,
     authenticator: JWTAuthenticator,
     pubsub_endpoint: PubSubEndpoint,
+    scopes_service: ScopesService,
 ):
     router = APIRouter()
 
@@ -171,8 +173,13 @@ async def delete_scope(
             logger.error(f"Unauthorized to delete scope: {repr(ex)}")
             raise
 
-        # TODO: This should also asynchronously clean the repo from the disk (if it's not used by other scopes)
-        await scopes.delete(scope_id)
+        try:
+            # Deletes the scope and also cleans the repo clone from disk and the
+            # GitPolicyFetcher in-memory caches (unless another scope shares them).
+            await scopes_service.delete_scope(scope_id)
+        except ScopeNotFoundError:
+            # Deleting a missing scope was always a silent no-op (204); keep it.
+            pass
 
         return Response(status_code=status.HTTP_204_NO_CONTENT)
 

packages/opal-server/opal_server/scopes/service.py

@@ -163,24 +163,35 @@ async def delete_scope(self, scope_id: str):
         with tracer.trace("scopes_service.delete_scope", resource=scope_id):
             logger.info(f"Delete scope: {scope_id}")
             scope = await self._scopes.get(scope_id)
-            url = scope.policy.url
-
-            scopes = await self._scopes.all()
-            remove_repo_clone = True
-
-            for scope in scopes:
-                if scope.scope_id != scope_id and scope.policy.url == url:
-                    logger.info(
-                        f"found another scope with same remote url ({scope.scope_id}), skipping clone deletion"
-                    )
-                    remove_repo_clone = False
-                    break
+            deleted_source = cast(GitPolicyScopeSource, scope.policy)
+            deleted_source_id = GitPolicyFetcher.source_id(deleted_source)
+            scope_dir = GitPolicyFetcher.repo_clone_path(self._base_dir, deleted_source)
+
+            # Clone dir, the `repos` handle cache, and `repos_last_fetched` are
+            # all keyed by source_id (= the clone path). A sibling only shares
+            # storage when it resolves to the same source_id; same url with a
+            # different branch can shard to a different source_id (and a
+            # different clone dir) when SCOPES_REPO_CLONES_SHARDS > 1, so gate on
+            # source_id, not url — otherwise the deleted scope's clone + pygit2
+            # handle leak.
+            other_scopes = [
+                s for s in await self._scopes.all() if s.scope_id != scope_id
+            ]
+            source_id_shared = any(
+                isinstance(s.policy, GitPolicyScopeSource)
+                and GitPolicyFetcher.source_id(s.policy) == deleted_source_id
+                for s in other_scopes
+            )
 
-            if remove_repo_clone:
-                scope_dir = GitPolicyFetcher.repo_clone_path(
-                    self._base_dir, cast(GitPolicyScopeSource, scope.policy)
+            if source_id_shared:
+                logger.info(
+                    "Another scope shares the same clone (source id), skipping clone deletion"
                 )
+            else:
                 shutil.rmtree(scope_dir, ignore_errors=True)
+                GitPolicyFetcher.forget_repo(str(scope_dir))
+                GitPolicyFetcher.repos_last_fetched.pop(deleted_source_id, None)
+                GitPolicyFetcher.repo_locks.pop(deleted_source_id, None)
 
             await self._scopes.delete(scope_id)
 

packages/opal-server/opal_server/server.py

@@ -4,6 +4,7 @@
 import sys
 import traceback
 from functools import partial
+from pathlib import Path
 from typing import List, Optional
 
 from fastapi import Depends, FastAPI, Request
@@ -40,6 +41,7 @@
 from opal_server.scopes.api import init_scope_router
 from opal_server.scopes.loader import load_scopes
 from opal_server.scopes.scope_repository import ScopeRepository
+from opal_server.scopes.service import ScopesService
 from opal_server.security.api import init_security_router
 from opal_server.security.jwks import JwksStaticEndpoint
 from opal_server.statistics import OpalStatistics, init_statistics_router
@@ -270,8 +272,15 @@ def _configure_api_routes(self, app: FastAPI):
         )
 
         if opal_server_config.SCOPES:
+            scopes_service = ScopesService(
+                base_dir=Path(opal_server_config.BASE_DIR),
+                scopes=self._scopes,
+                pubsub_endpoint=self.pubsub.endpoint,
+            )
             app.include_router(
-                init_scope_router(self._scopes, authenticator, self.pubsub.endpoint),
+                init_scope_router(
+                    self._scopes, authenticator, self.pubsub.endpoint, scopes_service
+                ),
                 tags=["Scopes"],
                 prefix="/scopes",
             )

packages/opal-server/opal_server/tests/delete_scope_cache_purge_test.py

@@ -0,0 +1,136 @@
+import pytest
+from opal_common.schemas.policy_source import GitPolicyScopeSource, NoAuthData
+from opal_common.schemas.scopes import Scope
+from opal_server.git_fetcher import GitPolicyFetcher
+from opal_server.scopes.scope_repository import ScopeNotFoundError
+from opal_server.scopes.service import ScopesService
+
+
+class FakeScopeRepository:
+    def __init__(self, scopes):
+        self._scopes = {s.scope_id: s for s in scopes}
+
+    async def get(self, scope_id):
+        if scope_id not in self._scopes:
+            raise ScopeNotFoundError(scope_id)
+        return self._scopes[scope_id]
+
+    async def all(self):
+        return list(self._scopes.values())
+
+    async def delete(self, scope_id):
+        self._scopes.pop(scope_id, None)
+
+
+def _scope(scope_id, url, branch="main"):
+    return Scope(
+        scope_id=scope_id,
+        policy=GitPolicyScopeSource(
+            source_type="git",
+            url=url,
+            branch=branch,
+            auth=NoAuthData(auth_type="none"),
+        ),
+        data={"entries": []},
+    )
+
+
+@pytest.fixture(autouse=True)
+def clear_caches():
+    GitPolicyFetcher.repos.clear()
+    GitPolicyFetcher.repos_last_fetched.clear()
+    GitPolicyFetcher.repo_locks.clear()
+    yield
+    GitPolicyFetcher.repos.clear()
+    GitPolicyFetcher.repos_last_fetched.clear()
+    GitPolicyFetcher.repo_locks.clear()
+
+
+@pytest.mark.asyncio
+async def test_delete_unique_scope_purges_caches(tmp_path, monkeypatch):
+    scope = _scope("only", "https://git/repo-a.git")
+    repo = FakeScopeRepository([scope])
+    svc = ScopesService(base_dir=tmp_path, scopes=repo, pubsub_endpoint=None)
+
+    src = scope.policy
+    sid = GitPolicyFetcher.source_id(src)
+    clone_path = str(GitPolicyFetcher.repo_clone_path(tmp_path, src))
+    GitPolicyFetcher.repos[clone_path] = object()
+    GitPolicyFetcher.repos_last_fetched[sid] = "ts"
+    GitPolicyFetcher.repo_locks[sid] = object()
+
+    monkeypatch.setattr(
+        "opal_server.scopes.service.shutil.rmtree", lambda *a, **k: None
+    )
+
+    await svc.delete_scope("only")
+
+    assert clone_path not in GitPolicyFetcher.repos
+    assert sid not in GitPolicyFetcher.repos_last_fetched
+    assert sid not in GitPolicyFetcher.repo_locks
+
+
+@pytest.mark.asyncio
+async def test_delete_keeps_caches_when_sibling_shares_source(tmp_path, monkeypatch):
+    a = _scope("a", "https://git/shared.git")
+    b = _scope("b", "https://git/shared.git")  # same url+branch -> same source_id
+    repo = FakeScopeRepository([a, b])
+    svc = ScopesService(base_dir=tmp_path, scopes=repo, pubsub_endpoint=None)
+
+    sid = GitPolicyFetcher.source_id(a.policy)
+    clone_path = str(GitPolicyFetcher.repo_clone_path(tmp_path, a.policy))
+    GitPolicyFetcher.repos[clone_path] = object()
+    GitPolicyFetcher.repos_last_fetched[sid] = "ts"
+    GitPolicyFetcher.repo_locks[sid] = object()
+
+    rmtree_calls = []
+    monkeypatch.setattr(
+        "opal_server.scopes.service.shutil.rmtree",
+        lambda p, **k: rmtree_calls.append(p),
+    )
+
+    await svc.delete_scope("a")
+
+    assert rmtree_calls == []  # sibling shares the source id; clone must survive
+    assert clone_path in GitPolicyFetcher.repos
+    assert sid in GitPolicyFetcher.repos_last_fetched
+    assert sid in GitPolicyFetcher.repo_locks
+
+
+@pytest.mark.asyncio
+async def test_delete_purges_when_sibling_shares_url_but_not_source(
+    tmp_path, monkeypatch
+):
+    """Same url, different branch, sharded clones (SCOPES_REPO_CLONES_SHARDS>1)
+    resolve to different source_ids -> different clone dirs.
+
+    Deleting one must still purge its own clone + caches; the url-
+    sharing sibling lives elsewhere.
+    """
+    # shards=4: branch "main" -> index 1, "dev" -> index 3 (distinct source_ids).
+    monkeypatch.setattr(
+        "opal_server.git_fetcher.opal_server_config.SCOPES_REPO_CLONES_SHARDS", 4
+    )
+    a = _scope("a", "https://git/shared.git", branch="main")
+    b = _scope("b", "https://git/shared.git", branch="dev")  # same url, diff source_id
+    assert GitPolicyFetcher.source_id(a.policy) != GitPolicyFetcher.source_id(b.policy)
+
+    repo = FakeScopeRepository([a, b])
+    svc = ScopesService(base_dir=tmp_path, scopes=repo, pubsub_endpoint=None)
+
+    sid_a = GitPolicyFetcher.source_id(a.policy)
+    clone_path_a = str(GitPolicyFetcher.repo_clone_path(tmp_path, a.policy))
+    GitPolicyFetcher.repos[clone_path_a] = object()
+    GitPolicyFetcher.repos_last_fetched[sid_a] = "ts"
+
+    rmtree_calls = []
+    monkeypatch.setattr(
+        "opal_server.scopes.service.shutil.rmtree",
+        lambda p, **k: rmtree_calls.append(str(p)),
+    )
+
+    await svc.delete_scope("a")
+
+    assert rmtree_calls == [clone_path_a]  # its own clone dir removed
+    assert clone_path_a not in GitPolicyFetcher.repos
+    assert sid_a not in GitPolicyFetcher.repos_last_fetched

packages/opal-server/opal_server/tests/delete_scope_route_test.py

@@ -0,0 +1,97 @@
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from opal_common.schemas.policy_source import GitPolicyScopeSource, NoAuthData
+from opal_common.schemas.scopes import Scope
+from opal_server.git_fetcher import GitPolicyFetcher
+from opal_server.scopes.api import init_scope_router
+from opal_server.scopes.scope_repository import ScopeNotFoundError
+from opal_server.scopes.service import ScopesService
+
+
+class FakeScopeRepository:
+    def __init__(self, scopes):
+        self._scopes = {s.scope_id: s for s in scopes}
+
+    async def get(self, scope_id):
+        if scope_id not in self._scopes:
+            raise ScopeNotFoundError(scope_id)
+        return self._scopes[scope_id]
+
+    async def all(self):
+        return list(self._scopes.values())
+
+    async def delete(self, scope_id):
+        self._scopes.pop(scope_id, None)
+
+
+class FakeAuthenticator:
+    """Mimics a JWTAuthenticator whose verifier is disabled (no public key)."""
+
+    enabled = False
+
+    def __call__(self):
+        return {}
+
+
+def _scope(scope_id, url, branch="main"):
+    return Scope(
+        scope_id=scope_id,
+        policy=GitPolicyScopeSource(
+            source_type="git",
+            url=url,
+            branch=branch,
+            auth=NoAuthData(auth_type="none"),
+        ),
+        data={"entries": []},
+    )
+
+
+def _client(repo, base_dir):
+    service = ScopesService(base_dir=base_dir, scopes=repo, pubsub_endpoint=None)
+    app = FastAPI()
+    app.include_router(
+        init_scope_router(repo, FakeAuthenticator(), None, service),
+        prefix="/scopes",
+    )
+    return TestClient(app)
+
+
+@pytest.fixture(autouse=True)
+def clear_caches():
+    GitPolicyFetcher.repos.clear()
+    GitPolicyFetcher.repos_last_fetched.clear()
+    GitPolicyFetcher.repo_locks.clear()
+    yield
+    GitPolicyFetcher.repos.clear()
+    GitPolicyFetcher.repos_last_fetched.clear()
+    GitPolicyFetcher.repo_locks.clear()
+
+
+def test_delete_route_purges_fetcher_caches(tmp_path, monkeypatch):
+    """DELETE /scopes/{id} must flow through ScopesService.delete_scope so the
+    GitPolicyFetcher caches drain (the git-leak churn gate)."""
+    scope = _scope("only", "https://git/repo-a.git")
+    repo = FakeScopeRepository([scope])
+
+    sid = GitPolicyFetcher.source_id(scope.policy)
+    clone_path = str(GitPolicyFetcher.repo_clone_path(tmp_path, scope.policy))
+    GitPolicyFetcher.repos[clone_path] = object()
+    GitPolicyFetcher.repos_last_fetched[sid] = "ts"
+
+    monkeypatch.setattr(
+        "opal_server.scopes.service.shutil.rmtree", lambda *a, **k: None
+    )
+
+    resp = _client(repo, tmp_path).delete("/scopes/only")
+
+    assert resp.status_code == 204
+    assert clone_path not in GitPolicyFetcher.repos
+    assert sid not in GitPolicyFetcher.repos_last_fetched
+
+
+def test_delete_route_missing_scope_stays_204(tmp_path):
+    """Deleting a nonexistent scope was a silent no-op (204) before the purge
+    wiring and must remain one."""
+    resp = _client(FakeScopeRepository([]), tmp_path).delete("/scopes/ghost")
+    assert resp.status_code == 204