fix(db): release every read connection and the owned cache on close

Database.close() closed only the calling thread's read connection, but reads
happen inside asyncio.to_thread workers, so each worker's connection outlived
stop(); the EmbeddingCache connection was never closed at all. A stopped
service now leaves zero open connections (was 3), which matters for the
Recreate rollout and for reopening the same index file.

Ownership rule: whoever creates the cache closes it. build() closes the one it
constructs; a caller-supplied cache survives the service, as the rebuild tests
require.

Also silences the two warning sources: httpx per-request cookies in the UI
tests, and one unclosed in-memory database per Hypothesis example in the FTS
fuzz test (396 of the 410 hidden ResourceWarnings).
This commit is contained in:
Markus committed 2026-09-12 18:53:21 +00:00
1 parent fc7537f2bb
commit 4cb963978d
9 files changed
+102 -32

No files matched your search

Binary file not shown.
+13 -4
View File
@@ -102,6 +102,10 @@ class Database:
self._write: sqlite3.Connection | None = None
self._lock = threading.RLock()
self._local = threading.local()
# read() caches one connection per thread, and the service reads inside
# asyncio.to_thread workers; keep a registry so close() releases them all
# rather than only the caller's.
self._reads: list[sqlite3.Connection] = []
# -- lifecycle ---------------------------------------------------------
def open(self) -> bool:
@@ -167,10 +171,13 @@ class Database:
if self._write is not None:
self._write.close()
self._write = None
conn = getattr(self._local, "read", None)
if conn is not None:
conn.close()
self._local.read = None
for conn in self._reads:
try:
conn.close()
except sqlite3.Error: # already closed by its own thread
pass
self._reads.clear()
self._local.read = None
# -- connections -------------------------------------------------------
def read(self) -> sqlite3.Connection:
@@ -178,6 +185,8 @@ class Database:
if conn is None:
conn = connect(self.path, read_only=True)
self._local.read = conn
with self._lock:
self._reads.append(conn)
return conn
@contextmanager
+4 -1
View File
@@ -84,7 +84,10 @@ class EmbeddingCache:
self.conn.commit()
def close(self) -> None:
self.conn.close()
try:
self.conn.close()
except sqlite3.Error: # pragma: no cover - already closed
pass
class Embedder(ABC):
+12 -2
View File
@@ -73,6 +73,9 @@ class MemoryService:
self.embedder = embedder
self.audit = audit or AuditLog(audit_path_from_env())
self.watcher: VaultWatcher | None = None
# Set by build() when it constructs the cache itself: whoever creates the
# cache closes it, so a caller-supplied one survives the service.
self._owned_cache: EmbeddingCache | None = None
self._embed_retry_task: asyncio.Task[None] | None = None
self.embed_ok = embedder is not None
self._write_lock = asyncio.Lock()
@@ -97,15 +100,19 @@ class MemoryService:
"embedding model or dim changed — vectors dropped, re-embedding",
extra={"fields": {"model": settings.embed_model, "dim": settings.embed_dim}},
)
owned_cache: EmbeddingCache | None = None
if embedder is None:
owned_cache = EmbeddingCache(settings.cache_db)
embedder = BifrostEmbedder(
settings.bifrost_url,
settings.bifrost_vk,
settings.embed_model,
settings.embed_dim,
cache=EmbeddingCache(settings.cache_db),
cache=owned_cache,
)
return cls(settings, db, vault, Indexer(db, vault), embedder)
svc = cls(settings, db, vault, Indexer(db, vault), embedder)
svc._owned_cache = owned_cache
return svc
async def start(self, *, watch: bool = True, embed: bool = True) -> None:
if self.embedder is not None and embed:
@@ -161,6 +168,9 @@ class MemoryService:
await self.watcher.stop()
if self.embedder is not None:
await self.embedder.aclose()
if self._owned_cache is not None:
self._owned_cache.close()
self._owned_cache = None
self.db.close()
# -- indexing ----------------------------------------------------------
+26
View File
@@ -111,3 +111,29 @@ def test_write_is_transactional(tmp_path):
w.execute("INSERT INTO notes(id,path,state,file_hash) VALUES('a','p','active','h')")
w.execute("INSERT INTO notes(id,path,state,file_hash) VALUES('a','p2','active','h')")
assert db.read().execute("SELECT count(*) FROM notes").fetchone()[0] == 0
def test_close_releases_read_connections_from_every_thread(tmp_path):
"""read() caches a connection per thread, and the service reads inside
asyncio.to_thread workers, so closing only the caller's leaks the rest."""
import threading
db = Database(tmp_path / "index.db", dim=4, trigram=False)
db.open()
from_worker: list = []
t = threading.Thread(target=lambda: from_worker.append(db.read()))
t.start()
t.join()
assert from_worker[0].execute("SELECT 1").fetchone()[0] == 1
db.close()
with pytest.raises(sqlite3.ProgrammingError):
from_worker[0].execute("SELECT 1")
def test_close_is_idempotent(tmp_path):
db = Database(tmp_path / "index.db", dim=4, trigram=False)
db.open()
db.read()
db.close()
db.close()
+1 -3
View File
@@ -148,9 +148,7 @@ async def test_matrix_never_mutates_the_configured_live_index(settings, seeded):
)
assert reports
assert settings.index_db.read_bytes() == before
assert svc.db.read().execute(
"SELECT 1 FROM sqlite_master WHERE name='chunks_tri'"
).fetchone()
assert svc.db.read().execute("SELECT 1 FROM sqlite_master WHERE name='chunks_tri'").fetchone()
def test_pick_default_keeps_the_baseline_when_nothing_clears_the_noise():
+9 -6
View File
@@ -58,9 +58,12 @@ def test_trigram_drops_short_tokens():
@given(st.text())
def test_sanitizer_never_produces_invalid_fts_syntax(s):
db = sqlite3.connect(":memory:")
db.execute("CREATE VIRTUAL TABLE t USING fts5(body)")
db.execute("INSERT INTO t(body) VALUES ('hello world')")
for q in (to_match(s), to_trigram_match(s)):
if q is None:
continue
db.execute("SELECT rowid FROM t WHERE t MATCH ?", (q,)).fetchall()
try:
db.execute("CREATE VIRTUAL TABLE t USING fts5(body)")
db.execute("INSERT INTO t(body) VALUES ('hello world')")
for q in (to_match(s), to_trigram_match(s)):
if q is None:
continue
db.execute("SELECT rowid FROM t WHERE t MATCH ?", (q,)).fetchall()
finally:
db.close() # one in-memory db per Hypothesis example adds up
+23
View File
@@ -113,3 +113,26 @@ async def test_embeddings_down_leaves_chunks_pending_and_retries_later(settings)
assert svc.db.read().execute("SELECT count(*) FROM chunks WHERE vec_pending=1").fetchone()[0] == 0
finally:
await svc.stop()
async def test_stop_closes_the_cache_the_service_created(settings):
"""Whoever creates the cache closes it: build() makes one when no embedder is
supplied, and that connection used to outlive stop()."""
import sqlite3
svc = MemoryService.build(settings) # no embedder -> build() owns the cache
cache = svc._owned_cache
assert cache is not None
await svc.start(watch=False, embed=False) # no network: skip the embed self-test
await svc.stop()
with pytest.raises(sqlite3.ProgrammingError):
cache.conn.execute("SELECT 1")
async def test_a_caller_supplied_cache_survives_the_service(settings, cache):
"""The counterpart rule: a cache handed in from outside is not the service's
to close, so it can be shared across service lifecycles (rebuild tests do)."""
svc = MemoryService.build(settings, embedder=FakeEmbedder(settings.embed_dim, cache=cache))
await svc.start(watch=False)
await svc.stop()
assert cache.conn.execute("SELECT 1").fetchone() == (1,)
+14 -16
View File
@@ -174,9 +174,12 @@ async def app_and_svc(auth_settings):
await svc.stop()
def _client(app):
def _client(app, cookie: str | None = None):
"""httpx deprecates per-request cookies=..., so the session goes on the client."""
return httpx.AsyncClient(
transport=httpx.ASGITransport(app=app), base_url="https://memory.cloud.traunseenet.com"
transport=httpx.ASGITransport(app=app),
base_url="https://memory.cloud.traunseenet.com",
cookies={SESSION_COOKIE: cookie} if cookie else None,
)
@@ -192,8 +195,8 @@ async def test_a_valid_session_gets_in(app_and_svc):
app, svc = app_and_svc
codec = SessionCodec(svc.settings.session_secret, 12)
cookie = codec.issue(Principal(sub="u1", name="Markus", email=""))
async with _client(app) as c:
r = await c.get("/ui/inbox", cookies={SESSION_COOKIE: cookie})
async with _client(app, cookie) as c:
r = await c.get("/ui/inbox")
assert r.status_code == 200 and "Posteingang" in r.text
@@ -225,8 +228,8 @@ async def test_readyz_is_green_while_embeddings_are_down(app_and_svc):
async def test_security_headers_are_set(app_and_svc):
app, svc = app_and_svc
cookie = SessionCodec(svc.settings.session_secret, 12).issue(Principal("u1", "M", ""))
async with _client(app) as c:
r = await c.get("/ui/inbox", cookies={SESSION_COOKIE: cookie})
async with _client(app, cookie) as c:
r = await c.get("/ui/inbox")
assert r.headers["content-security-policy"].startswith("default-src 'self'")
assert r.headers["x-frame-options"] == "DENY"
@@ -235,11 +238,10 @@ async def test_post_from_a_foreign_origin_is_refused(app_and_svc):
app, svc = app_and_svc
cookie = SessionCodec(svc.settings.session_secret, 12).issue(Principal("u1", "M", ""))
res = await svc.write_note(title="T", body="b", type="note")
async with _client(app) as c:
async with _client(app, cookie) as c:
r = await c.post(
f"/ui/n/{res['id']}/approve",
headers={"origin": "https://evil.example.com"},
cookies={SESSION_COOKIE: cookie},
data={},
)
assert r.status_code == 403
@@ -250,11 +252,10 @@ async def test_post_from_the_configured_origin_is_allowed(app_and_svc):
app, svc = app_and_svc
cookie = SessionCodec(svc.settings.session_secret, 12).issue(Principal("u1", "M", ""))
res = await svc.write_note(title="T", body="b", type="note")
async with _client(app) as c:
async with _client(app, cookie) as c:
r = await c.post(
f"/ui/n/{res['id']}/approve",
headers={"origin": "https://memory.cloud.traunseenet.com"},
cookies={SESSION_COOKIE: cookie},
data={},
)
assert r.status_code in (303, 302)
@@ -274,10 +275,9 @@ async def test_post_without_origin_or_referer_is_refused(app_and_svc):
app, svc = app_and_svc
cookie = SessionCodec(svc.settings.session_secret, 12).issue(Principal("u1", "M", ""))
res = await svc.write_note(title="T", body="b", type="note")
async with _client(app) as c:
async with _client(app, cookie) as c:
r = await c.post(
f"/ui/n/{res['id']}/approve",
cookies={SESSION_COOKIE: cookie},
data={},
)
assert r.status_code == 403
@@ -290,11 +290,10 @@ async def test_post_with_a_same_origin_referer_and_no_origin_header_is_allowed(a
app, svc = app_and_svc
cookie = SessionCodec(svc.settings.session_secret, 12).issue(Principal("u1", "M", ""))
res = await svc.write_note(title="T", body="b", type="note")
async with _client(app) as c:
async with _client(app, cookie) as c:
r = await c.post(
f"/ui/n/{res['id']}/approve",
headers={"referer": "https://memory.cloud.traunseenet.com/ui/n/xyz/edit"},
cookies={SESSION_COOKIE: cookie},
data={},
)
assert r.status_code in (303, 302)
@@ -305,11 +304,10 @@ async def test_post_with_a_foreign_referer_and_no_origin_is_refused(app_and_svc)
app, svc = app_and_svc
cookie = SessionCodec(svc.settings.session_secret, 12).issue(Principal("u1", "M", ""))
res = await svc.write_note(title="T", body="b", type="note")
async with _client(app) as c:
async with _client(app, cookie) as c:
r = await c.post(
f"/ui/n/{res['id']}/approve",
headers={"referer": "https://evil.example.com/attack"},
cookies={SESSION_COOKIE: cookie},
data={},
)
assert r.status_code == 403