From 861d7629e1f9373a3a119f292145f3e5ca980ea3 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 21:47:34 -0500 Subject: [PATCH 1/2] perf: offload blocking SQLite I/O off the event loop sqlite3.connect(), cursor.execute()/fetchall() and connection.close() all ran directly on the event loop. That froze the loop for the entire duration of every query and silently defeated execute_batch_queries' semaphore + gather concurrency, since nothing underneath ever yielded. Measured on a 400k-row table with 16 batched queries: the batch took 230.0ms against a computed serial floor of 230.5ms, and a 5ms heartbeat task got zero ticks. Offloaded, the same batch takes 64.9ms (3.55x) and the heartbeat reaches 83% of its ideal tick count. check_same_thread=False is required rather than incidental: the connection is now created on a worker thread but is still used from the loop thread by initialize_database_optimization(). The asyncpg path is untouched. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/services/database_optimizer.py | 52 ++++- tests/unit/test_database_optimizer.py | 212 +++++++++++++++++- 2 files changed, 252 insertions(+), 12 deletions(-) diff --git a/src/youtube_extension/backend/services/database_optimizer.py b/src/youtube_extension/backend/services/database_optimizer.py index 1db349f0b..03215bc6c 100644 --- a/src/youtube_extension/backend/services/database_optimizer.py +++ b/src/youtube_extension/backend/services/database_optimizer.py @@ -186,7 +186,12 @@ async def initialize(self): else: # SQLite or other - prepare SQLite path if self._sqlite_path and self._sqlite_path != ":memory:": - os.makedirs(os.path.dirname(self._sqlite_path), exist_ok=True) + # Filesystem metadata calls block; keep them off the loop. + await asyncio.to_thread( + os.makedirs, + os.path.dirname(self._sqlite_path), + exist_ok=True, + ) logger.info(f"✅ SQLite database configured at {self._sqlite_path}") else: logger.info("✅ SQLite in-memory database configured") @@ -203,8 +208,23 @@ async def get_connection(self): if self.pool: connection = await self.pool.acquire() else: - # SQLite fallback (file or memory) - connection = sqlite3.connect(self._sqlite_path or ":memory:") + # SQLite fallback (file or memory). sqlite3.connect performs + # blocking filesystem work, so it runs on a worker thread. + # + # check_same_thread=False is required, not optional: the + # connection is created on a worker thread but is subsequently + # used from the event loop thread and from other worker threads + # (see QueryOptimizer.execute_query). Without it sqlite3 raises + # ProgrammingError on the first cross-thread use. + # + # This is safe because a connection is owned by exactly one + # caller between get_connection() and release_connection(), so + # it is never touched by two threads at the same time. + connection = await asyncio.to_thread( + sqlite3.connect, + self._sqlite_path or ":memory:", + check_same_thread=False, + ) connection_time = (time.time() - start_time) * 1000 @@ -229,7 +249,9 @@ async def release_connection(self, connection): if self.pool and hasattr(self.pool, "release"): await self.pool.release(connection) elif hasattr(connection, "close"): - connection.close() + # Closing a SQLite connection flushes and releases the file + # handle; keep that blocking work off the event loop. + await asyncio.to_thread(connection.close) with self._lock: self.pool_stats["connections_in_use"] = max( @@ -367,13 +389,21 @@ async def execute_query( else: result = await connection.fetch(query) elif hasattr(connection, "execute"): - # SQLite or other - cursor = connection.cursor() - if params: - cursor.execute(query, params) - else: - cursor.execute(query) - result = cursor.fetchall() + # SQLite or other DB-API connection. cursor.execute() and + # fetchall() are blocking calls that hold the GIL only while + # not waiting on I/O, so running them on a worker thread lets + # the event loop keep serving other work — and lets + # execute_batch_queries' gather actually overlap queries + # instead of running them back to back. + def _run_sync_query() -> Any: + cursor = connection.cursor() + if params: + cursor.execute(query, params) + else: + cursor.execute(query) + return cursor.fetchall() + + result = await asyncio.to_thread(_run_sync_query) else: raise Exception(f"Unsupported connection type: {type(connection)}") diff --git a/tests/unit/test_database_optimizer.py b/tests/unit/test_database_optimizer.py index 4cf7f7483..89df2eb3f 100644 --- a/tests/unit/test_database_optimizer.py +++ b/tests/unit/test_database_optimizer.py @@ -4,11 +4,15 @@ from __future__ import annotations import asyncio +import contextlib +import sqlite3 import sys +import threading +import time from collections import deque from datetime import datetime, timezone from pathlib import Path -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -1345,3 +1349,209 @@ async def test_haspg_false_uses_sqlite_path(self, tmp_path) -> None: await pool.initialize() finally: _mod.HAS_POSTGRESQL = orig + + +# =========================================================================== +# Event-loop offloading of blocking SQLite I/O +# +# The SQLite branch of DatabaseConnectionPool/QueryOptimizer used to run +# sqlite3.connect(), cursor.execute()/fetchall() and connection.close() +# directly on the event loop. That froze the loop for the whole duration of +# every query and silently defeated execute_batch_queries' semaphore+gather +# concurrency, because nothing underneath ever yielded. +# +# These tests assert on *thread identity* rather than wall-clock timings so +# they stay deterministic on loaded CI hosts. +# =========================================================================== + + +class _RecordingCursor: + """DB-API cursor stub that records which thread ran the query.""" + + def __init__(self, sink: dict, delay: float) -> None: + self._sink = sink + self._delay = delay + + def execute(self, query, params=None): # noqa: ARG002 + self._sink.setdefault("execute_tids", []).append(threading.get_ident()) + if self._delay: + time.sleep(self._delay) + return self + + def fetchall(self): + return [("row",)] + + +class _RecordingConnection: + """Minimal DB-API connection stub (has .execute, so no .fetch/asyncpg).""" + + def __init__(self, sink: dict, delay: float = 0.0) -> None: + self._sink = sink + self._delay = delay + + def cursor(self): + return _RecordingCursor(self._sink, self._delay) + + def execute(self, query, params=None): # pragma: no cover - branch selector + return self.cursor().execute(query, params) + + def close(self): + self._sink.setdefault("close_tids", []).append(threading.get_ident()) + + +class TestSqliteQueryOffloadedToThread: + @pytest.mark.asyncio + async def test_cursor_work_runs_off_the_event_loop_thread(self) -> None: + """The query itself must not execute on the loop thread.""" + sink: dict = {} + pool = MagicMock() + pool.get_connection = AsyncMock(return_value=_RecordingConnection(sink)) + pool.release_connection = AsyncMock() + optimizer = QueryOptimizer(pool) + + loop_tid = threading.get_ident() + await optimizer.execute_query("SELECT 1", use_cache=False) + + assert sink["execute_tids"], "query never executed" + assert loop_tid not in sink["execute_tids"], ( + "cursor.execute ran on the event loop thread; it must be offloaded " + "via asyncio.to_thread so the loop stays responsive" + ) + + @pytest.mark.asyncio + async def test_event_loop_stays_responsive_during_query(self) -> None: + """A concurrent task must still get scheduled while a query runs.""" + sink: dict = {} + pool = MagicMock() + pool.get_connection = AsyncMock( + return_value=_RecordingConnection(sink, delay=0.25) + ) + pool.release_connection = AsyncMock() + optimizer = QueryOptimizer(pool) + + ticks = 0 + stop = False + + async def heartbeat() -> None: + nonlocal ticks + while not stop: + await asyncio.sleep(0.005) + ticks += 1 + + hb = asyncio.create_task(heartbeat()) + await asyncio.sleep(0) # let the heartbeat start + await optimizer.execute_query("SELECT 1", use_cache=False) + stop = True + hb.cancel() + with contextlib.suppress(asyncio.CancelledError): + await hb + + assert ticks > 0, ( + "the event loop got zero ticks during a 250ms query — it was " + "blocked for the entire query duration" + ) + + +class TestSqliteConnectionLifecycleOffloaded: + @pytest.mark.asyncio + async def test_connect_runs_off_the_event_loop_thread(self, tmp_path) -> None: + pool = DatabaseConnectionPool(f"sqlite:///{tmp_path / 'offload.db'}") + await pool.initialize() + + seen: list[int] = [] + real_connect = sqlite3.connect + + def spy(*args, **kwargs): + seen.append(threading.get_ident()) + return real_connect(*args, **kwargs) + + with patch.object(sqlite3, "connect", spy): + conn = await pool.get_connection() + + assert seen, "sqlite3.connect was never called" + assert threading.get_ident() not in seen, ( + "sqlite3.connect ran on the event loop thread" + ) + await pool.release_connection(conn) + + @pytest.mark.asyncio + async def test_connection_is_usable_from_the_event_loop_thread( + self, tmp_path + ) -> None: + """Regression guard for check_same_thread=False. + + The connection is created on a worker thread but + initialize_database_optimization() uses it directly from the loop + thread. Without check_same_thread=False sqlite3 raises + ProgrammingError on that first cross-thread use. + """ + pool = DatabaseConnectionPool(f"sqlite:///{tmp_path / 'xthread.db'}") + await pool.initialize() + conn = await pool.get_connection() + try: + cur = conn.cursor() + cur.execute("CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY)") + cur.execute("INSERT INTO t (id) VALUES (1)") + cur.execute("SELECT id FROM t") + assert cur.fetchall() == [(1,)] + finally: + await pool.release_connection(conn) + + @pytest.mark.asyncio + async def test_close_runs_off_the_event_loop_thread(self) -> None: + sink: dict = {} + pool = DatabaseConnectionPool("sqlite:///:memory:") + await pool.release_connection(_RecordingConnection(sink)) + + assert sink["close_tids"], "connection.close() was never called" + assert threading.get_ident() not in sink["close_tids"], ( + "connection.close() ran on the event loop thread" + ) + + +class TestBatchQueriesActuallyOverlap: + @pytest.mark.asyncio + async def test_batch_queries_are_not_serialised(self) -> None: + """execute_batch_queries documents concurrency; prove it is real. + + With blocking cursor work the batch degenerates to N * delay because + nothing yields. Offloaded, the queries overlap across worker threads. + """ + sink: dict = {} + delay = 0.10 + count = 6 + pool = MagicMock() + pool.get_connection = AsyncMock( + side_effect=lambda: _RecordingConnection(sink, delay=delay) + ) + pool.release_connection = AsyncMock() + optimizer = QueryOptimizer(pool) + + queries = [{"query": f"SELECT {i}", "use_cache": False} for i in range(count)] + started = time.perf_counter() + results = await optimizer.execute_batch_queries(queries) + elapsed = time.perf_counter() - started + + assert len(results) == count + serial_floor = delay * count + assert elapsed < serial_floor / 2, ( + f"batch took {elapsed:.3f}s against a serial floor of " + f"{serial_floor:.3f}s — the queries did not overlap" + ) + + @pytest.mark.asyncio + async def test_batch_queries_use_distinct_threads(self) -> None: + sink: dict = {} + pool = MagicMock() + pool.get_connection = AsyncMock( + side_effect=lambda: _RecordingConnection(sink, delay=0.05) + ) + pool.release_connection = AsyncMock() + optimizer = QueryOptimizer(pool) + + queries = [{"query": f"SELECT {i}", "use_cache": False} for i in range(4)] + await optimizer.execute_batch_queries(queries) + + assert len(set(sink["execute_tids"])) > 1, ( + "every query ran on the same thread — they were serialised" + ) From 7c4dc0931706b2541739654dcdbed832626a2c9c Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 22:01:30 -0500 Subject: [PATCH 2/2] fix: drain sqlite worker before releasing connection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cancelling a query in execute_batch_queries unwound the await but left the worker thread running cursor.execute(). CancelledError is a BaseException, so `except Exception` did not catch it and `finally` released — and therefore closed — the connection on a second thread while the first was still using it. check_same_thread=False disables sqlite3's check, not the single-owner requirement. Run the offload as a shielded task and drain it in `finally` before release. Also fixes two batch tests that passed dicts where the API takes (query, params) tuples, so they never exercised the real contract. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/services/database_optimizer.py | 29 ++++- tests/unit/test_database_optimizer.py | 114 ++++++++++++++++-- 2 files changed, 133 insertions(+), 10 deletions(-) diff --git a/src/youtube_extension/backend/services/database_optimizer.py b/src/youtube_extension/backend/services/database_optimizer.py index 03215bc6c..d901df173 100644 --- a/src/youtube_extension/backend/services/database_optimizer.py +++ b/src/youtube_extension/backend/services/database_optimizer.py @@ -18,6 +18,7 @@ """ import asyncio +import contextlib import hashlib import json import logging @@ -379,6 +380,7 @@ async def execute_query( # Execute query connection = None + sync_worker: asyncio.Future[Any] | None = None try: connection = await self.connection_pool.get_connection() @@ -403,7 +405,18 @@ def _run_sync_query() -> Any: cursor.execute(query) return cursor.fetchall() - result = await asyncio.to_thread(_run_sync_query) + # Run the blocking work as a *task* and await it shielded. A + # worker thread cannot be cancelled: if this coroutine is + # cancelled (execute_batch_queries does exactly that when a + # sibling query fails) a bare `await asyncio.to_thread(...)` + # would unwind here while the thread keeps using `connection`, + # and the `finally` below would then close that connection on a + # second thread. The shield keeps the worker running and the + # drain in `finally` waits for it before any release/close. + sync_worker = asyncio.ensure_future( + asyncio.to_thread(_run_sync_query) + ) + result = await asyncio.shield(sync_worker) else: raise Exception(f"Unsupported connection type: {type(connection)}") @@ -443,6 +456,20 @@ def _run_sync_query() -> Any: raise finally: + # A cancellation may have unwound the await above while the worker + # thread is still running _run_sync_query against `connection`. + # Drain it before releasing: release_connection() closes the + # connection on *another* thread, and check_same_thread=False + # disables sqlite3's same-thread *check*, not the underlying + # requirement that operations on one connection never overlap. + # Each shielded await can itself be cancelled, so loop until the + # worker has genuinely finished (it always does — it is bounded by + # the query, not by us). + if sync_worker is not None: + while not sync_worker.done(): + with contextlib.suppress(BaseException): + await asyncio.shield(sync_worker) + if connection: await self.connection_pool.release_connection(connection) diff --git a/tests/unit/test_database_optimizer.py b/tests/unit/test_database_optimizer.py index 89df2eb3f..0109d615b 100644 --- a/tests/unit/test_database_optimizer.py +++ b/tests/unit/test_database_optimizer.py @@ -1351,6 +1351,9 @@ async def test_haspg_false_uses_sqlite_path(self, tmp_path) -> None: _mod.HAS_POSTGRESQL = orig +_DBOPT = "youtube_extension.backend.services.database_optimizer" + + # =========================================================================== # Event-loop offloading of blocking SQLite I/O # @@ -1368,14 +1371,36 @@ async def test_haspg_false_uses_sqlite_path(self, tmp_path) -> None: class _RecordingCursor: """DB-API cursor stub that records which thread ran the query.""" - def __init__(self, sink: dict, delay: float) -> None: + def __init__(self, sink: dict, delay: float, fail: bool = False, owner=None) -> None: self._sink = sink self._delay = delay + self._fail = fail + self._owner = owner def execute(self, query, params=None): # noqa: ARG002 + lock = self._sink.setdefault("lock", threading.Lock()) self._sink.setdefault("execute_tids", []).append(threading.get_ident()) - if self._delay: - time.sleep(self._delay) + if self._fail: + raise RuntimeError("boom") + owner = self._owner + if owner is not None: + with owner._lock: + owner._inflight += 1 + with lock: + inflight = self._sink.get("inflight", 0) + 1 + self._sink["inflight"] = inflight + self._sink["max_inflight"] = max( + self._sink.get("max_inflight", 0), inflight + ) + try: + if self._delay: + time.sleep(self._delay) + finally: + with lock: + self._sink["inflight"] -= 1 + if owner is not None: + with owner._lock: + owner._inflight -= 1 return self def fetchall(self): @@ -1385,17 +1410,25 @@ def fetchall(self): class _RecordingConnection: """Minimal DB-API connection stub (has .execute, so no .fetch/asyncpg).""" - def __init__(self, sink: dict, delay: float = 0.0) -> None: + def __init__(self, sink: dict, delay: float = 0.0, fail: bool = False) -> None: self._sink = sink self._delay = delay + self._fail = fail + self._lock = threading.Lock() + self._inflight = 0 def cursor(self): - return _RecordingCursor(self._sink, self._delay) + return _RecordingCursor(self._sink, self._delay, self._fail, owner=self) def execute(self, query, params=None): # pragma: no cover - branch selector return self.cursor().execute(query, params) def close(self): + # Only this connection's own in-flight work matters: closing connection + # A while connection B is busy is perfectly legal. + with self._lock: + if self._inflight > 0: + self._sink["close_overlapped_query"] = True self._sink.setdefault("close_tids", []).append(threading.get_ident()) @@ -1527,12 +1560,21 @@ async def test_batch_queries_are_not_serialised(self) -> None: pool.release_connection = AsyncMock() optimizer = QueryOptimizer(pool) - queries = [{"query": f"SELECT {i}", "use_cache": False} for i in range(count)] + queries = [(f"SELECT {i}", ()) for i in range(count)] started = time.perf_counter() - results = await optimizer.execute_batch_queries(queries) + with ( + patch(f"{_DBOPT}.cache_get", AsyncMock(return_value=None)), + patch(f"{_DBOPT}.cache_set", AsyncMock()), + ): + results = await optimizer.execute_batch_queries(queries) elapsed = time.perf_counter() - started assert len(results) == count + assert sink["execute_tids"], "no query ever executed" + assert sink.get("max_inflight", 0) > 1, ( + "never more than one query in flight at a time — the batch " + "serialised despite gather()" + ) serial_floor = delay * count assert elapsed < serial_floor / 2, ( f"batch took {elapsed:.3f}s against a serial floor of " @@ -1549,9 +1591,63 @@ async def test_batch_queries_use_distinct_threads(self) -> None: pool.release_connection = AsyncMock() optimizer = QueryOptimizer(pool) - queries = [{"query": f"SELECT {i}", "use_cache": False} for i in range(4)] - await optimizer.execute_batch_queries(queries) + queries = [(f"SELECT {i}", ()) for i in range(4)] + with ( + patch(f"{_DBOPT}.cache_get", AsyncMock(return_value=None)), + patch(f"{_DBOPT}.cache_set", AsyncMock()), + ): + await optimizer.execute_batch_queries(queries) assert len(set(sink["execute_tids"])) > 1, ( "every query ran on the same thread — they were serialised" ) + + +class TestBatchCancellationDoesNotCloseConnectionMidQuery: + """A cancelled query must not have its connection closed under it. + + ``execute_batch_queries`` cancels in-flight siblings when one query fails. + A worker thread cannot be cancelled, so the offloaded ``cursor.execute()`` + keeps running; if ``execute_query``'s ``finally`` releases (and therefore + closes) the connection straight away, two threads touch one sqlite3 + connection at once. ``check_same_thread=False`` disables the *check*, not + the requirement, so this is real corruption risk — and it only became + reachable once the query was offloaded. + """ + + @pytest.mark.asyncio + async def test_close_never_overlaps_an_in_flight_query(self) -> None: + sink: dict = {} + # First query fails immediately; second is slow and will be cancelled. + conns = [ + _RecordingConnection(sink, fail=True), + _RecordingConnection(sink, delay=0.20), + ] + pool = MagicMock() + pool.get_connection = AsyncMock(side_effect=conns) + + async def _release(conn): + # Mirror DatabaseConnectionPool.release_connection: close off-loop. + await asyncio.to_thread(conn.close) + + pool.release_connection = AsyncMock(side_effect=_release) + optimizer = QueryOptimizer(pool) + + with ( + patch(f"{_DBOPT}.cache_get", AsyncMock(return_value=None)), + patch(f"{_DBOPT}.cache_set", AsyncMock()), + contextlib.suppress(RuntimeError), + ): + await optimizer.execute_batch_queries( + [("SELECT 0", ()), ("SELECT 1", ())] + ) + + # Give any improperly-detached worker thread time to finish so the + # overlap flag is definitely observable either way. + await asyncio.sleep(0.35) + + assert sink.get("close_overlapped_query") is not True, ( + "connection.close() ran while a worker thread was still executing " + "a query on that same connection — the cancellation path must " + "drain the offloaded query before releasing the connection" + )