From d2ce2c852d919b43a17f5966095d969ff8fa7407 Mon Sep 17 00:00:00 2001 From: yoniebans Date: Mon, 29 Jun 2026 14:22:46 +0200 Subject: [PATCH] test(gateway): assert interleaving safety of concurrent offloaded DB calls --- tests/gateway/test_async_session_db.py | 30 ++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/tests/gateway/test_async_session_db.py b/tests/gateway/test_async_session_db.py index 4a3674757d3..c09b5485ae7 100644 --- a/tests/gateway/test_async_session_db.py +++ b/tests/gateway/test_async_session_db.py @@ -370,3 +370,33 @@ def test_offloaded_helpers_never_called_bare_on_loop(): "Offloaded sync helper called bare on the gateway loop — wrap in " "await asyncio.to_thread(self., ...):\n " + "\n ".join(violations) ) + + +# -------------------------------------------------------------------------- +# Interleaving safety: offloading opens await points where coroutines can +# interleave against the same session rows. The gateway relies on SessionDB's +# atomic operations (compare-and-set, INSERT OR IGNORE) to stay single-winner. +# These pin that the defenses hold when driven concurrently through the facade. +# -------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_concurrent_claim_handoff_single_winner(tmp_path): + db = AsyncSessionDB(hermes_state.SessionDB(db_path=tmp_path / "state.db")) + sid = "s-handoff" + await db.create_session(sid, "test") + await db.request_handoff(sid, "telegram") + + results = await asyncio.gather(*(db.claim_handoff(sid) for _ in range(20))) + + assert sum(results) == 1, f"exactly one claim must win, got {sum(results)}" + + +@pytest.mark.asyncio +async def test_concurrent_create_session_idempotent(tmp_path): + db = AsyncSessionDB(hermes_state.SessionDB(db_path=tmp_path / "state.db")) + sid = "s-create" + + await asyncio.gather(*(db.create_session(sid, "test") for _ in range(20))) + + rows = await db.list_sessions_rich(limit=100) + assert sum(1 for r in rows if r["id"] == sid) == 1