diff --git a/surfsense_local/backend/shared/db.py b/surfsense_local/backend/shared/db.py index 1b4579110e..cbad5d06ca 100644 --- a/surfsense_local/backend/shared/db.py +++ b/surfsense_local/backend/shared/db.py @@ -18,6 +18,8 @@ ) from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker +from shared.sqlite import enable_wal + # SQLite is the only backend that lets constraints stay unnamed, and Alembic's # batch mode cannot drop what it cannot name. Retrofitting this later would not # match the names already on disk, so it has to hold from the first migration. @@ -94,10 +96,11 @@ def _apply_pragmas(dbapi_connection: Any, _record: Any) -> None: dbapi_connection.enable_load_extension(False) cursor = dbapi_connection.cursor() - cursor.execute("PRAGMA journal_mode = WAL") - cursor.execute("PRAGMA foreign_keys = ON") + # First, so what follows waits for a concurrent writer instead of failing. cursor.execute("PRAGMA busy_timeout = 5000") + cursor.execute("PRAGMA foreign_keys = ON") cursor.close() + enable_wal(dbapi_connection) # The API sets this for the span of one request. Waiting for the write lock on diff --git a/surfsense_local/backend/shared/queue.py b/surfsense_local/backend/shared/queue.py index 259ea97a6b..f41d4762e2 100644 --- a/surfsense_local/backend/shared/queue.py +++ b/surfsense_local/backend/shared/queue.py @@ -1,6 +1,10 @@ +import sqlite3 +from contextlib import closing + from huey import SqliteHuey from shared.config import get_storage_settings +from shared.sqlite import enable_wal _settings = get_storage_settings() # SqliteHuey opens the file as it is constructed. @@ -9,6 +13,10 @@ # Its own file: constant polling must not hold the write lock on the database. _QUEUE_FILE = str(_settings.queue_path) +# Before huey connects: its own journal_mode switch cannot wait for the lock. +with closing(sqlite3.connect(_QUEUE_FILE, timeout=5)) as _connection: + enable_wal(_connection) + # One queue per consumer, so an import never queues ahead of a summary. ingest_queue = SqliteHuey(name="ingest", filename=_QUEUE_FILE) studio_queue = SqliteHuey(name="studio", filename=_QUEUE_FILE) diff --git a/surfsense_local/backend/shared/sqlite.py b/surfsense_local/backend/shared/sqlite.py new file mode 100644 index 0000000000..bc05b43c6a --- /dev/null +++ b/surfsense_local/backend/shared/sqlite.py @@ -0,0 +1,22 @@ +import sqlite3 +import time +from typing import Any + +# Switching a fresh file to WAL wants an exclusive lock, and SQLite answers +# SQLITE_BUSY without running the busy handler, so busy_timeout does not cover +# it. ponytail: fixed 5s poll, matching busy_timeout; it only ever waits on +# another process creating a schema. +_ATTEMPTS = 50 +_DELAY = 0.1 + + +def enable_wal(connection: Any, attempts: int = _ATTEMPTS) -> None: + """Put a SQLite file in WAL mode, waiting out a concurrent first opener.""" + for remaining in reversed(range(attempts)): + try: + connection.execute("PRAGMA journal_mode = WAL") + return + except sqlite3.OperationalError: + if not remaining: + raise + time.sleep(_DELAY) diff --git a/surfsense_local/backend/tests/unit/shared/test_sqlite.py b/surfsense_local/backend/tests/unit/shared/test_sqlite.py new file mode 100644 index 0000000000..acc18c625c --- /dev/null +++ b/surfsense_local/backend/tests/unit/shared/test_sqlite.py @@ -0,0 +1,53 @@ +import sqlite3 +import threading +from pathlib import Path + +import pytest + +from shared.sqlite import enable_wal + +pytestmark = pytest.mark.unit + + +def test_wal_waits_for_the_process_still_creating_the_file(tmp_path: Path) -> None: + """A sidecar that opens a fresh file while another is creating its schema + gets SQLITE_BUSY on the switch, and dies on it without this wait.""" + path = tmp_path / "queue.db" + creator = sqlite3.connect(path, isolation_level=None) + creator.execute("CREATE TABLE task (id integer primary key)") + creator.execute("BEGIN IMMEDIATE") + + latecomer = sqlite3.connect(path, timeout=5, check_same_thread=False) + with pytest.raises(sqlite3.OperationalError, match="locked"): + latecomer.execute("PRAGMA journal_mode = WAL") + + switch = threading.Thread(target=enable_wal, args=(latecomer,)) + switch.start() + try: + assert switch.is_alive() + creator.execute("COMMIT") + switch.join(timeout=5) + assert not switch.is_alive() + assert latecomer.execute("PRAGMA journal_mode").fetchone()[0] == "wal" + finally: + switch.join(timeout=5) + creator.close() + latecomer.close() + + +def test_wal_is_a_no_op_once_the_file_is_in_wal(tmp_path: Path) -> None: + """So a later connection may run the pragma while someone else writes.""" + path = tmp_path / "queue.db" + first = sqlite3.connect(path) + enable_wal(first) + holder = sqlite3.connect(path, isolation_level=None) + holder.execute("CREATE TABLE task (id integer primary key)") + holder.execute("BEGIN IMMEDIATE") + + second = sqlite3.connect(path, timeout=0) + try: + enable_wal(second, attempts=1) + finally: + holder.close() + first.close() + second.close() diff --git a/surfsense_local/frontend/src/features/studio/studio-panel.test.tsx b/surfsense_local/frontend/src/features/studio/studio-panel.test.tsx index 9c68463ef3..d8b058fe04 100644 --- a/surfsense_local/frontend/src/features/studio/studio-panel.test.tsx +++ b/surfsense_local/frontend/src/features/studio/studio-panel.test.tsx @@ -53,7 +53,6 @@ function StudioHarness({ onOpen={vi.fn()} onRegenerate={(id) => void studio.regenerate(id)} onDelete={(id) => void studio.remove(id)} - onRetry={(id) => void studio.retry(id)} /> )