Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 26 additions & 2 deletions fintick/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
100 changes: 100 additions & 0 deletions tests/test_storage_locking.py
Original file line number Diff line number Diff line change
@@ -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()