From b079dbc0484033190250ed4e73f18ac135504b31 Mon Sep 17 00:00:00 2001 From: Michael Sitarzewski Date: Fri, 28 Aug 2026 10:07:41 -0500 Subject: [PATCH] fix: stop racing on the WAL switch at worker startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every boot and deploy cost a full 900s aggregation cycle: 14:00:17 aggregate worker started interval=900s 14:00:18 aggregate cycle failed: OperationalError: database is locked One second after start, on every restart, because all four workers open the database in the same instant. The obvious reading is lock contention, but it does not survive the evidence: Python's sqlite3 already waits five seconds by default, and the failure came after one. Something was refusing to wait at all. It was the journal_mode switch. journal_mode is a PERSISTENT property of the database file, so re-issuing it on every open was already pointless — and it is not free: switching journal mode needs an exclusive lock, and SQLite answers SQLITE_BUSY for it immediately rather than honouring the busy handler. No timeout could ever have rescued it. The workers were racing on a no-op. It now reads the current mode and writes only when it differs. The busy timeout is raised to 30s as well. That is defence in depth for the schema/migration block, which every open runs inside a write transaction, and not the fix for this bug. Four tests. The one that guards the real cause traces the statements issued during a second open and asserts the WAL switch is not among them; reverting the conditional fails it. The concurrency test is behavioural only and says so — a 0.4s hold would pass under the old five-second default too, and claiming otherwise would be a test that looks like a guard without being one. Co-Authored-By: Claude Opus 5 (1M context) --- fintick/storage.py | 28 +++++++++- tests/test_storage_locking.py | 100 ++++++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+), 2 deletions(-) create mode 100644 tests/test_storage_locking.py diff --git a/fintick/storage.py b/fintick/storage.py index 5ef6136..0997876 100644 --- a/fintick/storage.py +++ b/fintick/storage.py @@ -264,14 +264,38 @@ def _bootstrap_post_aggregation_decisions(connection: sqlite3.Connection) -> Non ) +# How long a connection waits for a lock before raising. Python's sqlite3 default is 5s, +# which is not enough when several workers open the database at the same instant and each +# runs the migration block in a write transaction. WAL lets readers and one writer coexist; +# this covers the writer-vs-writer moment at startup. +BUSY_TIMEOUT_SECONDS = 30.0 + + @contextmanager def open_database(path: str | Path) -> Iterator[sqlite3.Connection]: """Open and initialize a FinTick database, committing on success.""" database = Path(path) database.parent.mkdir(parents=True, exist_ok=True) - connection = sqlite3.connect(database) + connection = sqlite3.connect(database, timeout=BUSY_TIMEOUT_SECONDS) try: - connection.execute("PRAGMA journal_mode=WAL") + # The timeout on connect() above is what actually makes a blocked open wait; the + # PRAGMA restates it for any connection opened elsewhere and makes the value visible + # to `PRAGMA busy_timeout`. Set it before the WAL switch, which can itself need a lock. + # + # Why this is needed: every open runs the schema/migration block inside a write + # transaction, so four workers starting in the same second all contend for one write + # lock before doing any real work. That surfaced as "OperationalError: database is + # locked" one second after every boot and deploy, costing a full 900s cycle. The + # contention lasts milliseconds — waiting it out is the correct response. + connection.execute(f"PRAGMA busy_timeout={BUSY_TIMEOUT_SECONDS * 1000:.0f}") + # journal_mode is a PERSISTENT property of the database file, so re-setting it on + # every open is pointless — and not free. Switching journal mode needs an exclusive + # lock, and SQLite answers SQLITE_BUSY for it IMMEDIATELY rather than honouring the + # busy handler, so no timeout can save it. Four workers opening in the same second + # therefore raced on a switch that was already a no-op. Read first, write only if + # it actually needs changing. + if connection.execute("PRAGMA journal_mode").fetchone()[0].lower() != "wal": + connection.execute("PRAGMA journal_mode=WAL") connection.execute("PRAGMA foreign_keys=ON") connection.execute("BEGIN") # Legacy databases need columns before indexes can refer to them. diff --git a/tests/test_storage_locking.py b/tests/test_storage_locking.py new file mode 100644 index 0000000..b369d77 --- /dev/null +++ b/tests/test_storage_locking.py @@ -0,0 +1,100 @@ +"""Concurrent opens must wait for the lock, not raise.""" + +from __future__ import annotations + +import sqlite3 +import tempfile +import threading +import time +import unittest +from unittest import mock +from pathlib import Path + +from fintick.storage import BUSY_TIMEOUT_SECONDS, open_database + + +class BusyTimeoutTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.database = Path(self.tmp.name) / "fintick.db" + + def test_busy_timeout_is_raised_above_the_python_default(self) -> None: + with open_database(self.database) as connection: + timeout = connection.execute("PRAGMA busy_timeout").fetchone()[0] + # Python's default is 5000ms, which is not enough for several workers opening at once. + self.assertGreaterEqual(timeout, 30000) + self.assertEqual(BUSY_TIMEOUT_SECONDS, 30.0) + + def test_wal_is_still_enabled(self) -> None: + with open_database(self.database) as connection: + self.assertEqual( + connection.execute("PRAGMA journal_mode").fetchone()[0].lower(), "wal" + ) + + def test_the_wal_switch_is_skipped_when_already_enabled(self) -> None: + """The actual cause of the startup failure. + + Switching journal_mode needs an exclusive lock, and SQLite answers SQLITE_BUSY for it + IMMEDIATELY rather than honouring the busy handler — so no timeout can rescue it. The + switch was being re-issued on every open despite journal_mode being persistent, which + is what four simultaneous workers actually collided on. + """ + with open_database(self.database): + pass # first open sets WAL + + # sqlite3.Connection is an immutable C type, so the statements are captured with the + # connection's own trace callback, attached as it is created. + statements: list[str] = [] + real_connect = sqlite3.connect + + def traced(*args, **kwargs): # type: ignore[no-untyped-def] + connection = real_connect(*args, **kwargs) + connection.set_trace_callback(statements.append) + return connection + + with mock.patch("fintick.storage.sqlite3.connect", traced): + with open_database(self.database): + pass + + writes = [s for s in statements if "journal_mode=" in s.replace(" ", "")] + self.assertEqual(writes, [], f"re-issued the WAL switch unnecessarily: {writes}") + + def test_a_concurrent_open_waits_instead_of_raising(self) -> None: + """Behavioural check: a held write lock delays an open rather than failing it. + + Note this does not by itself prove the fix — Python's 5s connect() default would + also pass a 0.4s hold. The guard for the raised timeout is the pragma test above; + the guard for the real cause is the WAL test above that. + """ + with open_database(self.database): + pass # create the schema first + + holder = sqlite3.connect(self.database) + holder.execute("PRAGMA busy_timeout=30000") + holder.execute("BEGIN IMMEDIATE") # take the write lock + self.addCleanup(holder.close) + + failure: list[BaseException] = [] + + def opener() -> None: + try: + with open_database(self.database): + pass + except BaseException as error: # noqa: BLE001 - recorded for the assertion + failure.append(error) + + thread = threading.Thread(target=opener) + thread.start() + # Hold the lock briefly, then release it from THIS thread — a sqlite3 connection + # may only be used by the thread that created it. + time.sleep(0.4) + holder.rollback() + thread.join(timeout=20) + + self.assertFalse(thread.is_alive(), "the concurrent open never completed") + self.assertEqual(failure, [], f"concurrent open raised instead of waiting: {failure}") + + +if __name__ == "__main__": + unittest.main()