diff --git a/plugins/memory/holographic/__init__.py b/plugins/memory/holographic/__init__.py index fa8dae97618..7e2e387902e 100644 --- a/plugins/memory/holographic/__init__.py +++ b/plugins/memory/holographic/__init__.py @@ -251,12 +251,12 @@ class HolographicMemoryProvider(MemoryProvider): logger.debug("Holographic memory_write mirror failed: %s", e) def shutdown(self) -> None: - # Close the SQLite connection deterministically instead of leaking it - # to GC. MemoryStore opens its connection with check_same_thread=False - # (store.py), so without an explicit close() the sqlite3.Connection's - # fd is released by refcount/GC at a non-deterministic time on a - # non-deterministic thread, churning a DB fd through the kernel's free - # pool on every session teardown. close() already exists and is cheap. + # Release the shared SQLite connection deterministically on the + # caller's thread. Dropping the reference alone leaves fd finalization + # to GC, which keeps the connection (and its write lock) alive on a + # long-running gateway and prolongs the "database is locked" contention + # this store's shared-connection refcounting is meant to eliminate. + # close() is idempotent and refcount-guarded, so siblings stay safe. if self._store is not None: try: self._store.close() diff --git a/plugins/memory/holographic/store.py b/plugins/memory/holographic/store.py index 7ffc7db0f19..a521b865d6b 100644 --- a/plugins/memory/holographic/store.py +++ b/plugins/memory/holographic/store.py @@ -98,6 +98,21 @@ def _clamp_trust(value: float) -> float: class MemoryStore: """SQLite-backed fact store with entity resolution and trust scoring.""" + # --- Process-wide shared connection registry ------------------------- + # SQLite permits only one writer at a time. Each MemoryStore instance used + # to open its own connection guarded by its own RLock, so the several + # providers that coexist in one process (the main agent plus every + # delegate_task subagent) raced as independent WAL writers. Combined with + # writes that were not rolled back on error, one connection could leave an + # open write transaction that pinned the write lock and made every other + # connection's write fail with "database is locked" for the full busy + # timeout. All instances for the same database now share ONE connection and + # ONE re-entrant lock, so access is fully serialized and cross-connection + # contention is impossible. The shared connection is refcounted, so closing + # one instance never tears the connection out from under a live sibling. + _shared: dict = {} + _shared_guard = threading.Lock() + def __init__( self, db_path: "str | Path | None" = None, @@ -112,14 +127,35 @@ class MemoryStore: self.default_trust = _clamp_trust(default_trust) self.hrr_dim = hrr_dim self._hrr_available = hrr._HAS_NUMPY - self._conn: sqlite3.Connection = sqlite3.connect( - str(self.db_path), - check_same_thread=False, - timeout=10.0, - ) - self._lock = threading.RLock() - self._conn.row_factory = sqlite3.Row - self._init_db() + + # Acquire (or open) the process-wide shared connection for this DB. + self._key = str(self.db_path) + with MemoryStore._shared_guard: + entry = MemoryStore._shared.get(self._key) + if entry is None: + conn = sqlite3.connect( + self._key, + check_same_thread=False, + timeout=10.0, + # Autocommit: every statement is its own transaction, so a + # write that raises mid-method can never leave a dangling + # transaction (and its write lock) open. The explicit + # commit() calls below become harmless no-ops. + isolation_level=None, + ) + conn.row_factory = sqlite3.Row + entry = {"conn": conn, "lock": threading.RLock(), "refs": 0, "ready": False} + MemoryStore._shared[self._key] = entry + entry["refs"] += 1 + self._entry = entry + self._conn = entry["conn"] + self._lock = entry["lock"] + + # Initialise the schema once per shared connection. + with self._lock: + if not self._entry["ready"]: + self._init_db() + self._entry["ready"] = True # ------------------------------------------------------------------ # Initialisation @@ -575,8 +611,25 @@ class MemoryStore: return dict(row) def close(self) -> None: - """Close the database connection.""" - self._conn.close() + """Release this instance's reference to the shared connection. + + The underlying connection is closed only when the last MemoryStore + referencing the same database is closed, so closing one instance can + never break sibling instances that still hold it. Idempotent. + """ + if getattr(self, "_entry", None) is None: + return + with MemoryStore._shared_guard: + entry = self._entry + if entry is None: + return + entry["refs"] -= 1 + if entry["refs"] <= 0: + try: + entry["conn"].close() + finally: + MemoryStore._shared.pop(self._key, None) + self._entry = None def __enter__(self) -> "MemoryStore": return self diff --git a/tests/plugins/memory/test_holographic_store.py b/tests/plugins/memory/test_holographic_store.py new file mode 100644 index 00000000000..59d8ee1cb81 --- /dev/null +++ b/tests/plugins/memory/test_holographic_store.py @@ -0,0 +1,225 @@ +"""Tests for the holographic MemoryStore shared-connection registry. + +MemoryStore instances pointing at the same database file must share one +process-wide SQLite connection and one re-entrant lock. Multiple providers +coexist in a single process (the main agent plus every delegate_task +subagent); when each instance owned a private connection they raced as +independent WAL writers and intermittently failed with "database is locked". + +Covers: connection sharing/refcounting, close() semantics, cross-instance +visibility, concurrent multi-instance writers, and write-lock release after +a failed write. +""" + +import sqlite3 +import threading + +import pytest + +from plugins.memory.holographic.store import MemoryStore + + +@pytest.fixture(autouse=True) +def _clean_shared_registry(): + """Each test starts and ends with an empty shared-connection registry.""" + # Drop any leakage from earlier tests in the same process. + for entry in list(MemoryStore._shared.values()): + try: + entry["conn"].close() + except sqlite3.Error: + pass + MemoryStore._shared.clear() + yield + leaked = list(MemoryStore._shared) + for entry in list(MemoryStore._shared.values()): + try: + entry["conn"].close() + except sqlite3.Error: + pass + MemoryStore._shared.clear() + assert not leaked, f"test leaked shared connections: {leaked}" + + +@pytest.fixture +def db_path(tmp_path): + return tmp_path / "memory_store.db" + + +class TestSharedConnection: + def test_same_path_shares_one_connection(self, db_path): + a = MemoryStore(db_path) + b = MemoryStore(db_path) + try: + assert a._conn is b._conn + assert a._lock is b._lock + assert len(MemoryStore._shared) == 1 + assert MemoryStore._shared[str(a.db_path)]["refs"] == 2 + finally: + a.close() + b.close() + + def test_different_paths_get_distinct_connections(self, tmp_path): + a = MemoryStore(tmp_path / "one.db") + b = MemoryStore(tmp_path / "two.db") + try: + assert a._conn is not b._conn + assert len(MemoryStore._shared) == 2 + finally: + a.close() + b.close() + + def test_writes_visible_across_instances(self, db_path): + a = MemoryStore(db_path) + b = MemoryStore(db_path) + try: + fact_id = a.add_fact("Hermes likes shared connections", category="test") + facts = b.list_facts(category="test") + assert [f["fact_id"] for f in facts] == [fact_id] + finally: + a.close() + b.close() + + def test_schema_initialised_once_per_connection(self, db_path): + a = MemoryStore(db_path) + b = MemoryStore(db_path) # must not re-run schema init / WAL probe + try: + assert MemoryStore._shared[str(a.db_path)]["ready"] is True + b.add_fact("schema still works") + finally: + a.close() + b.close() + + +class TestCloseSemantics: + def test_closing_one_instance_keeps_sibling_alive(self, db_path): + a = MemoryStore(db_path) + b = MemoryStore(db_path) + a.close() + try: + # The shared connection must survive the sibling's close(). + fact_id = b.add_fact("survivor write") + assert fact_id > 0 + finally: + b.close() + + def test_last_close_releases_connection(self, db_path): + a = MemoryStore(db_path) + b = MemoryStore(db_path) + conn = a._conn + a.close() + b.close() + assert MemoryStore._shared == {} + with pytest.raises(sqlite3.ProgrammingError): + conn.execute("SELECT 1") + + def test_close_is_idempotent(self, db_path): + a = MemoryStore(db_path) + b = MemoryStore(db_path) + a.close() + a.close() # double close must not steal b's reference + try: + b.add_fact("still alive after double close") + assert MemoryStore._shared[str(b.db_path)]["refs"] == 1 + finally: + b.close() + + def test_context_manager_releases_reference(self, db_path): + with MemoryStore(db_path) as store: + store.add_fact("context managed") + assert MemoryStore._shared == {} + + def test_reopen_after_full_close(self, db_path): + with MemoryStore(db_path) as store: + store.add_fact("first lifetime") + with MemoryStore(db_path) as store: + facts = store.list_facts() + assert [f["content"] for f in facts] == ["first lifetime"] + + +class TestConcurrency: + def test_concurrent_multi_instance_writers(self, db_path): + """Many instances writing from many threads must never hit + 'database is locked' — the failure mode of per-instance connections.""" + n_threads, n_facts = 8, 15 + errors: list[BaseException] = [] + + def writer(idx: int) -> None: + store = MemoryStore(db_path) + try: + for i in range(n_facts): + store.add_fact(f"fact thread={idx} seq={i}", category="load") + except BaseException as exc: # noqa: BLE001 - recorded for assert + errors.append(exc) + finally: + store.close() + + threads = [threading.Thread(target=writer, args=(i,)) for i in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors, f"concurrent writers failed: {errors[:3]}" + with MemoryStore(db_path) as store: + facts = store.list_facts(category="load", limit=500) + assert len(facts) == n_threads * n_facts + assert MemoryStore._shared == {} + + def test_failed_write_does_not_pin_write_lock(self, db_path, monkeypatch): + """A write that raises mid-method must not leave an open transaction + holding the SQLite write lock (autocommit isolation_level=None).""" + broken = MemoryStore(db_path) + sibling = MemoryStore(db_path) + try: + monkeypatch.setattr( + MemoryStore, + "_rebuild_bank", + lambda self, category: (_ for _ in ()).throw(RuntimeError("boom")), + ) + with pytest.raises(RuntimeError, match="boom"): + broken.add_fact("write that fails after the INSERT") + monkeypatch.undo() + + # No dangling transaction: the connection reports autocommit state + # and the sibling can write immediately. + assert broken._conn.in_transaction is False + sibling.add_fact("sibling write right after the failure") + finally: + broken.close() + sibling.close() + + +class TestProviderShutdown: + """The provider's shutdown() must release its shared connection, not just + drop the reference. Leaving finalization to GC keeps the connection (and + its write lock) alive on a long-running gateway, which is exactly the + "database is locked" contention the shared-connection registry removes.""" + + def test_shutdown_releases_shared_connection(self, db_path): + from plugins.memory.holographic import HolographicMemoryProvider + + provider = HolographicMemoryProvider(config={"db_path": str(db_path)}) + provider.initialize("session-shutdown") + assert MemoryStore._shared[str(db_path)]["refs"] == 1 + + provider.shutdown() + + assert provider._store is None + assert MemoryStore._shared == {} + + def test_shutdown_keeps_sibling_provider_alive(self, db_path): + from plugins.memory.holographic import HolographicMemoryProvider + + a = HolographicMemoryProvider(config={"db_path": str(db_path)}) + b = HolographicMemoryProvider(config={"db_path": str(db_path)}) + a.initialize("session-a") + b.initialize("session-b") + assert MemoryStore._shared[str(db_path)]["refs"] == 2 + + a.shutdown() + # Sibling still holds a live, writable connection. + assert MemoryStore._shared[str(db_path)]["refs"] == 1 + assert b._store is not None + b._store.add_fact("write after sibling shutdown") + b.shutdown() + assert MemoryStore._shared == {}