From fdf003c889f6b3692b505972a97ceffc9ab4876c Mon Sep 17 00:00:00 2001 From: "j.a." Date: Mon, 16 Feb 2026 04:01:45 +0100 Subject: [PATCH 1/2] fix: clean up partial dirs and broken symlinks on git clone/fetch failure (#634) When clone_repository or remote.fetch fails (e.g. GitHub is unreachable), the partially created repo directory and any broken symlinks are now properly cleaned up via _cleanup_repo_path(). The class-level repos cache is also invalidated to prevent subsequent operations from using a broken Repository object. Changes: - Add _cleanup_repo_path() helper for safe directory/symlink removal - Add _invalidate_repo_cache() to evict broken repos from class caches - _clone(): clean up repo_path and cache on pygit2.GitError - fetch_and_notify_on_changes(): catch fetch GitError, clean up, return - Invalid repo discovery path now also invalidates cache before cleanup - Add comprehensive tests for all cleanup and cache invalidation paths Closes #634 --- .../opal-server/opal_server/git_fetcher.py | 44 ++++- .../tests/test_git_fetcher_cleanup.py | 168 ++++++++++++++++++ 2 files changed, 207 insertions(+), 5 deletions(-) create mode 100644 packages/opal-server/tests/test_git_fetcher_cleanup.py diff --git a/packages/opal-server/opal_server/git_fetcher.py b/packages/opal-server/opal_server/git_fetcher.py index 67e1016e9..0af0200ac 100644 --- a/packages/opal-server/opal_server/git_fetcher.py +++ b/packages/opal-server/opal_server/git_fetcher.py @@ -190,10 +190,21 @@ async def fetch_and_notify_on_changes( GitPolicyFetcher.repos_last_fetched[ self.source_id ] = datetime.datetime.now() - await run_sync( - repo.remotes[self._remote].fetch, - callbacks=self._auth_callbacks, - ) + try: + await run_sync( + repo.remotes[self._remote].fetch, + callbacks=self._auth_callbacks, + ) + except pygit2.GitError: + logger.exception( + "Failed to fetch remote {remote} for {url}, " + "cleaning up invalid repo", + remote=self._remote, + url=self._source.url, + ) + self._invalidate_repo_cache() + self._cleanup_repo_path(self._repo_path) + return logger.debug(f"Fetch completed: {self._source.url}") # New commits might be present because of a previous fetch made by another scope @@ -204,7 +215,8 @@ async def fetch_and_notify_on_changes( logger.warning( "Deleting invalid repo: {path}", path=self._repo_path ) - shutil.rmtree(self._repo_path) + self._invalidate_repo_cache() + self._cleanup_repo_path(self._repo_path) else: logger.info("Repo not found at {path}", path=self._repo_path) @@ -215,6 +227,26 @@ def _discover_repository(self, path: Path) -> bool: git_path: Path = path / ".git" return discover_repository(str(path)) and git_path.exists() + @staticmethod + def _cleanup_repo_path(repo_path: Path): + """Safely remove a repository path, handling broken symlinks and + partial directories left behind by failed operations.""" + path = Path(repo_path) + if path.is_symlink() or path.exists(): + logger.info( + "Cleaning up repo path: {path}", + path=repo_path, + ) + shutil.rmtree(str(repo_path), ignore_errors=True) + + def _invalidate_repo_cache(self): + """Remove this repo from the class-level caches if present.""" + path = str(self._repo_path) + GitPolicyFetcher.repos.pop(path, None) + GitPolicyFetcher.repos_last_fetched.pop( + GitPolicyFetcher.source_id(self._source), None + ) + async def _clone(self): logger.info( "Cloning repo at '{url}' to '{path}'", @@ -230,6 +262,8 @@ async def _clone(self): ) except pygit2.GitError: logger.exception(f"Could not clone repo at {self._source.url}") + self._cleanup_repo_path(self._repo_path) + self._invalidate_repo_cache() else: logger.info(f"Clone completed: {self._source.url}") await self._notify_on_changes(repo) diff --git a/packages/opal-server/tests/test_git_fetcher_cleanup.py b/packages/opal-server/tests/test_git_fetcher_cleanup.py new file mode 100644 index 000000000..2ae82e974 --- /dev/null +++ b/packages/opal-server/tests/test_git_fetcher_cleanup.py @@ -0,0 +1,168 @@ +"""Tests for git_fetcher cleanup and cache invalidation (issue #634). + +Verifies that: +- Failed clone operations clean up partial directories and broken symlinks. +- Fetch failures (e.g. GitHub down) are handled gracefully with cleanup. +- Invalid repos are removed from the class-level cache. +""" + +import asyncio +import shutil +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pygit2 +import pytest + +from opal_server.git_fetcher import GitPolicyFetcher + + +def _make_fetcher(tmp_path: Path) -> GitPolicyFetcher: + """Create a GitPolicyFetcher with a minimal mock source.""" + source = MagicMock() + source.url = "https://github.com/example/repo.git" + source.branch = "main" + source.auth = None + + fetcher = GitPolicyFetcher( + base_dir=tmp_path, + scope_id="test-scope", + source=source, + callbacks=MagicMock(), + ) + return fetcher + + +class TestCloneCleanup: + """_clone() must clean up self._repo_path when clone_repository fails.""" + + @pytest.mark.asyncio + async def test_clone_failure_removes_partial_directory(self, tmp_path): + fetcher = _make_fetcher(tmp_path) + + # Simulate a partial directory left by a failed clone + fetcher._repo_path.mkdir(parents=True, exist_ok=True) + (fetcher._repo_path / "partial_file").touch() + + with patch( + "opal_server.git_fetcher.run_sync", + side_effect=pygit2.GitError("connection refused"), + ): + await fetcher._clone() + + assert not fetcher._repo_path.exists(), ( + "Partial repo directory should be removed after clone failure" + ) + + @pytest.mark.asyncio + async def test_clone_failure_removes_broken_symlink(self, tmp_path): + fetcher = _make_fetcher(tmp_path) + + # Create a broken symlink at repo_path + target = tmp_path / "nonexistent_target" + fetcher._repo_path.symlink_to(target) + assert fetcher._repo_path.is_symlink() + + with patch( + "opal_server.git_fetcher.run_sync", + side_effect=pygit2.GitError("connection refused"), + ): + await fetcher._clone() + + assert not fetcher._repo_path.is_symlink(), ( + "Broken symlink should be removed after clone failure" + ) + + @pytest.mark.asyncio + async def test_clone_failure_invalidates_cache(self, tmp_path): + fetcher = _make_fetcher(tmp_path) + path_key = str(fetcher._repo_path) + source_id = GitPolicyFetcher.source_id(fetcher._source) + + # Pre-populate the cache + GitPolicyFetcher.repos[path_key] = MagicMock() + GitPolicyFetcher.repos_last_fetched[source_id] = "some-time" + + with patch( + "opal_server.git_fetcher.run_sync", + side_effect=pygit2.GitError("connection refused"), + ): + await fetcher._clone() + + assert path_key not in GitPolicyFetcher.repos + assert source_id not in GitPolicyFetcher.repos_last_fetched + + +class TestFetchFailureCleanup: + """fetch_and_notify_on_changes() must handle fetch errors gracefully.""" + + @pytest.mark.asyncio + async def test_fetch_failure_cleans_up_and_invalidates(self, tmp_path): + fetcher = _make_fetcher(tmp_path) + path_key = str(fetcher._repo_path) + + # Create a fake repo directory with .git so _discover_repository passes + (fetcher._repo_path / ".git").mkdir(parents=True, exist_ok=True) + + mock_repo = MagicMock() + mock_remote = MagicMock() + mock_remote.fetch.side_effect = pygit2.GitError("GitHub is down") + mock_repo.remotes = {fetcher._remote: mock_remote} + + GitPolicyFetcher.repos[path_key] = mock_repo + + with ( + patch.object(fetcher, "_discover_repository", return_value=True), + patch.object(fetcher, "_get_valid_repo", return_value=mock_repo), + patch.object(fetcher, "_should_fetch", return_value=True), + patch( + "opal_server.git_fetcher.run_sync", + side_effect=pygit2.GitError("GitHub is down"), + ), + ): + # Should not raise + await fetcher.fetch_and_notify_on_changes(force_fetch=True) + + assert path_key not in GitPolicyFetcher.repos, ( + "Failed repo should be removed from cache" + ) + assert not fetcher._repo_path.exists(), ( + "Repo directory should be cleaned up after fetch failure" + ) + + +class TestCacheInvalidation: + """Invalid repos must be evicted from GitPolicyFetcher.repos.""" + + @pytest.mark.asyncio + async def test_invalid_repo_removed_from_cache_on_discovery(self, tmp_path): + fetcher = _make_fetcher(tmp_path) + path_key = str(fetcher._repo_path) + + # Simulate: directory exists but repo is invalid + (fetcher._repo_path / ".git").mkdir(parents=True, exist_ok=True) + GitPolicyFetcher.repos[path_key] = MagicMock() + + mock_clone_repo = MagicMock() + + with ( + patch.object(fetcher, "_discover_repository", return_value=True), + patch.object(fetcher, "_get_valid_repo", return_value=None), + patch.object(fetcher, "_clone", new_callable=AsyncMock) as mock_clone, + ): + await fetcher.fetch_and_notify_on_changes() + mock_clone.assert_called_once() + + assert path_key not in GitPolicyFetcher.repos, ( + "Invalid repo should be evicted from cache" + ) + + def test_cleanup_repo_path_is_idempotent(self, tmp_path): + """Calling _cleanup_repo_path on a non-existent path should not raise.""" + nonexistent = tmp_path / "does_not_exist" + GitPolicyFetcher._cleanup_repo_path(nonexistent) # should not raise + + def test_invalidate_repo_cache_is_safe_when_empty(self, tmp_path): + """_invalidate_repo_cache should not raise even if caches are empty.""" + fetcher = _make_fetcher(tmp_path) + fetcher._invalidate_repo_cache() # should not raise From 775c7831c4aacd40c86ce88a6ff9809cac7fb32a Mon Sep 17 00:00:00 2001 From: "j.a." Date: Mon, 16 Feb 2026 08:55:47 +0100 Subject: [PATCH 2/2] fix: log broadcast URI hostname resolution errors at ERROR level (closes #716) --- packages/opal-server/opal_server/pubsub.py | 42 +++++++++- packages/opal-server/opal_server/server.py | 17 +++- .../tests/test_broadcast_uri_validation.py | 77 +++++++++++++++++++ 3 files changed, 134 insertions(+), 2 deletions(-) create mode 100644 packages/opal-server/tests/test_broadcast_uri_validation.py diff --git a/packages/opal-server/opal_server/pubsub.py b/packages/opal-server/opal_server/pubsub.py index c7a3b875e..caf8fc414 100644 --- a/packages/opal-server/opal_server/pubsub.py +++ b/packages/opal-server/opal_server/pubsub.py @@ -1,8 +1,10 @@ +import socket import time from contextlib import contextmanager from contextvars import ContextVar from threading import Lock from typing import Dict, Generator, List, Optional, Set, Tuple, Union, cast +from urllib.parse import urlparse from uuid import UUID, uuid4 from fastapi import APIRouter, Depends, WebSocket @@ -141,7 +143,12 @@ def __init__(self, signer: JWTSigner, broadcaster_uri: str = None): self.broadcaster = None if broadcaster_uri is not None: - logger.info(f"Initializing broadcaster for server<->server communication") + safe_uri = self._mask_uri_password(broadcaster_uri) + logger.info( + "Initializing broadcaster for server<->server communication, uri={uri}", + uri=safe_uri, + ) + self._validate_broadcast_uri(broadcaster_uri) self.broadcaster = EventBroadcaster( broadcaster_uri, notifier=self.notifier, @@ -202,6 +209,39 @@ async def websocket_rpc_endpoint( finally: await websocket.close() + @staticmethod + def _mask_uri_password(uri: str) -> str: + """Return the URI with password masked.""" + parsed = urlparse(uri) + if parsed.password: + masked = parsed._replace( + netloc=f"{parsed.username}:***@{parsed.hostname}" + + (f":{parsed.port}" if parsed.port else "") + ) + return masked.geturl() + return uri + + @staticmethod + def _validate_broadcast_uri(uri: str): + """Validate the broadcast URI hostname can be resolved. Logs ERROR on + failure but does not raise.""" + parsed = urlparse(uri) + hostname = parsed.hostname + if not hostname: + logger.error( + "Broadcast URI has no hostname: {uri}. Check your BROADCAST_URI configuration.", + uri=uri, + ) + return + try: + socket.getaddrinfo(hostname, None) + except socket.gaierror as e: + logger.error( + "Cannot resolve broadcast URI hostname '{hostname}': {error}. Check your BROADCAST_URI configuration.", + hostname=hostname, + error=str(e), + ) + @staticmethod async def _verify_permitted_topics( topics: Union[TopicList, ALL_TOPICS], channel: RpcChannel diff --git a/packages/opal-server/opal_server/server.py b/packages/opal-server/opal_server/server.py index f320484ea..23a380635 100644 --- a/packages/opal-server/opal_server/server.py +++ b/packages/opal-server/opal_server/server.py @@ -329,7 +329,7 @@ async def start_server_background_tasks(self): await self.broadcast_listening_context.__aenter__() # if the broadcast channel is closed, we want to restart worker process because statistics can't be reliable anymore self.broadcast_listening_context._event_broadcaster.get_reader_task().add_done_callback( - lambda _: self._graceful_shutdown() + self._on_broadcaster_disconnected ) asyncio.create_task(self.opal_statistics.run()) self.pubsub.endpoint.notifier.register_unsubscribe_event( @@ -398,6 +398,21 @@ async def stop_server_background_tasks(self): except Exception: logger.exception("exception while shutting down background tasks") + def _on_broadcaster_disconnected(self, task: asyncio.Task): + """Callback when the broadcast listener task completes unexpectedly.""" + try: + exc = task.exception() + except (asyncio.CancelledError, asyncio.InvalidStateError): + exc = None + if exc is not None: + logger.error( + "Broadcast channel connection failed: {error}. Check BROADCAST_URI configuration.", + error=exc, + ) + else: + logger.warning("Broadcast channel disconnected.") + self._graceful_shutdown() + def _graceful_shutdown(self): logger.info("Trigger worker graceful shutdown") os.kill(os.getpid(), signal.SIGTERM) diff --git a/packages/opal-server/tests/test_broadcast_uri_validation.py b/packages/opal-server/tests/test_broadcast_uri_validation.py new file mode 100644 index 000000000..ede6d1b24 --- /dev/null +++ b/packages/opal-server/tests/test_broadcast_uri_validation.py @@ -0,0 +1,77 @@ +"""Tests for broadcast URI validation and error logging (issue #716).""" + +import asyncio +import logging +import socket +from unittest.mock import MagicMock, patch + +import pytest + +from opal_server.pubsub import PubSub + + +class TestValidateBroadcastUri: + def test_invalid_hostname_logs_error(self, caplog): + """Unresolvable hostname should be logged at ERROR level.""" + with caplog.at_level(logging.ERROR, logger="opal.server"): + with patch( + "opal_server.pubsub.socket.getaddrinfo", + side_effect=socket.gaierror("Name or service not known"), + ): + PubSub._validate_broadcast_uri("postgres://invalid-host-xyz:5432/db") + assert any("Cannot resolve broadcast URI hostname" in r.message for r in caplog.records) + + def test_valid_hostname_no_error(self, caplog): + """Resolvable hostname should not produce ERROR logs.""" + with caplog.at_level(logging.ERROR, logger="opal.server"): + with patch("opal_server.pubsub.socket.getaddrinfo", return_value=[(None,) * 5]): + PubSub._validate_broadcast_uri("postgres://localhost:5432/db") + assert not any("Cannot resolve" in r.message for r in caplog.records) + + def test_uri_without_hostname_logs_error(self, caplog): + """URI with no hostname should be logged at ERROR level.""" + with caplog.at_level(logging.ERROR, logger="opal.server"): + PubSub._validate_broadcast_uri("postgres://") + assert any("has no hostname" in r.message for r in caplog.records) + + +class TestMaskUriPassword: + def test_password_masked(self): + result = PubSub._mask_uri_password("postgres://user:secret@host:5432/db") + assert "secret" not in result + assert "***" in result + assert "user" in result + + def test_no_password(self): + uri = "postgres://host:5432/db" + assert PubSub._mask_uri_password(uri) == uri + + +class TestOnBroadcasterDisconnected: + def test_callback_logs_exception_on_error(self, caplog): + """done_callback should log task exceptions at ERROR level.""" + from opal_server.server import OpalServer + + task = MagicMock(spec=asyncio.Task) + task.exception.return_value = ConnectionError("connection refused") + + server = object.__new__(OpalServer) + + with caplog.at_level(logging.ERROR, logger="opal.server"): + with patch.object(OpalServer, "_graceful_shutdown"): + server._on_broadcaster_disconnected(task) + + assert any("Broadcast channel connection failed" in r.message for r in caplog.records) + + def test_callback_calls_graceful_shutdown(self): + """done_callback should call _graceful_shutdown.""" + from opal_server.server import OpalServer + + task = MagicMock(spec=asyncio.Task) + task.exception.return_value = None + + server = object.__new__(OpalServer) + + with patch.object(OpalServer, "_graceful_shutdown") as mock_shutdown: + server._on_broadcaster_disconnected(task) + mock_shutdown.assert_called_once()