From 41419eaf60fc5b6ea289d1137ec64dcdbbb89305 Mon Sep 17 00:00:00 2001 From: Samuel Groot Date: Wed, 26 Aug 2026 14:48:08 +0000 Subject: [PATCH 1/5] Run SQLite in WAL mode Under SQLite's default rollback journal, a connection holding an unfinished read keeps the shared lock, and any concurrent COMMIT then fails immediately with SQLITE_BUSY ("database is locked"). SQLite does not consult the busy handler for that conflict, so busy_timeout does not help - only WAL does. This is reachable in the scheduler: `job_update_commiter` commits job updates while other tasks read the same file, and it swallows the failure, so a losing COMMIT silently drops a job update - including the backend_id that resuming a job relies on. It also made the sqlite test job flaky. Cancelling a task inside a query - which every scheduler test does on teardown - abandons its statement half-read, so that connection holds the shared lock until it is closed. A sweep of 40 cancel timings failed 15 times before this change and 0 times after. Engine creation moves into `build_engine` so all callers get the pragma, and it applies `echo` from the config instead of each caller repeating it. Two test fixtures also dropped `engine.dispose()` whenever their `drop_all` teardown raised. Every backend shares one database across the session, so that leaked the failing test's connections - and their locks - into every later test, which is how one flake failed the two following tests too. Both now dispose in a `finally`. --- tests/api/conftest.py | 12 ++++++--- tests/scheduler/conftest.py | 23 +++++++++------- tests/test_database.py | 39 ++++++++++++++++++++++++++++ warden/api/routes/dependencies/db.py | 6 ++--- warden/lib/db/database.py | 32 +++++++++++++++++++++++ warden/scheduler/main.py | 5 ++-- 6 files changed, 99 insertions(+), 18 deletions(-) create mode 100644 tests/test_database.py diff --git a/tests/api/conftest.py b/tests/api/conftest.py index eef0b32..e76b0ac 100644 --- a/tests/api/conftest.py +++ b/tests/api/conftest.py @@ -31,9 +31,15 @@ async def app(db_backend_config: DatabaseConfig) -> AsyncGenerator[FastAPI, None # create tables in the test database async with app.state.db_engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) - yield app - async with app.state.db_engine.begin() as conn: - await conn.run_sync(Base.metadata.drop_all) + try: + yield app + + async with app.state.db_engine.begin() as conn: + await conn.run_sync(Base.metadata.drop_all) + finally: + # See tests/scheduler/conftest.py: the database is shared, so the + # engine has to be disposed even when the teardown above fails. + await app.state.db_engine.dispose() @pytest_asyncio.fixture diff --git a/tests/scheduler/conftest.py b/tests/scheduler/conftest.py index 12b83f2..f681d80 100644 --- a/tests/scheduler/conftest.py +++ b/tests/scheduler/conftest.py @@ -1,24 +1,29 @@ """Pytest fixture and configurations""" import pytest_asyncio -from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.ext.asyncio import async_sessionmaker -from warden.lib.db.database import Base, build_db_url +from warden.lib.db.database import Base, build_engine @pytest_asyncio.fixture(scope="function") async def db_engine(config_db): - engine = create_async_engine(build_db_url(config_db.database)) + engine = build_engine(config_db.database) async with engine.begin() as conn: # Create all tables once await conn.run_sync(Base.metadata.create_all) - yield engine - - async with engine.begin() as conn: - # Delete tables - await conn.run_sync(Base.metadata.drop_all) - await engine.dispose() + try: + yield engine + + async with engine.begin() as conn: + # Delete tables + await conn.run_sync(Base.metadata.drop_all) + finally: + # Always dispose: every backend shares one database across the test + # session, so an engine left open on a failing teardown leaks its + # connections - and their transactions - into every later test. + await engine.dispose() @pytest_asyncio.fixture(scope="function") diff --git a/tests/test_database.py b/tests/test_database.py new file mode 100644 index 0000000..25d6d26 --- /dev/null +++ b/tests/test_database.py @@ -0,0 +1,39 @@ +"""Testing warden.lib.db.database""" + +import pytest +from sqlalchemy import text + +from warden.lib.config.config import SqliteConfig +from warden.lib.db.database import build_engine + + +@pytest.mark.asyncio +async def test_sqlite_commit_while_a_read_is_in_flight(tmp_path): + """An unfinished read must not make a concurrent COMMIT fail. + + Cancelling a task while it sits inside a query - which the scheduler does + on shutdown, and every scheduler test does on teardown - abandons its + statement half-read, so that connection keeps SQLite's shared lock until it + is closed. Under the default rollback journal the next COMMIT then fails + right away with "database is locked"; SQLite does not consult the busy + handler for that conflict, so `busy_timeout` is no help either. Only WAL + lets the writer through. + """ + engine = build_engine(SqliteConfig(name=str(tmp_path / "warden.db"))) + try: + async with engine.begin() as conn: + await conn.execute(text("CREATE TABLE t (v INTEGER)")) + await conn.execute(text("INSERT INTO t VALUES (1), (2), (3)")) + + # Stand-in for the connection of a task cancelled mid-query: a + # statement left stepping, never finished, never rolled back. + reader = await engine.connect() + in_flight = await reader.stream(text("SELECT v FROM t")) + await in_flight.fetchone() + + async with engine.begin() as conn: + await conn.execute(text("INSERT INTO t VALUES (4)")) + + await reader.close() + finally: + await engine.dispose() diff --git a/warden/api/routes/dependencies/db.py b/warden/api/routes/dependencies/db.py index 6af1224..4ccf6b8 100644 --- a/warden/api/routes/dependencies/db.py +++ b/warden/api/routes/dependencies/db.py @@ -1,15 +1,15 @@ from typing import Annotated, AsyncGenerator from fastapi import Depends, FastAPI, Request -from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from warden.lib.config import DatabaseConfig -from warden.lib.db.database import build_db_url +from warden.lib.db.database import build_engine def init_db(app: FastAPI, db_config: DatabaseConfig): """Initialize the async engine and session factory with the given DB URL.""" - engine = create_async_engine(build_db_url(db_config), echo=db_config.echo) + engine = build_engine(db_config) # TODO: ensure isolation between concurrent requests session_factory = async_sessionmaker(bind=engine, expire_on_commit=False) diff --git a/warden/lib/db/database.py b/warden/lib/db/database.py index 496d60c..46b4388 100644 --- a/warden/lib/db/database.py +++ b/warden/lib/db/database.py @@ -1,6 +1,10 @@ """Warden db utils""" +from typing import Any + +from sqlalchemy import event from sqlalchemy.engine.url import URL +from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine from sqlalchemy.orm import declarative_base from warden.lib.config import DatabaseConfig @@ -35,3 +39,31 @@ def build_db_url(cfg: DatabaseConfig) -> str: ).render_as_string(hide_password=False) raise ValueError(f"Unsupported backend: {cfg.backend}") + + +def build_engine(cfg: DatabaseConfig) -> AsyncEngine: + """Build the async engine for `cfg`. + + SQLite is put in WAL mode. With the default rollback journal, a connection + holding an open read transaction makes any concurrent COMMIT fail straight + away with SQLITE_BUSY ("database is locked"): SQLite does not consult the + busy handler for that conflict, so `busy_timeout` is no help. The scheduler + commits job updates while other tasks read the same file, so the conflict + is reachable - and `job_update_commiter` would silently drop the update. + WAL lets readers and a writer coexist. The pragma is stored in the database + file, so it only has to be set once, but setting it per connection keeps it + correct for a freshly created file. + """ + engine = create_async_engine(build_db_url(cfg), echo=cfg.echo) + + if cfg.backend == "sqlite": + + @event.listens_for(engine.sync_engine, "connect") + def _enable_wal(dbapi_connection: Any, _connection_record: Any) -> None: + cursor = dbapi_connection.cursor() + try: + cursor.execute("PRAGMA journal_mode=WAL") + finally: + cursor.close() + + return engine diff --git a/warden/scheduler/main.py b/warden/scheduler/main.py index c48a989..7b2f1f4 100644 --- a/warden/scheduler/main.py +++ b/warden/scheduler/main.py @@ -9,11 +9,10 @@ from sqlalchemy.ext.asyncio import ( AsyncEngine, async_sessionmaker, - create_async_engine, ) from warden.lib.config import Config -from warden.lib.db.database import build_db_url +from warden.lib.db.database import build_engine from warden.lib.models import Job from warden.scheduler.cancellation_worker import cancellation_worker from warden.scheduler.db import job_update_commiter @@ -144,7 +143,7 @@ async def main_async(conf: Config | None = None): conf = Config() logging.config.dictConfig(config=conf.logging) - engine = create_async_engine(build_db_url(conf.database), echo=conf.database.echo) + engine = build_engine(conf.database) loop = asyncio.get_running_loop() stop_event = asyncio.Event() From 6bebf89f55c67b2c894f1a5ff4899d069bcf8e57 Mon Sep 17 00:00:00 2001 From: Samuel Groot Date: Fri, 4 Sep 2026 14:34:45 +0200 Subject: [PATCH 2/5] Remove WAL --- warden/lib/db/database.py | 22 +--------------------- 1 file changed, 1 insertion(+), 21 deletions(-) diff --git a/warden/lib/db/database.py b/warden/lib/db/database.py index 46b4388..1949333 100644 --- a/warden/lib/db/database.py +++ b/warden/lib/db/database.py @@ -42,28 +42,8 @@ def build_db_url(cfg: DatabaseConfig) -> str: def build_engine(cfg: DatabaseConfig) -> AsyncEngine: - """Build the async engine for `cfg`. + """Build the async engine for `cfg`.""" - SQLite is put in WAL mode. With the default rollback journal, a connection - holding an open read transaction makes any concurrent COMMIT fail straight - away with SQLITE_BUSY ("database is locked"): SQLite does not consult the - busy handler for that conflict, so `busy_timeout` is no help. The scheduler - commits job updates while other tasks read the same file, so the conflict - is reachable - and `job_update_commiter` would silently drop the update. - WAL lets readers and a writer coexist. The pragma is stored in the database - file, so it only has to be set once, but setting it per connection keeps it - correct for a freshly created file. - """ engine = create_async_engine(build_db_url(cfg), echo=cfg.echo) - if cfg.backend == "sqlite": - - @event.listens_for(engine.sync_engine, "connect") - def _enable_wal(dbapi_connection: Any, _connection_record: Any) -> None: - cursor = dbapi_connection.cursor() - try: - cursor.execute("PRAGMA journal_mode=WAL") - finally: - cursor.close() - return engine From c161a2f317da6c25862583c9797c703f5798788d Mon Sep 17 00:00:00 2001 From: Samuel Groot Date: Wed, 9 Sep 2026 13:46:20 +0200 Subject: [PATCH 3/5] Remove unused tests --- tests/test_database.py | 39 --------------------------------------- 1 file changed, 39 deletions(-) delete mode 100644 tests/test_database.py diff --git a/tests/test_database.py b/tests/test_database.py deleted file mode 100644 index 25d6d26..0000000 --- a/tests/test_database.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Testing warden.lib.db.database""" - -import pytest -from sqlalchemy import text - -from warden.lib.config.config import SqliteConfig -from warden.lib.db.database import build_engine - - -@pytest.mark.asyncio -async def test_sqlite_commit_while_a_read_is_in_flight(tmp_path): - """An unfinished read must not make a concurrent COMMIT fail. - - Cancelling a task while it sits inside a query - which the scheduler does - on shutdown, and every scheduler test does on teardown - abandons its - statement half-read, so that connection keeps SQLite's shared lock until it - is closed. Under the default rollback journal the next COMMIT then fails - right away with "database is locked"; SQLite does not consult the busy - handler for that conflict, so `busy_timeout` is no help either. Only WAL - lets the writer through. - """ - engine = build_engine(SqliteConfig(name=str(tmp_path / "warden.db"))) - try: - async with engine.begin() as conn: - await conn.execute(text("CREATE TABLE t (v INTEGER)")) - await conn.execute(text("INSERT INTO t VALUES (1), (2), (3)")) - - # Stand-in for the connection of a task cancelled mid-query: a - # statement left stepping, never finished, never rolled back. - reader = await engine.connect() - in_flight = await reader.stream(text("SELECT v FROM t")) - await in_flight.fetchone() - - async with engine.begin() as conn: - await conn.execute(text("INSERT INTO t VALUES (4)")) - - await reader.close() - finally: - await engine.dispose() From 3cc09f0daa0112fa04bd3cbec8c4178e84c1bc60 Mon Sep 17 00:00:00 2001 From: Samuel Groot Date: Wed, 9 Sep 2026 14:42:04 +0200 Subject: [PATCH 4/5] Fix lint --- warden/lib/db/database.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/warden/lib/db/database.py b/warden/lib/db/database.py index 1949333..f33cd87 100644 --- a/warden/lib/db/database.py +++ b/warden/lib/db/database.py @@ -1,8 +1,5 @@ """Warden db utils""" -from typing import Any - -from sqlalchemy import event from sqlalchemy.engine.url import URL from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine from sqlalchemy.orm import declarative_base From ec4d9101a7e6f83e2db7303d0dd613d43e2a655b Mon Sep 17 00:00:00 2001 From: Samuel Groot Date: Wed, 9 Sep 2026 14:49:15 +0200 Subject: [PATCH 5/5] Add Claude .gitignore --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.gitignore b/.gitignore index d816355..4ef1c6e 100644 --- a/.gitignore +++ b/.gitignore @@ -25,3 +25,5 @@ config.*.yaml venv .venv dist + +.claude