diff --git a/haiku_rag_slim/haiku/rag/ingester/app.py b/haiku_rag_slim/haiku/rag/ingester/app.py index be62684b..4e1ba821 100644 --- a/haiku_rag_slim/haiku/rag/ingester/app.py +++ b/haiku_rag_slim/haiku/rag/ingester/app.py @@ -43,8 +43,12 @@ class IngesterApp: ingester_cfg = self._config.ingester self._queue_conn = await open_queue(ingester_cfg.queue.path) try: - self._jobs = JobRepo(self._queue_conn) - self._sync = SyncStateRepo(self._queue_conn) + # Single lock shared by both repos so cross-repo calls on the + # same connection (e.g. worker's mark_succeeded then sync.upsert) + # serialize at the cursor/commit boundary. + queue_lock = asyncio.Lock() + self._jobs = JobRepo(self._queue_conn, lock=queue_lock) + self._sync = SyncStateRepo(self._queue_conn, lock=queue_lock) supported_extensions = get_converter(self._config).supported_extensions retry = RetryPolicy( diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py index 7fef15ee..3890b49e 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py +++ b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py @@ -50,7 +50,11 @@ def _row_to_sync_state(row: aiosqlite.Row) -> SyncStateRow: class JobRepo: - def __init__(self, conn: aiosqlite.Connection): + def __init__( + self, + conn: aiosqlite.Connection, + lock: asyncio.Lock | None = None, + ): # Row access by name in helpers below. conn.row_factory = aiosqlite.Row self._conn = conn @@ -58,8 +62,10 @@ class JobRepo: # coroutine don't sit "in progress" when another tries to commit. # aiosqlite executes statements on a single worker thread, but # individual cursors don't finalize until closed or GC'd — SQLite - # then refuses commit() with "SQL statements in progress". - self._lock = asyncio.Lock() + # then refuses commit() with "SQL statements in progress". When + # JobRepo and SyncStateRepo share the same connection, callers must + # pass the same lock instance so cross-repo calls also serialize. + self._lock = lock or asyncio.Lock() async def enqueue( self, @@ -363,11 +369,16 @@ class JobRepo: class SyncStateRepo: - def __init__(self, conn: aiosqlite.Connection): + def __init__( + self, + conn: aiosqlite.Connection, + lock: asyncio.Lock | None = None, + ): conn.row_factory = aiosqlite.Row self._conn = conn - # See JobRepo for why we serialize on the shared connection. - self._lock = asyncio.Lock() + # Pass the same lock instance JobRepo uses when both wrap one + # connection. See JobRepo for the SQLite cursor + commit constraint. + self._lock = lock or asyncio.Lock() async def get_snapshot(self, source_id: str) -> dict[str, str]: """uri -> revision map for the source. Drops rows where revision is diff --git a/tests/ingester/test_api.py b/tests/ingester/test_api.py index 67d08972..8d80ea83 100644 --- a/tests/ingester/test_api.py +++ b/tests/ingester/test_api.py @@ -1,3 +1,4 @@ +import asyncio from datetime import UTC, datetime import aiosqlite @@ -28,13 +29,18 @@ async def conn(tmp_path): @pytest.fixture -def jobs(conn): - return JobRepo(conn) +def queue_lock(): + return asyncio.Lock() @pytest.fixture -def sync(conn): - return SyncStateRepo(conn) +def jobs(conn, queue_lock): + return JobRepo(conn, lock=queue_lock) + + +@pytest.fixture +def sync(conn, queue_lock): + return SyncStateRepo(conn, lock=queue_lock) @pytest.fixture diff --git a/tests/ingester/test_pollers.py b/tests/ingester/test_pollers.py index 9c4534c7..d1e06e23 100644 --- a/tests/ingester/test_pollers.py +++ b/tests/ingester/test_pollers.py @@ -36,13 +36,18 @@ async def conn(tmp_path): @pytest.fixture -def jobs(conn): - return JobRepo(conn) +def queue_lock(): + return asyncio.Lock() @pytest.fixture -def sync(conn): - return SyncStateRepo(conn) +def jobs(conn, queue_lock): + return JobRepo(conn, lock=queue_lock) + + +@pytest.fixture +def sync(conn, queue_lock): + return SyncStateRepo(conn, lock=queue_lock) class _StubSource: diff --git a/tests/ingester/test_queue.py b/tests/ingester/test_queue.py index 470d69db..762c5db0 100644 --- a/tests/ingester/test_queue.py +++ b/tests/ingester/test_queue.py @@ -20,13 +20,18 @@ async def conn(tmp_path): @pytest.fixture -def jobs(conn): - return JobRepo(conn) +def queue_lock(): + return asyncio.Lock() @pytest.fixture -def sync(conn): - return SyncStateRepo(conn) +def jobs(conn, queue_lock): + return JobRepo(conn, lock=queue_lock) + + +@pytest.fixture +def sync(conn, queue_lock): + return SyncStateRepo(conn, lock=queue_lock) # --- migrations / schema --- @@ -578,6 +583,29 @@ async def test_counts_by_source_no_statuses_returns_empty(jobs): assert await jobs.counts_by_source() == {} +# --- cross-repo lock sharing --- + + +def test_repos_share_lock_when_constructed_with_one(conn): + """JobRepo and SyncStateRepo wrap one shared aiosqlite.Connection in + production. They must accept and share a single asyncio.Lock so cross- + repo calls serialize at the cursor/commit boundary — otherwise a + SyncStateRepo.upsert cursor open while JobRepo.mark_succeeded tries + to commit would trip SQLite's 'SQL statements in progress' error.""" + shared = asyncio.Lock() + j = JobRepo(conn, lock=shared) + s = SyncStateRepo(conn, lock=shared) + assert j._lock is s._lock is shared + + +def test_repos_default_to_independent_locks_when_used_alone(conn): + """Backward-compat: a single-repo caller can still construct without + passing a lock and each repo creates its own.""" + j = JobRepo(conn) + s = SyncStateRepo(conn) + assert j._lock is not s._lock + + # --- sync state --- diff --git a/tests/ingester/test_serve_integration.py b/tests/ingester/test_serve_integration.py index e4a10ba7..2b9d7137 100644 --- a/tests/ingester/test_serve_integration.py +++ b/tests/ingester/test_serve_integration.py @@ -28,13 +28,18 @@ async def conn(tmp_path): @pytest.fixture -def jobs(conn): - return JobRepo(conn) +def queue_lock(): + return asyncio.Lock() @pytest.fixture -def sync(conn): - return SyncStateRepo(conn) +def jobs(conn, queue_lock): + return JobRepo(conn, lock=queue_lock) + + +@pytest.fixture +def sync(conn, queue_lock): + return SyncStateRepo(conn, lock=queue_lock) async def _wait_for(predicate, *, timeout: float = 5.0, interval: float = 0.05): diff --git a/tests/ingester/test_workers.py b/tests/ingester/test_workers.py index 4f229af1..74fa65b2 100644 --- a/tests/ingester/test_workers.py +++ b/tests/ingester/test_workers.py @@ -25,13 +25,18 @@ async def conn(tmp_path): @pytest.fixture -def jobs(conn): - return JobRepo(conn) +def queue_lock(): + return asyncio.Lock() @pytest.fixture -def sync(conn): - return SyncStateRepo(conn) +def jobs(conn, queue_lock): + return JobRepo(conn, lock=queue_lock) + + +@pytest.fixture +def sync(conn, queue_lock): + return SyncStateRepo(conn, lock=queue_lock) @pytest.fixture