From 75ce8c4f3a9fc0438152ce5b08ea6809e03f6ba9 Mon Sep 17 00:00:00 2001 From: ZaynJarvis Date: Mon, 3 Aug 2026 14:18:54 +0000 Subject: [PATCH] fix(session): bound _wait_for_previous_archive_done to prevent permanent SessionCommit queue stall Phase 2 of a session commit waits for all earlier archives of the same session to reach a terminal state before starting. Since #3380 that wait is an unbounded while-True poll loop: if an earlier archive is orphaned (Phase 1 finished with status=ready but its queue item was lost to a crash, upgrade, or historical enqueue bug), the wait never returns. Each stuck commit permanently occupies one worker semaphore slot before tracker.start() is ever reached, so once max_concurrent such commits accumulate the whole session_commit queue freezes: everything pending, nothing running, zero errors, and restarts do not help. Bound the wait with a timeout (default 30 min, matching the existing _PHASE2_QUEUE_WAIT_TIMEOUT_SECONDS). On expiry, mark the still-pending predecessors terminally failed via the existing .failed.json marker (so their raw messages are replayed into a later commit by the established _prepare_phase2_archive_messages roll-forward) and raise TimeoutError so the current task is evicted to failed state through the normal failure path: .failed.json for the current archive, tracker.fail, queue ack, and a released worker slot. No data is silently dropped and the next commit proceeds normally. Co-Authored-By: Claude Fable 5 --- openviking/session/session.py | 53 ++++++++++++++++++- .../test_session_retention_integration.py | 39 ++++++++++++++ 2 files changed, 90 insertions(+), 2 deletions(-) diff --git a/openviking/session/session.py b/openviking/session/session.py index bc6e6a77c1..ea5882433e 100644 --- a/openviking/session/session.py +++ b/openviking/session/session.py @@ -9,6 +9,7 @@ import inspect import json import re +import time from dataclasses import dataclass, field from datetime import datetime, timezone from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Literal, Optional @@ -68,6 +69,7 @@ logger = get_logger(__name__) _ARCHIVE_WAIT_POLL_SECONDS = 0.1 +_ARCHIVE_WAIT_TIMEOUT_SECONDS = 1800.0 _PHASE2_QUEUE_WAIT_TIMEOUT_SECONDS = 1800.0 _MEMORY_EXTRACTION_MAX_RETRIES = 3 _MEMORY_EXTRACTION_RETRY_BASE_DELAY_SECONDS = 1.0 @@ -3743,11 +3745,32 @@ def _archive_index_from_uri(archive_uri: str) -> int: raise ValueError(f"Invalid archive URI: {archive_uri}") return int(match.group(1)) - async def _wait_for_previous_archive_done(self, archive_index: int) -> bool: - """Wait until every earlier archive reaches a terminal state.""" + async def _wait_for_previous_archive_done( + self, + archive_index: int, + timeout: float = _ARCHIVE_WAIT_TIMEOUT_SECONDS, + ) -> bool: + """Wait until every earlier archive reaches a terminal state. + + The wait is bounded: an earlier archive can stay pending forever when + its Phase 1 completed (``phase1.status=ready``) but the matching queue + item was lost (process crash, upgrade, or a historical enqueue bug). + Such an orphan is invisible from here, so after ``timeout`` seconds of + polling the still-pending predecessors are marked terminally failed and + a ``TimeoutError`` is raised. The caller's failure path then writes the + current archive's ``.failed.json`` and fails the task, so the worker + slot is released instead of being wedged forever; the next commit + replays the raw messages of every failed archive through the existing + ``_prepare_phase2_archive_messages`` roll-forward, so no data is lost. + + Raises: + TimeoutError: earlier archives were still pending after ``timeout`` + seconds; they have been marked failed for later raw replay. + """ if archive_index <= 1 or not self._viking_fs: return True + deadline = time.monotonic() + timeout while True: earlier_states = [ state for state in await self._scan_archive_states() if state.index < archive_index @@ -3775,6 +3798,32 @@ async def _wait_for_previous_archive_done(self, archive_index: int) -> bool: reconciled = True if reconciled: continue + if time.monotonic() >= deadline: + stuck_ids = [state.archive_id for state in pending_states] + for state in pending_states: + logger.error( + "Archive %s stayed pending for over %.0fs while blocking " + "archive_%03d Phase 2; its queue item is presumed lost. " + "Marking it failed so its raw messages replay into a " + "later commit.", + state.archive_id, + timeout, + archive_index, + ) + await self._write_failed_marker( + state.archive_uri, + stage="phase2_wait_timeout", + error=( + f"Archive stayed pending for over {timeout:.0f}s; " + "its Phase 2 queue item is presumed lost" + ), + ) + raise TimeoutError( + f"Timed out after {timeout:.0f}s waiting for earlier archives " + f"{stuck_ids} to finish before archive_{archive_index:03d} " + "Phase 2; they were marked failed for raw replay by a later " + "commit" + ) await asyncio.sleep(_ARCHIVE_WAIT_POLL_SECONDS) async def _prepare_phase2_archive_messages( diff --git a/tests/session/test_session_retention_integration.py b/tests/session/test_session_retention_integration.py index 2660a8f60a..63dbe25487 100644 --- a/tests/session/test_session_retention_integration.py +++ b/tests/session/test_session_retention_integration.py @@ -357,6 +357,45 @@ async def test_phase2_waits_for_all_earlier_pending_archives( assert await asyncio.wait_for(waiter, timeout=0.5) +async def test_phase2_wait_times_out_and_fails_orphaned_pending_archive( + client: AsyncOpenViking, + monkeypatch, +): + """A pending predecessor whose queue item was lost must not stall forever. + + After the bounded wait expires, the orphan is marked terminally failed for + raw replay and the current commit fails cleanly instead of wedging its + worker slot (see #3396). + """ + session = client.session(session_id="phase2_wait_timeout_orphan_test") + await session.ensure_exists() + first_uri = await _write_archive( + session, + 1, + [_text_message("u1", "user", "orphaned pending one")], + ) + monkeypatch.setattr("openviking.session.session._ARCHIVE_WAIT_POLL_SECONDS", 0.01) + + with pytest.raises(TimeoutError, match="archive_001"): + await asyncio.wait_for( + session._wait_for_previous_archive_done(2, timeout=0.05), + timeout=2.0, + ) + + failed = json.loads( + await session._viking_fs.read_file(f"{first_uri}/.failed.json", ctx=session.ctx) + ) + assert failed["stage"] == "phase2_wait_timeout" + states = {state.archive_id: state.state for state in await session._scan_archive_states()} + assert states["archive_001"] == "failed" + + # The next commit's wait now sees only terminal predecessors and proceeds. + assert await asyncio.wait_for( + session._wait_for_previous_archive_done(2, timeout=0.05), + timeout=2.0, + ) + + async def test_missing_previous_archive_directory_does_not_block_phase2( client: AsyncOpenViking, ):