Merge pull request #441 from ggozad/fix/sqlite-pool

Widen SQLite ingester queue pool to serve concurrent connections

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 ## [Unreleased]
 
+### Fixed
+
+- SQLite ingester queue runs with a multi-connection pool (`pool_size=5, max_overflow=5`) instead of a single connection. API reads (`/stats`, `/jobs`) no longer time out with `QueuePool limit of size 1 reached` while workers hold the connection.
+
 ## [0.58.0] - 2026-06-15
 
 ### Added

haiku_rag_slim/haiku/rag/ingester/queue/migrations.py

@@ -16,10 +16,11 @@ __all__ = ["SCHEMA_VERSION", "apply_migrations", "make_engine", "open_queue"]
 def make_engine(config: QueueConfig) -> AsyncEngine:
     """Build the queue's AsyncEngine from config. Uses `dburi` when set,
     otherwise a `sqlite+aiosqlite` URL pointing at the resolved `path`
-    (creating the parent directory). SQLite is capped to a single pooled
+    (creating the parent directory). SQLite runs in WAL mode with a small pool
-    connection so the claim stays atomic without row locks; Postgres uses
+    so reads (API stats/jobs) proceed concurrently with worker writes; the
-    pool_pre_ping so a long-running ingester survives a DB restart or idle
+    claim stays atomic via its single UPDATE statement, not the pool size.
-    connection drop."""
+    Postgres uses pool_pre_ping so a long-running ingester survives a DB
+    restart or idle connection drop."""
     if config.dburi:
         url = make_url(config.dburi)
     else:
@@ -30,7 +31,7 @@ def make_engine(config: QueueConfig) -> AsyncEngine:
         url = URL.create("sqlite+aiosqlite", database=str(path))
 
     if url.get_backend_name() == "sqlite":
-        engine = create_async_engine(url, pool_size=1, max_overflow=0)
+        engine = create_async_engine(url, pool_size=5, max_overflow=5)
     else:
         engine = create_async_engine(url, pool_pre_ping=True)
     install_sqlite_pragmas(engine)

tests/ingester/test_queue.py

@@ -60,6 +60,33 @@ async def test_make_engine_postgres_is_pre_ping():
         await engine.dispose()
 
 
+@pytest.mark.asyncio
+async def test_sqlite_engine_serves_concurrent_connections(tmp_path):
+    """The SQLite queue pool hands out more than one connection at a time so an
+    API read does not starve while a worker holds a connection. WAL mode makes
+    concurrent readers safe; the claim stays atomic via its single UPDATE."""
+    engine = make_engine(QueueConfig(path=tmp_path / "queue.db"))
+    await apply_migrations(engine)
+    held: list = []
+    try:
+
+        async def acquire() -> None:
+            conn = await engine.connect()
+            held.append(conn)
+            await conn.execute(sa.text("SELECT 1"))
+
+        # All three checkouts are held at once; a single-connection pool blocks
+        # the second and third until the checkout timeout.
+        await asyncio.wait_for(
+            asyncio.gather(*(acquire() for _ in range(3))), timeout=3
+        )
+        assert len(held) == 3
+    finally:
+        for conn in held:
+            await conn.close()
+        await engine.dispose()
+
+
 def test_insert_uses_dialect_specific_construct():
     """_insert dispatches to the dialect's INSERT (which exposes on_conflict_*)."""
     from sqlalchemy.dialects.postgresql import Insert as PostgresInsert