feat(transaction): add redo recovery toggle for session commits (#1934)

Allow crash-recovery redo for session commit phase 2 to be disabled via configuration while keeping the default behavior unchanged. This lets deployments skip pending redo marker writes and startup redo recovery when the mechanism is not wanted.

Author: zhoujh01

openviking/service/core.py

@@ -147,6 +147,7 @@ def _init_storage(
             agfs=self._agfs_client,
             lock_timeout=tx_cfg.lock_timeout,
             lock_expire=tx_cfg.lock_expire,
+            redo_recovery_enabled=tx_cfg.redo_recovery_enabled,
         )
 
     @property

openviking/session/session.py

@@ -26,10 +26,12 @@ def __init__(
         agfs: AGFSClient,
         lock_timeout: float = 0.0,
         lock_expire: float = 300.0,
+        redo_recovery_enabled: bool = True,
     ):
         self._agfs = agfs
         self._path_lock = PathLock(agfs, lock_expire=lock_expire)
         self._lock_timeout = lock_timeout
+        self._redo_recovery_enabled = redo_recovery_enabled
         self._redo_log = RedoLog(agfs)
         self._handles: Dict[str, LockHandle] = {}
         self._cleanup_task: Optional[asyncio.Task] = None
@@ -40,6 +42,10 @@ def __init__(
     def redo_log(self) -> RedoLog:
         return self._redo_log
 
+    @property
+    def redo_recovery_enabled(self) -> bool:
+        return self._redo_recovery_enabled
+
     def _mark_handle_active(self, handle: LockHandle) -> None:
         handle.last_active_at = time.time()
 
@@ -55,7 +61,10 @@ async def start(self) -> None:
         """Start background cleanup and redo recovery."""
         self._running = True
         self._cleanup_task = asyncio.create_task(self._stale_cleanup_loop())
-        self._redo_task = asyncio.create_task(self._recover_pending_redo())
+        if self._redo_recovery_enabled:
+            self._redo_task = asyncio.create_task(self._recover_pending_redo())
+        else:
+            logger.info("Redo recovery disabled by config; skipping pending redo recovery")
 
     async def stop(self) -> None:
         """Stop cleanup and release all active locks."""
@@ -367,9 +376,15 @@ def init_lock_manager(
     agfs: AGFSClient,
     lock_timeout: float = 0.0,
     lock_expire: float = 300.0,
+    redo_recovery_enabled: bool = True,
 ) -> LockManager:
     global _lock_manager
-    _lock_manager = LockManager(agfs=agfs, lock_timeout=lock_timeout, lock_expire=lock_expire)
+    _lock_manager = LockManager(
+        agfs=agfs,
+        lock_timeout=lock_timeout,
+        lock_expire=lock_expire,
+        redo_recovery_enabled=redo_recovery_enabled,
+    )
     return _lock_manager
 
 

openviking/storage/transaction/lock_manager.py

@@ -29,4 +29,12 @@ class TransactionConfig(BaseModel):
         ),
     )
 
+    redo_recovery_enabled: bool = Field(
+        default=True,
+        description=(
+            "Enable session commit phase-2 crash-recovery redo. "
+            "When false, pending redo markers are not written and startup redo recovery is skipped."
+        ),
+    )
+
     model_config = {"extra": "forbid"}

openviking_cli/utils/config/transaction_config.py

@@ -5,13 +5,15 @@
 
 import asyncio
 import json
+from unittest.mock import MagicMock
 
 import pytest
 
 from openviking import AsyncOpenViking
 from openviking.message import TextPart
 from openviking.service.task_tracker import get_task_tracker
 from openviking.session import Session
+from openviking.storage.transaction import get_lock_manager
 from openviking_cli.exceptions import FailedPreconditionError
 
 
@@ -236,3 +238,20 @@ async def failing_extract(*args, **kwargs):
         session.add_message("user", [TextPart("Second round message")])
         with pytest.raises(FailedPreconditionError, match="unresolved failed archive"):
             await session.commit_async()
+
+    async def test_commit_skips_redo_when_recovery_disabled(
+        self, session_with_messages: Session, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Phase 2 should not write or clear redo markers when redo recovery is disabled."""
+
+        redo_log = MagicMock()
+        lock_manager = get_lock_manager()
+        monkeypatch.setattr(lock_manager, "_redo_recovery_enabled", False)
+        monkeypatch.setattr(lock_manager, "_redo_log", redo_log)
+
+        result = await session_with_messages.commit_async()
+        task_result = await _wait_for_task(result["task_id"])
+
+        assert task_result["status"] == "completed"
+        redo_log.write_pending.assert_not_called()
+        redo_log.mark_done.assert_not_called()

tests/session/test_session_commit.py

@@ -198,6 +198,24 @@ def test_openviking_config_retrieval_hotness_alpha_defaults_to_zero(monkeypatch)
 
     assert config.retrieval.hotness_alpha == 0.0
     assert config.retrieval.score_propagation_alpha == 0.5
+    assert config.storage.transaction.redo_recovery_enabled is True
+
+    OpenVikingConfigSingleton.reset_instance()
+
+
+def test_openviking_config_transaction_redo_recovery_enabled_can_be_disabled(monkeypatch):
+    monkeypatch.setenv(OPENVIKING_CONFIG_ENV, "/tmp/codex-no-config.json")
+
+    from openviking_cli.utils.config.open_viking_config import (
+        OpenVikingConfig,
+        OpenVikingConfigSingleton,
+    )
+
+    config = OpenVikingConfig.from_dict(
+        {"storage": {"transaction": {"redo_recovery_enabled": False}}}
+    )
+
+    assert config.storage.transaction.redo_recovery_enabled is False
 
     OpenVikingConfigSingleton.reset_instance()
 

tests/test_config_loader.py

@@ -99,3 +99,20 @@ async def test_recover_pending_redo_preserves_cancelled_error(self, lm):
             await lm._recover_pending_redo()
 
         lm._redo_log.mark_done.assert_not_called()
+
+    async def test_start_skips_redo_recovery_when_disabled(self, client):
+        lm_disabled = LockManager(
+            agfs=client._client.service._agfs_client,
+            lock_timeout=1.0,
+            lock_expire=1.0,
+            redo_recovery_enabled=False,
+        )
+        lm_disabled._recover_pending_redo = AsyncMock()
+
+        await lm_disabled.start()
+        await asyncio.sleep(0)
+
+        assert lm_disabled._redo_task is None
+        lm_disabled._recover_pending_redo.assert_not_called()
+
+        await lm_disabled.stop()