fix(storage/database): reopen connection on lifespan restart

Database opened its sqlite3 connection eagerly in __init__ and closed it
in close(); the lifespan called close() on shutdown. In production this
is fine — the lifespan runs once per process. Under pytest the module-
level ``db`` singleton survives across every TestClient session, so the
second test file's lifespan startup hit
``sqlite3.ProgrammingError: Cannot operate on a closed database`` at
fixture-setup time (AutoBackupEngine.__init__ → db.get_setting("…")
was the first reader). 65 spurious "errors" on a full Windows pytest run.

- Database: extract _open() from __init__, add ensure_open() that
  reopens iff _conn is None, and have close() null _conn after the
  TRUNCATE checkpoint so re-close is idempotent.
- main.py lifespan startup: call db.ensure_open() before any setting
  read, so subsequent TestClient sessions get a live connection.
- tests/storage/test_database_reopen.py: pin the four invariants —
  close→ensure_open round-trips data, ensure_open is a no-op when
  open, close is idempotent, and using the DB after close without
  ensure_open raises (callers must opt in).

Full backend suite: 1551 pass / 1 skip / 0 errors. Ruff clean.
This commit is contained in:
2026-05-26 00:26:36 +03:00
parent f6486f9b34
commit f591e258f7
3 changed files with 95 additions and 1 deletions
+6
View File
@@ -220,6 +220,12 @@ async def lifespan(app: FastAPI):
Handles startup and shutdown events.
"""
# Startup
# Reopen the DB if a previous lifespan cycle closed it. No-op in
# production (lifespan runs once per process) but required under
# pytest where the module-level ``db`` singleton outlives multiple
# TestClient sessions and would otherwise be stuck on a closed
# connection.
db.ensure_open()
logger.info(f"Starting LED Grab v{__version__}")
logger.info(f"Python version: {sys.version}")
logger.info(f"Server listening on {config.server.host}:{config.server.port}")
+27 -1
View File
@@ -83,6 +83,18 @@ class Database:
def __init__(self, db_path: str | Path):
self._path = Path(db_path)
self._path.parent.mkdir(parents=True, exist_ok=True)
self._lock = threading.RLock()
self._conn: sqlite3.Connection | None = None
self._open()
def _open(self) -> None:
"""Open the SQLite connection, apply PRAGMAs, and ensure schema.
Split out from ``__init__`` so :meth:`ensure_open` can re-establish
the connection after a previous :meth:`close` — needed under pytest
where the same module-level ``db`` singleton survives across multiple
FastAPI lifespan cycles.
"""
self._conn = sqlite3.connect(
str(self._path),
check_same_thread=False,
@@ -98,11 +110,22 @@ class Database:
# window of unsynced data stays small even if close() is skipped.
self._conn.execute("PRAGMA wal_autocheckpoint=100")
self._conn.execute("PRAGMA busy_timeout=5000")
self._lock = threading.RLock()
self._ensure_schema()
logger.info(f"Database opened: {self._path}")
def ensure_open(self) -> None:
"""Reopen the connection if a prior :meth:`close` left it released.
Idempotent — a no-op when the connection is already live. Production
only calls this once per process (lifespan startup); under pytest a
single ``Database`` instance can outlive multiple TestClient
lifespans, each of which closes the connection on shutdown.
"""
with self._lock:
if self._conn is None:
self._open()
# -- Schema management ---------------------------------------------------
def _ensure_schema(self) -> None:
@@ -352,9 +375,12 @@ class Database:
loses the WAL between graceful app shutdown and a later PC shutdown.
"""
with self._lock:
if self._conn is None:
return
try:
self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
except sqlite3.Error as e:
logger.warning("WAL checkpoint on close failed: %s", e)
self._conn.close()
self._conn = None
logger.info("Database connection closed")
@@ -0,0 +1,62 @@
"""Regression coverage for the lifespan reopen path on ``Database``.
The production server only runs a single FastAPI lifespan per process, so
``Database.close()`` releasing the connection is fine. Pytest is different:
``ledgrab.main`` is imported once and its module-level ``db`` singleton
survives across every TestClient session, each of which closes the
connection on shutdown.
Before :meth:`Database.ensure_open` existed, the second TestClient session
hit ``sqlite3.ProgrammingError: Cannot operate on a closed database`` at
fixture-setup time when the lifespan startup tried to read settings (the
``AutoBackupEngine`` constructor was the first reader). These tests pin
the close+reopen cycle so the cascade can't silently come back.
"""
import sqlite3
import pytest
from ledgrab.storage.database import Database
def test_close_then_ensure_open_reopens_connection(tmp_path):
db = Database(tmp_path / "reopen.db")
db.set_setting("k", {"v": 1})
db.close()
db.ensure_open()
assert db.get_setting("k") == {"v": 1}
db.close()
def test_ensure_open_is_idempotent_when_already_open(tmp_path):
db = Database(tmp_path / "idempotent.db")
original_conn = db._conn
db.ensure_open()
# Same live connection — no spurious reconnect when already open.
assert db._conn is original_conn
db.close()
def test_close_is_idempotent(tmp_path):
db = Database(tmp_path / "double_close.db")
db.close()
# Second close must not raise (lifespan shutdown can run twice in
# quick test sessions). Connection stays released.
db.close()
assert db._conn is None
def test_operation_after_close_without_reopen_raises(tmp_path):
db = Database(tmp_path / "no_reopen.db")
db.close()
# Without ensure_open, attempting to use the DB fails loudly rather
# than silently re-connecting — callers must opt in.
with pytest.raises((sqlite3.ProgrammingError, AttributeError)):
db.get_setting("anything")