Serialize JobRepo and SyncStateRepo on one shared lock

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(

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

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

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:

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 ---
 
 

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):

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