fix(agent): stop a locked lease refresh once its lifetime runs out
A SQLite lock is still a missed tick, but only while the next attempt lands before the row expiry. Past that, acquisition can hand the session to someone else while this turn is still active.
This commit is contained in:
@@ -10,6 +10,7 @@ bodies run on per-handle workers), not per-turn threads.
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
from contextlib import nullcontext
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, List, Optional
|
||||
@@ -39,6 +40,8 @@ class DurableTurnLease:
|
||||
self.db = db
|
||||
self.session_id = session_id # id at admission; release always targets this row
|
||||
self.holder = holder
|
||||
# Wall clock, same as the row's expires_at. A failed refresh must not move it.
|
||||
self._authority_deadline = time.time() + LEASE_TTL_SECONDS
|
||||
self.stop = threading.Event()
|
||||
self.refresh_interval = float(getattr(agent, "_session_turn_lease_refresh_interval", 60.0))
|
||||
self._lock = threading.Lock()
|
||||
@@ -183,8 +186,14 @@ class DurableTurnLease:
|
||||
|
||||
def refresh_tick(self):
|
||||
"""One periodic renewal (every ``refresh_interval`` via the shared scheduler). A real
|
||||
miss (rowcount 0) or a non-lock error interrupts the turn. A SQLite lock is only a
|
||||
missed tick: the timer stays armed. Returning False stops the timer.
|
||||
miss (rowcount 0) or a non-lock error interrupts the turn. A SQLite lock is a missed
|
||||
tick only while the next attempt still lands before ``_authority_deadline``. Returning
|
||||
False stops the timer.
|
||||
|
||||
Acquisition reclaims a row once ``expires_at <= now``, even if the old process is
|
||||
alive. Holder-qualified release does not block that path, so a run of lock errors
|
||||
must not keep this turn active up to and past the deadline. A successful renewal
|
||||
is the only thing that moves the deadline.
|
||||
|
||||
The holder-qualified UPDATE fences a late refresher from a successor lease. The façade's
|
||||
finally sets ``stop`` before releasing, so a holder-fenced miss observed after stop is not
|
||||
@@ -195,6 +204,7 @@ class DurableTurnLease:
|
||||
if self.db.refresh_session_turn_lease(
|
||||
self._current_session_id(), self.holder, ttl_seconds=LEASE_TTL_SECONDS
|
||||
):
|
||||
self._authority_deadline = time.time() + LEASE_TTL_SECONDS
|
||||
return None
|
||||
if self.stop.is_set():
|
||||
return False
|
||||
@@ -205,12 +215,22 @@ class DurableTurnLease:
|
||||
except Exception as exc:
|
||||
if self.stop.is_set():
|
||||
return False
|
||||
# A lock is contention, not a lost holder. The row is still ours
|
||||
# (release is holder-qualified), so a convoy must not kill the turn.
|
||||
# Keep the timer: the next tick retries, and one miss fits inside the TTL.
|
||||
from hermes_state_errors import is_sqlite_lock_error
|
||||
|
||||
if is_sqlite_lock_error(exc):
|
||||
now = time.time()
|
||||
# The scheduler will not try again until one interval from now. If that
|
||||
# attempt would meet or pass the row expiry, stop before a successor
|
||||
# can claim it.
|
||||
if now + self.refresh_interval >= self._authority_deadline:
|
||||
logger.warning(
|
||||
"Session turn lease refresh stayed locked through its lifetime: %s",
|
||||
self._current_session_id(),
|
||||
)
|
||||
self._interrupt_turn(
|
||||
"Session turn lease could not be refreshed; stopping to protect the transcript."
|
||||
)
|
||||
return False
|
||||
logger.warning(
|
||||
"Session turn lease refresh hit a SQLite lock; will retry: %s",
|
||||
self._current_session_id(),
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Unit tests for agent.turn_facade_lease (admission + lease bracket)."""
|
||||
import sqlite3
|
||||
import threading
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
|
||||
from agent.turn_facade_lease import (
|
||||
@@ -145,6 +146,56 @@ def test_refresh_tick_sqlite_lock_keeps_the_turn():
|
||||
assert lease.stop.is_set() is False
|
||||
|
||||
|
||||
def test_repeated_lock_misses_do_not_outlive_the_lease(monkeypatch):
|
||||
now = [1000.0]
|
||||
monkeypatch.setattr(time, "time", lambda: now[0])
|
||||
db = _Db()
|
||||
|
||||
def locked(session_id, holder, **kwargs):
|
||||
raise sqlite3.OperationalError("database is locked")
|
||||
|
||||
db.refresh_session_turn_lease = locked
|
||||
lease, calls = _active_lease(db)
|
||||
result = None
|
||||
for elapsed in (60, 120, 180, 240, 300):
|
||||
now[0] = 1000.0 + elapsed
|
||||
result = lease.refresh_tick()
|
||||
if result is False:
|
||||
break
|
||||
assert result is False
|
||||
assert calls == [
|
||||
"Session turn lease could not be refreshed; stopping to protect the transcript."
|
||||
]
|
||||
assert now[0] <= 1000.0 + LEASE_TTL_SECONDS
|
||||
|
||||
|
||||
def test_one_lock_then_renewal_opens_a_fresh_window(monkeypatch):
|
||||
now = [1000.0]
|
||||
monkeypatch.setattr(time, "time", lambda: now[0])
|
||||
db = _Db()
|
||||
locked = {"on": True}
|
||||
|
||||
def refresh(session_id, holder, **kwargs):
|
||||
if locked["on"]:
|
||||
raise sqlite3.OperationalError("database is locked")
|
||||
return True
|
||||
|
||||
db.refresh_session_turn_lease = refresh
|
||||
lease, calls = _active_lease(db)
|
||||
now[0] = 1060.0
|
||||
assert lease.refresh_tick() is None and calls == []
|
||||
locked["on"] = False
|
||||
now[0] = 1120.0
|
||||
assert lease.refresh_tick() is None and calls == []
|
||||
locked["on"] = True
|
||||
# Original admission deadline was 1300. The renewal moved it to 1420.
|
||||
now[0] = 1300.0
|
||||
assert lease.refresh_tick() is None and calls == []
|
||||
now[0] = 1360.0
|
||||
assert lease.refresh_tick() is False
|
||||
assert calls
|
||||
|
||||
|
||||
def test_refresh_tick_real_loss_still_interrupts():
|
||||
db = _Db()
|
||||
db.refresh_session_turn_lease = lambda session_id, holder, **kwargs: False
|
||||
|
||||
Reference in New Issue
Block a user