diff --git a/CHANGELOG.md b/CHANGELOG.md index ade9f73..9254d1e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- The Wiki.js change listener now survives a database restart. It held a single + `LISTEN` connection with no supervision, so when the connection dropped it was + gone permanently while `running` stayed `True` — the service kept reporting + healthy and silently stopped indexing every page edit until someone restarted + the container. This happened for real on 2026-08-08 when `postgres-shared` was + redeployed. It now detects the drop via asyncpg's termination callback and + reconnects with bounded exponential backoff (1s doubling to a 60s cap), + retrying indefinitely because a database in maintenance does come back and + giving up would recreate the same silent deafness. +- `WikiChangeListener.running` is derived from the live connection instead of + being assigned once at startup, so it can no longer claim a subscription that + does not exist. + +### Added + +- `/health` reports the change listener under `services.wiki_listener` + (`subscribed`, `reconnects`, `last_gap_seconds`). Nothing previously exposed + its state anywhere, which is why a dead listener went unnoticed. It is + deliberately excluded from the overall healthy/degraded verdict: it recovers on + its own, and flipping the container unhealthy for the duration of a database + outage would add a restart loop to an incident rather than information. +- On reconnect the listener logs the outage duration and warns that `NOTIFY` + events emitted during the gap were lost and cannot be replayed, pointing at + `POST /maintenance/integrity-check` to reconcile. A reconciliation pass is not + performed automatically — tenant attribution for a changed page is non-trivial + here, and guessing it wrong writes content into the wrong user's namespace. + ## [1.9.0] - 2026-07-20 ### Security diff --git a/src/main.py b/src/main.py index f25a90c..67a1826 100644 --- a/src/main.py +++ b/src/main.py @@ -123,10 +123,19 @@ async def health(settings: Settings = Depends(get_settings)) -> HealthResponse: # Check service connectivity service_health = await check_service_health() - # Overall status is healthy if at least Neo4j and Qdrant are up + # Overall status is healthy if at least Neo4j and Qdrant are up. + # + # The Wiki.js change listener is reported below but deliberately excluded + # from this decision. It supervises and reconnects itself, and a database + # restart would otherwise flip the container unhealthy for the duration of + # an outage it is already recovering from. It is reported so the state is + # observable at all — previously nothing anywhere exposed it, which is how a + # dead listener went unnoticed while this endpoint answered "healthy". all_healthy = service_health.get("neo4j", False) and service_health.get("qdrant", False) overall_status = "healthy" if all_healthy else "degraded" + wiki_listener = getattr(app.state, "wiki_listener", None) + return HealthResponse( status=overall_status, app_name=settings.app_name, @@ -152,6 +161,11 @@ async def health(settings: Settings = Depends(get_settings)) -> HealthResponse: "url": settings.ollama_url, "model": settings.ollama_llm_model, "healthy": service_health.get("ollama", False) + }, + "wiki_listener": { + "subscribed": bool(wiki_listener and wiki_listener.running), + "reconnects": getattr(wiki_listener, "reconnects", 0), + "last_gap_seconds": getattr(wiki_listener, "last_gap_seconds", None) } } ) diff --git a/src/services/wiki_change_listener.py b/src/services/wiki_change_listener.py index 85a34ba..2382712 100644 --- a/src/services/wiki_change_listener.py +++ b/src/services/wiki_change_listener.py @@ -27,21 +27,59 @@ class WikiChangeListener: NOTIFY events on INSERT/UPDATE/DELETE to the pages table. """ + CHANNEL = 'wiki_page_changes' + + # Bounded exponential backoff between reconnect attempts. + BACKOFF_INITIAL_SECONDS = 1.0 + BACKOFF_MAX_SECONDS = 60.0 + def __init__(self): self.settings = get_settings() self.connection: Optional[asyncpg.Connection] = None - self.running = False # Loop prevention: Track recently processed pages # Key: page_id, Value: timestamp of last processing self._recent_notifications = {} self._debounce_seconds = self.settings.wikijs_change_listener_debounce_seconds - async def start(self): - """Start listening to database changes.""" - logger.info("Starting Wiki.js database change listener") + # Supervision state. A single LISTEN connection does not heal itself the + # way an asyncpg pool does, so the drop has to be detected and repaired + # explicitly — see _supervise(). + self._stopping = False + self._disconnected = asyncio.Event() + self._supervisor_task: Optional[asyncio.Task] = None + self._disconnected_at: Optional[datetime] = None + self.reconnects = 0 + self.last_gap_seconds: Optional[float] = None - # Connect to Wiki.js PostgreSQL database + @property + def running(self) -> bool: + """ + Whether a live subscription actually exists. + + Derived rather than assigned. The previous implementation set a flag once + in start() and never revisited it, so after the connection dropped the + listener reported itself as running while being deaf to every event. + """ + return ( + not self._stopping + and self.connection is not None + and not self.connection.is_closed() + ) + + async def start(self): + """Start listening to database changes, and keep listening.""" + logger.info("Starting Wiki.js database change listener") + self._stopping = False + self._disconnected.clear() + + await self._connect() + + self._supervisor_task = asyncio.create_task(self._supervise()) + logger.info("Listening for Wiki.js page changes via PostgreSQL NOTIFY") + + async def _connect(self): + """Open a connection and subscribe. Raises if the database is unreachable.""" self.connection = await asyncpg.connect( host=self.settings.wikijs_db_host, port=self.settings.wikijs_db_port, @@ -50,18 +88,136 @@ class WikiChangeListener: database=self.settings.wikijs_db_name ) - # Listen to the wiki_page_changes channel - await self.connection.add_listener('wiki_page_changes', self._handle_notification) + await self.connection.add_listener(self.CHANNEL, self._handle_notification) - self.running = True - logger.info("Listening for Wiki.js page changes via PostgreSQL NOTIFY") + # Must be re-registered on every connection: asyncpg clears its + # termination listeners as soon as it fires them, so this is one-shot. + self.connection.add_termination_listener(self._on_connection_lost) + + def _on_connection_lost(self, connection): + """ + Called by asyncpg when the connection terminates. + + Dispatched through loop.call_soon, so it must stay synchronous — the work + of reconnecting belongs to _supervise(), which this only wakes. + """ + if self._stopping: + return + + self._disconnected_at = datetime.now() + logger.error( + "Wiki.js change listener lost its database connection — " + "page changes are NOT being processed until it reconnects" + ) + self._disconnected.set() + + async def _supervise(self, max_iterations: Optional[int] = None) -> int: + """ + Reconnect whenever the subscription drops. + + Args: + max_iterations: Stop after N reconnect cycles (None = run forever; + used by tests) + + Returns: + Number of completed reconnect cycles + """ + iterations = 0 + while max_iterations is None or iterations < max_iterations: + await self._disconnected.wait() + if self._stopping: + break + + self._disconnected.clear() + await self._reconnect_with_backoff() + iterations += 1 + + return iterations + + async def _reconnect_with_backoff(self, max_attempts: Optional[int] = None) -> bool: + """ + Re-establish the subscription, backing off between failures. + + Keeps trying indefinitely by default: a database that is down for + maintenance will come back, and giving up would recreate exactly the + silent-deafness this supervision exists to prevent. + """ + delay = self.BACKOFF_INITIAL_SECONDS + attempts = 0 + + while not self._stopping and (max_attempts is None or attempts < max_attempts): + attempts += 1 + await self._close_connection() + + try: + await self._connect() + except Exception as e: + logger.warning( + f"Wiki.js change listener reconnect attempt {attempts} failed: {e}; " + f"retrying in {delay:.0f}s" + ) + await asyncio.sleep(delay) + delay = min(delay * 2, self.BACKOFF_MAX_SECONDS) + continue + + self.reconnects += 1 + gap = None + if self._disconnected_at is not None: + gap = (datetime.now() - self._disconnected_at).total_seconds() + self.last_gap_seconds = gap + self._disconnected_at = None + + # NOTIFY is fire-and-forget: anything emitted while we were gone was + # delivered to nobody and cannot be replayed. Say so, and say what + # closes the gap, rather than reporting a clean recovery. + outage = f" after {gap:.0f}s" if gap is not None else "" + logger.warning( + f"Wiki.js change listener reconnected{outage} " + f"(reconnect #{self.reconnects}). NOTIFY events emitted during the " + f"outage were lost and cannot be replayed — run " + f"POST /maintenance/integrity-check to reconcile pages that " + f"changed while the listener was down." + ) + return True + + return False + + async def _close_connection(self): + """Drop the current connection, tolerating one that is already dead.""" + if not self.connection: + return + + try: + if not self.connection.is_closed(): + await self.connection.remove_listener( + self.CHANNEL, self._handle_notification + ) + await self.connection.close() + except Exception as e: + # A terminated connection raises on both calls; that is expected here. + logger.debug(f"Error closing Wiki.js listener connection: {e}") + finally: + self.connection = None async def stop(self): """Stop listening and close connection.""" - if self.connection: - await self.connection.remove_listener('wiki_page_changes', self._handle_notification) - await self.connection.close() - self.running = False + self._stopping = True + + # Wake the supervisor so it observes _stopping and exits rather than + # racing us to reconnect the connection we are about to close. + self._disconnected.set() + + if self._supervisor_task: + self._supervisor_task.cancel() + try: + await self._supervisor_task + except asyncio.CancelledError: + pass + except Exception as e: + logger.debug(f"Wiki.js listener supervisor ended with: {e}") + self._supervisor_task = None + + await self._close_connection() logger.info("Stopped Wiki.js change listener") async def _handle_notification(self, connection, pid, channel, payload): diff --git a/tests/test_wiki_change_listener.py b/tests/test_wiki_change_listener.py index c8c62b0..669f062 100644 --- a/tests/test_wiki_change_listener.py +++ b/tests/test_wiki_change_listener.py @@ -308,6 +308,10 @@ class TestWikiChangeListener: async def test_connection_lifecycle(self, listener, mock_settings): """Test listener connection start and stop lifecycle.""" mock_connection = AsyncMock() + # asyncpg's is_closed() is synchronous. Left as an AsyncMock it returns a + # coroutine, which is always truthy, so the connection would read as + # closed the moment `running` started deriving from it. + mock_connection.is_closed = MagicMock(return_value=False) with patch('src.services.wiki_change_listener.asyncpg.connect', return_value=mock_connection) as mock_connect: # Start listener @@ -426,3 +430,168 @@ class TestWikiChangeListenerIntegration: # This would test actual pg_notify() calls from triggers # and verify the listener receives and processes them + + +class TestWikiChangeListenerReconnect: + """ + Supervision and reconnect behaviour. + + Regression cover for 2026-08-08: redeploying postgres-shared dropped the + listener's connection and it never came back. The container kept reporting + healthy because nothing observed the subscription, so Wiki.js page edits + silently stopped being indexed until someone restarted the service. + """ + + @pytest.mark.asyncio + async def test_running_is_false_when_connection_closed(self, listener): + """running reflects the live connection, not a flag set once at startup.""" + connection = MagicMock() + connection.is_closed.return_value = False + listener.connection = connection + assert listener.running is True + + # The exact failure mode: connection dies, nothing reassigns a flag. + connection.is_closed.return_value = True + assert listener.running is False + + @pytest.mark.asyncio + async def test_running_is_false_while_stopping(self, listener): + """A listener being torn down does not advertise itself as subscribed.""" + connection = MagicMock() + connection.is_closed.return_value = False + listener.connection = connection + listener._stopping = True + assert listener.running is False + + @pytest.mark.asyncio + async def test_termination_callback_wakes_supervisor(self, listener): + """asyncpg's termination callback signals the supervisor and records the time.""" + assert not listener._disconnected.is_set() + + listener._on_connection_lost(MagicMock()) + + assert listener._disconnected.is_set() + assert listener._disconnected_at is not None + + @pytest.mark.asyncio + async def test_termination_callback_ignored_while_stopping(self, listener): + """A deliberate shutdown must not trigger a reconnect.""" + listener._stopping = True + + listener._on_connection_lost(MagicMock()) + + assert not listener._disconnected.is_set() + + @pytest.mark.asyncio + async def test_supervisor_reconnects_after_drop(self, listener): + """One drop produces one reconnect cycle.""" + connection = MagicMock() + connection.is_closed.return_value = True + listener.connection = connection + + with patch.object(listener, '_connect', new=AsyncMock()) as connect: + listener._disconnected.set() + cycles = await listener._supervise(max_iterations=1) + + assert cycles == 1 + connect.assert_awaited_once() + assert listener.reconnects == 1 + + @pytest.mark.asyncio + async def test_supervisor_exits_without_reconnecting_when_stopping(self, listener): + """stop() wakes the supervisor to exit, not to re-establish the connection.""" + with patch.object(listener, '_connect', new=AsyncMock()) as connect: + listener._stopping = True + listener._disconnected.set() + cycles = await listener._supervise(max_iterations=1) + + assert cycles == 0 + connect.assert_not_awaited() + + @pytest.mark.asyncio + async def test_reconnect_backs_off_and_retries(self, listener): + """A database that is still down is retried, with growing delay.""" + attempts = [] + + async def fail_twice_then_succeed(): + attempts.append(1) + if len(attempts) < 3: + raise OSError("connection refused") + + sleeps = [] + + async def fake_sleep(seconds): + sleeps.append(seconds) + + with patch.object(listener, '_connect', new=AsyncMock(side_effect=fail_twice_then_succeed)), \ + patch.object(listener, '_close_connection', new=AsyncMock()), \ + patch('asyncio.sleep', new=fake_sleep): + ok = await listener._reconnect_with_backoff() + + assert ok is True + assert len(attempts) == 3 + assert sleeps == [1.0, 2.0] # doubling + assert listener.reconnects == 1 + + @pytest.mark.asyncio + async def test_reconnect_backoff_is_capped(self, listener): + """Backoff does not grow without bound during a long outage.""" + sleeps = [] + + async def fake_sleep(seconds): + sleeps.append(seconds) + + with patch.object(listener, '_connect', new=AsyncMock(side_effect=OSError("down"))), \ + patch.object(listener, '_close_connection', new=AsyncMock()), \ + patch('asyncio.sleep', new=fake_sleep): + ok = await listener._reconnect_with_backoff(max_attempts=12) + + assert ok is False + assert max(sleeps) == listener.BACKOFF_MAX_SECONDS + assert listener.reconnects == 0 + + @pytest.mark.asyncio + async def test_reconnect_records_outage_duration(self, listener): + """The gap is measured so the lost-notification window is reportable.""" + listener._disconnected_at = datetime.now() - timedelta(seconds=30) + + with patch.object(listener, '_connect', new=AsyncMock()), \ + patch.object(listener, '_close_connection', new=AsyncMock()): + await listener._reconnect_with_backoff() + + assert listener.last_gap_seconds is not None + assert 29 <= listener.last_gap_seconds <= 32 + assert listener._disconnected_at is None + + @pytest.mark.asyncio + async def test_close_connection_tolerates_dead_connection(self, listener): + """Cleaning up an already-terminated connection must not raise.""" + connection = MagicMock() + connection.is_closed.return_value = False + connection.remove_listener = AsyncMock(side_effect=Exception("connection is closed")) + connection.close = AsyncMock() + listener.connection = connection + + await listener._close_connection() + + assert listener.connection is None + + @pytest.mark.asyncio + async def test_start_registers_termination_listener(self, listener): + """ + The termination listener is re-registered on every connection. + + asyncpg clears its termination listeners as soon as it fires them, so a + registration that happened only once would survive exactly one drop. + """ + connection = MagicMock() + connection.add_listener = AsyncMock() + connection.is_closed.return_value = False + + with patch('asyncpg.connect', new=AsyncMock(return_value=connection)): + await listener._connect() + + connection.add_listener.assert_awaited_once() + connection.add_termination_listener.assert_called_once_with( + listener._on_connection_lost + )