diff --git a/hermes_state.py b/hermes_state.py index 08d084e4d16bb..87992e0d273b2 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -3715,18 +3715,6 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) ): return False - # An expired row is harmless: a publisher must revalidate its lease - # before committing, while an active row indicates a handoff may - # still be in flight. - active_lock = conn.execute( - "SELECT 1 FROM compression_locks " - "WHERE session_id = ? " - "AND (expires_at IS NULL OR expires_at >= ?) LIMIT 1", - (session_id, time.time()), - ).fetchone() - if active_lock is not None: - return False - # Treat any direct non-branch/non-delegate/non-tool child as a # continuation, regardless of its current ended state. Reopening # in that case could create a second live head for one lineage. @@ -3745,6 +3733,29 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) if child is not None: return False + # refresh_compression_lock() deliberately lets an owner revive its + # own expired row. Reclaim that row inside this write transaction + # before reopening: refresh-first makes the lease active and aborts + # recovery; recovery-first deletes the holder identity so a later + # refresh cannot resurrect it. + now = time.time() + lock_row = conn.execute( + "SELECT holder, expires_at FROM compression_locks " + "WHERE session_id = ?", + (session_id,), + ).fetchone() + if lock_row is not None: + expires_at = lock_row["expires_at"] + if expires_at is None or float(expires_at) >= now: + return False + deleted = conn.execute( + "DELETE FROM compression_locks " + "WHERE session_id = ? AND holder = ? AND expires_at = ?", + (session_id, lock_row["holder"], expires_at), + ) + if deleted.rowcount != 1: + return False + updated = conn.execute( "UPDATE sessions SET ended_at = NULL, end_reason = NULL " "WHERE id = ? AND ended_at IS NOT NULL " diff --git a/tests/state/test_compression_lineage_guard.py b/tests/state/test_compression_lineage_guard.py index bc00758d8afea..1665c1c52433e 100644 --- a/tests/state/test_compression_lineage_guard.py +++ b/tests/state/test_compression_lineage_guard.py @@ -2,6 +2,8 @@ from __future__ import annotations +import time + import pytest from hermes_state import SessionDB @@ -102,6 +104,44 @@ def test_reopen_orphaned_compression_session_fails_closed_with_active_lease( assert db.get_session("leased-parent")["end_reason"] == "compression" +def test_reopen_orphaned_compression_session_reclaims_expired_lease( + db: SessionDB, +) -> None: + _compression_parent(db, "expired-lease-parent") + now = time.time() + db._conn.execute( + "INSERT INTO compression_locks " + "(session_id, holder, acquired_at, expires_at) VALUES (?, ?, ?, ?)", + ("expired-lease-parent", "old-compressor", now - 60, now - 30), + ) + db._conn.commit() + + assert db.reopen_orphaned_compression_session("expired-lease-parent") is True + assert db.refresh_compression_lock( + "expired-lease-parent", "old-compressor" + ) is False + assert db.get_compression_lock_holder("expired-lease-parent") is None + + +def test_reopen_orphaned_compression_session_loses_to_expired_lease_refresh( + db: SessionDB, +) -> None: + _compression_parent(db, "refreshed-lease-parent") + now = time.time() + db._conn.execute( + "INSERT INTO compression_locks " + "(session_id, holder, acquired_at, expires_at) VALUES (?, ?, ?, ?)", + ("refreshed-lease-parent", "live-compressor", now - 60, now - 30), + ) + db._conn.commit() + + assert db.refresh_compression_lock( + "refreshed-lease-parent", "live-compressor" + ) is True + assert db.reopen_orphaned_compression_session("refreshed-lease-parent") is False + assert db.get_session("refreshed-lease-parent")["end_reason"] == "compression" + + def test_find_live_compression_child_ignores_non_continuation_children(