Widen SQLite ingester queue pool to serve concurrent connections
This commit is contained in:
parent
f94c419c31
commit
7e20b47e98
3 changed files with 37 additions and 5 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
connection so the claim stays atomic without row locks; Postgres uses
|
||||
pool_pre_ping so a long-running ingester survives a DB restart or idle
|
||||
connection drop."""
|
||||
(creating the parent directory). SQLite runs in WAL mode with a small pool
|
||||
so reads (API stats/jobs) proceed concurrently with worker writes; the
|
||||
claim stays atomic via its single UPDATE statement, not the pool size.
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in a new issue