Spaces:
Running
perf(sidecar): stop blocking the event loop on 2.7GB of local SQLite
Browse filesapp/local_meta.py is synchronous sqlite3 over a 2.7 GB file, and three
call sites invoked it directly from inside `async def`. A blocking call
there stalls EVERY in-flight request, including ones that never touch the
sidecar, and this deploys as a single uvicorn worker on 2 vCPUs — there
is no second worker to absorb it.
Measured on a 300k-row stand-in: a 72 ms query held the event loop for
78 ms; the same query through a thread held it for 7 ms, which is just
scheduler granularity. The real sidecar has 1.8M rows.
Moved to asyncio.to_thread:
- turso_svc.fetch_metadata_batch -> local_meta.fetch_rows
- turso_svc.fetch_trending_by_categories -> local_meta.fetch_trending,
which matters most: it is the single most expensive query in the
system and the one that lands on brand-new users
- space.py _titles_for, now async; all three of its callers were
already coroutines
This is safe by construction rather than by luck: local_meta opens its
connection read-only with check_same_thread=False, with a comment saying
it is so several threads can share the handle.
fts_svc.search_sparse was already correct — it uses run_in_executor — so
the FTS path was never affected.
A test walks app/ with ast and fails if any coroutine calls a blocking
local_meta function directly, plus a positive control so the guard cannot
silently match nothing, plus a timing test asserting a threaded query does
not stall the loop.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
- app/routers/space.py +13 -6
- app/turso_svc.py +16 -4
- tests/test_event_loop_blocking.py +113 -0
|
@@ -18,6 +18,7 @@ rewrite -> encode -> dense+sparse fanout -> RRF -> title boost, and
|
|
| 18 |
arxiv_id. These are thin, honest wrappers over both. Qdrant credentials stay
|
| 19 |
here and are never handed to the edge app.
|
| 20 |
"""
|
|
|
|
| 21 |
import time
|
| 22 |
|
| 23 |
from fastapi import APIRouter, Header, HTTPException, Query
|
|
@@ -46,12 +47,18 @@ def _authorize(authorization: str | None) -> None:
|
|
| 46 |
raise HTTPException(status_code=401, detail="invalid service token")
|
| 47 |
|
| 48 |
|
| 49 |
-
def _titles_for(arxiv_ids: list[str]) -> dict[str, str]:
|
| 50 |
-
"""Best-effort titles from the local sidecar; absent ids simply stay absent.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 51 |
if not arxiv_ids:
|
| 52 |
return {}
|
| 53 |
try:
|
| 54 |
-
rows = local_meta.fetch_rows
|
| 55 |
except Exception as exc: # sidecar is optional, never fatal
|
| 56 |
print(f"[space] title lookup failed ({exc})")
|
| 57 |
return {}
|
|
@@ -91,7 +98,7 @@ async def space_search(
|
|
| 91 |
print(f"[space] search failed ({exc})")
|
| 92 |
raise HTTPException(status_code=502, detail="search unavailable") from exc
|
| 93 |
|
| 94 |
-
titles = _titles_for(list(ids))
|
| 95 |
return {
|
| 96 |
"query": query,
|
| 97 |
"tookMs": round((time.perf_counter() - started) * 1000),
|
|
@@ -188,7 +195,7 @@ async def space_similarity(
|
|
| 188 |
raise HTTPException(status_code=502, detail="vector store unavailable") from exc
|
| 189 |
|
| 190 |
va, vb = vectors.get(left), vectors.get(right)
|
| 191 |
-
titles = _titles_for([left, right])
|
| 192 |
both = va is not None and vb is not None
|
| 193 |
return {
|
| 194 |
"a": {"arxivId": left, "title": titles.get(left, ""), "found": va is not None},
|
|
@@ -233,7 +240,7 @@ async def space_neighbors(
|
|
| 233 |
raise HTTPException(status_code=502, detail="vector store unavailable") from exc
|
| 234 |
|
| 235 |
hits = hits[:limit]
|
| 236 |
-
titles = _titles_for([h["arxiv_id"] for h in hits])
|
| 237 |
return {
|
| 238 |
"arxivId": anchor,
|
| 239 |
"found": True,
|
|
|
|
| 18 |
arxiv_id. These are thin, honest wrappers over both. Qdrant credentials stay
|
| 19 |
here and are never handed to the edge app.
|
| 20 |
"""
|
| 21 |
+
import asyncio
|
| 22 |
import time
|
| 23 |
|
| 24 |
from fastapi import APIRouter, Header, HTTPException, Query
|
|
|
|
| 47 |
raise HTTPException(status_code=401, detail="invalid service token")
|
| 48 |
|
| 49 |
|
| 50 |
+
async def _titles_for(arxiv_ids: list[str]) -> dict[str, str]:
|
| 51 |
+
"""Best-effort titles from the local sidecar; absent ids simply stay absent.
|
| 52 |
+
|
| 53 |
+
Off the event loop: local_meta is synchronous sqlite3 over a 2.7 GB file,
|
| 54 |
+
and blocking here stalls every other in-flight request. Two of the three
|
| 55 |
+
call sites pass a handful of ids, but the batch one (line 94) passes a
|
| 56 |
+
whole page of results.
|
| 57 |
+
"""
|
| 58 |
if not arxiv_ids:
|
| 59 |
return {}
|
| 60 |
try:
|
| 61 |
+
rows = await asyncio.to_thread(local_meta.fetch_rows, arxiv_ids)
|
| 62 |
except Exception as exc: # sidecar is optional, never fatal
|
| 63 |
print(f"[space] title lookup failed ({exc})")
|
| 64 |
return {}
|
|
|
|
| 98 |
print(f"[space] search failed ({exc})")
|
| 99 |
raise HTTPException(status_code=502, detail="search unavailable") from exc
|
| 100 |
|
| 101 |
+
titles = await _titles_for(list(ids))
|
| 102 |
return {
|
| 103 |
"query": query,
|
| 104 |
"tookMs": round((time.perf_counter() - started) * 1000),
|
|
|
|
| 195 |
raise HTTPException(status_code=502, detail="vector store unavailable") from exc
|
| 196 |
|
| 197 |
va, vb = vectors.get(left), vectors.get(right)
|
| 198 |
+
titles = await _titles_for([left, right])
|
| 199 |
both = va is not None and vb is not None
|
| 200 |
return {
|
| 201 |
"a": {"arxivId": left, "title": titles.get(left, ""), "found": va is not None},
|
|
|
|
| 240 |
raise HTTPException(status_code=502, detail="vector store unavailable") from exc
|
| 241 |
|
| 242 |
hits = hits[:limit]
|
| 243 |
+
titles = await _titles_for([h["arxiv_id"] for h in hits])
|
| 244 |
return {
|
| 245 |
"arxivId": anchor,
|
| 246 |
"found": True,
|
|
@@ -13,6 +13,7 @@ Table: papers (arxiv_id UNIQUE INDEX)
|
|
| 13 |
"""
|
| 14 |
from __future__ import annotations
|
| 15 |
|
|
|
|
| 16 |
import json
|
| 17 |
import time
|
| 18 |
from collections import OrderedDict
|
|
@@ -115,7 +116,15 @@ async def fetch_metadata_batch(arxiv_ids: list[str]) -> dict[str, dict]:
|
|
| 115 |
# missing or partially-built sidecar only costs us the speedup.
|
| 116 |
from app import local_meta
|
| 117 |
if local_meta.is_available():
|
| 118 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 119 |
paper = _to_paper_dict(row)
|
| 120 |
if paper:
|
| 121 |
output[paper["arxiv_id"]] = paper
|
|
@@ -369,9 +378,12 @@ async def fetch_trending_by_categories(
|
|
| 369 |
# this becomes an index range read.
|
| 370 |
from app import local_meta
|
| 371 |
if local_meta.is_available():
|
| 372 |
-
|
| 373 |
-
|
| 374 |
-
|
|
|
|
|
|
|
|
|
|
| 375 |
)
|
| 376 |
if rows:
|
| 377 |
papers = [p for p in (_to_paper_dict(r) for r in rows) if p]
|
|
|
|
| 13 |
"""
|
| 14 |
from __future__ import annotations
|
| 15 |
|
| 16 |
+
import asyncio
|
| 17 |
import json
|
| 18 |
import time
|
| 19 |
from collections import OrderedDict
|
|
|
|
| 116 |
# missing or partially-built sidecar only costs us the speedup.
|
| 117 |
from app import local_meta
|
| 118 |
if local_meta.is_available():
|
| 119 |
+
# Off the event loop. local_meta is synchronous sqlite3 against a 2.7 GB
|
| 120 |
+
# file, and a blocking call inside an async handler stalls EVERY other
|
| 121 |
+
# in-flight request, including ones that never touch the sidecar.
|
| 122 |
+
# Measured on a 300k-row stand-in, a 72 ms query held the loop for
|
| 123 |
+
# 78 ms; off-loop that drops to scheduler granularity. The connection is
|
| 124 |
+
# opened read-only with check_same_thread=False precisely so it can be
|
| 125 |
+
# shared across threads (local_meta._probe).
|
| 126 |
+
sidecar_rows = await asyncio.to_thread(local_meta.fetch_rows, misses)
|
| 127 |
+
for row in sidecar_rows:
|
| 128 |
paper = _to_paper_dict(row)
|
| 129 |
if paper:
|
| 130 |
output[paper["arxiv_id"]] = paper
|
|
|
|
| 378 |
# this becomes an index range read.
|
| 379 |
from app import local_meta
|
| 380 |
if local_meta.is_available():
|
| 381 |
+
# Off the event loop — see fetch_metadata_batch. This one matters more:
|
| 382 |
+
# trending is the single most expensive query in the system and the one
|
| 383 |
+
# that lands on brand-new users.
|
| 384 |
+
rows = await asyncio.to_thread(
|
| 385 |
+
local_meta.fetch_trending, set(categories),
|
| 386 |
+
limit, config.TRENDING_RECENCY_MONTHS,
|
| 387 |
)
|
| 388 |
if rows:
|
| 389 |
papers = [p for p in (_to_paper_dict(r) for r in rows) if p]
|
|
@@ -0,0 +1,113 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
"""
|
| 2 |
+
Nothing may block the event loop with sidecar I/O.
|
| 3 |
+
|
| 4 |
+
`app/local_meta.py` is synchronous sqlite3 over a 2.7 GB file. A blocking call
|
| 5 |
+
inside an `async def` stalls EVERY in-flight request, including ones that never
|
| 6 |
+
touch the sidecar — and this runs as a single uvicorn worker on 2 vCPUs, so
|
| 7 |
+
there is no other worker to absorb it.
|
| 8 |
+
|
| 9 |
+
Measured on a 300k-row stand-in: a 72 ms query held the loop for 78 ms; the
|
| 10 |
+
same query via a thread held it for 7 ms. The real sidecar has 1.8M rows.
|
| 11 |
+
|
| 12 |
+
The shared connection is opened read-only with check_same_thread=False
|
| 13 |
+
precisely so it can be used from a thread (local_meta._probe).
|
| 14 |
+
"""
|
| 15 |
+
import ast
|
| 16 |
+
import asyncio
|
| 17 |
+
import pathlib
|
| 18 |
+
import sqlite3
|
| 19 |
+
import tempfile
|
| 20 |
+
import time
|
| 21 |
+
|
| 22 |
+
import pytest
|
| 23 |
+
|
| 24 |
+
# Functions that touch the sidecar's connection and therefore must not be
|
| 25 |
+
# called directly from a coroutine.
|
| 26 |
+
BLOCKING = {"fetch_rows", "fetch_trending", "pub_year_month", "newest_update_date"}
|
| 27 |
+
|
| 28 |
+
# is_available() and stats() are cheap after the first probe and are excluded.
|
| 29 |
+
|
| 30 |
+
|
| 31 |
+
def _async_callers_of_blocking_sidecar_fns():
|
| 32 |
+
"""Every `local_meta.<blocking>()` call lexically inside an async def."""
|
| 33 |
+
offenders = []
|
| 34 |
+
for path in sorted(pathlib.Path("app").rglob("*.py")):
|
| 35 |
+
if path.name == "local_meta.py":
|
| 36 |
+
continue
|
| 37 |
+
tree = ast.parse(path.read_text(), filename=str(path))
|
| 38 |
+
for fn in ast.walk(tree):
|
| 39 |
+
if not isinstance(fn, ast.AsyncFunctionDef):
|
| 40 |
+
continue
|
| 41 |
+
for node in ast.walk(fn):
|
| 42 |
+
if not isinstance(node, ast.Call):
|
| 43 |
+
continue
|
| 44 |
+
f = node.func
|
| 45 |
+
if (isinstance(f, ast.Attribute) and f.attr in BLOCKING
|
| 46 |
+
and isinstance(f.value, ast.Name)
|
| 47 |
+
and f.value.id == "local_meta"):
|
| 48 |
+
offenders.append(f"{path}:{node.lineno} in {fn.name}()")
|
| 49 |
+
return offenders
|
| 50 |
+
|
| 51 |
+
|
| 52 |
+
def test_no_coroutine_calls_the_sidecar_directly():
|
| 53 |
+
offenders = _async_callers_of_blocking_sidecar_fns()
|
| 54 |
+
assert not offenders, (
|
| 55 |
+
"sidecar I/O called directly from a coroutine — wrap in "
|
| 56 |
+
"asyncio.to_thread: " + ", ".join(offenders))
|
| 57 |
+
|
| 58 |
+
|
| 59 |
+
def test_the_check_would_catch_a_direct_call():
|
| 60 |
+
"""Guard against the guard silently matching nothing."""
|
| 61 |
+
src = (
|
| 62 |
+
"import asyncio\n"
|
| 63 |
+
"from app import local_meta\n"
|
| 64 |
+
"async def handler(ids):\n"
|
| 65 |
+
" return local_meta.fetch_rows(ids)\n"
|
| 66 |
+
)
|
| 67 |
+
tree = ast.parse(src)
|
| 68 |
+
hits = [
|
| 69 |
+
n.lineno for fn in ast.walk(tree) if isinstance(fn, ast.AsyncFunctionDef)
|
| 70 |
+
for n in ast.walk(fn)
|
| 71 |
+
if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
|
| 72 |
+
and n.func.attr in BLOCKING
|
| 73 |
+
]
|
| 74 |
+
assert hits == [4]
|
| 75 |
+
|
| 76 |
+
|
| 77 |
+
async def test_a_threaded_query_does_not_stall_the_loop():
|
| 78 |
+
"""The property itself, not just the call shape."""
|
| 79 |
+
with tempfile.TemporaryDirectory() as d:
|
| 80 |
+
path = f"{d}/t.db"
|
| 81 |
+
c = sqlite3.connect(path)
|
| 82 |
+
c.execute("CREATE TABLE papers (arxiv_id TEXT, blob TEXT)")
|
| 83 |
+
c.executemany("INSERT INTO papers VALUES (?,?)",
|
| 84 |
+
[(f"p{i}", "x" * 200) for i in range(120_000)])
|
| 85 |
+
c.commit()
|
| 86 |
+
c.close()
|
| 87 |
+
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True,
|
| 88 |
+
check_same_thread=False)
|
| 89 |
+
|
| 90 |
+
def slow():
|
| 91 |
+
return conn.execute(
|
| 92 |
+
"SELECT COUNT(*) FROM papers WHERE blob LIKE '%zz%'").fetchone()
|
| 93 |
+
|
| 94 |
+
gaps = []
|
| 95 |
+
stop = asyncio.Event()
|
| 96 |
+
|
| 97 |
+
async def heartbeat():
|
| 98 |
+
last = time.perf_counter()
|
| 99 |
+
while not stop.is_set():
|
| 100 |
+
await asyncio.sleep(0.005)
|
| 101 |
+
now = time.perf_counter()
|
| 102 |
+
gaps.append(now - last)
|
| 103 |
+
last = now
|
| 104 |
+
|
| 105 |
+
hb = asyncio.create_task(heartbeat())
|
| 106 |
+
await asyncio.sleep(0.03)
|
| 107 |
+
await asyncio.to_thread(slow)
|
| 108 |
+
stop.set()
|
| 109 |
+
await hb
|
| 110 |
+
conn.close()
|
| 111 |
+
|
| 112 |
+
assert max(gaps) < 0.05, (
|
| 113 |
+
f"loop stalled {max(gaps)*1000:.0f} ms even through a thread")
|