SQLite locking issues in holographic memory store

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #61726 fix(memory): share one SQLite connection per holographic store database

  merged to main

  files: plugins/memory/holographic/__init__.py, plugins/memory/holographic/store.py, scripts/release.py, tests/plugins/memory/test_holographic_store.py

  Show the diff

  • plugins/memory/holographic/__init__.py

    @@ -251,12 +251,12 @@ def on_memory_write(self, action: str, target: str, content: str) -> None:
                     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()

  • 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,41 @@ def __init__(
             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.
    +        # resolve() (not just expanduser) so symlinked/relative paths to the
    +        # same file share ONE connection instead of silently reintroducing
    +        # the multi-writer contention this registry exists to prevent.
    +        try:
    +            self._key = str(self.db_path.resolve())
    +        except OSError:
    +            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 +617,25 @@ def _row_to_dict(self, row: sqlite3.Row) -> dict:
             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

  • scripts/release.py

    @@ -258,6 +258,7 @@
         "[redacted:email]": "necoweb3",
         "[redacted:email]": "hydracoco7",  # PR #61382 salvage (id-less cron job freeze)
         "[email protected]": "AlexFucuson9",  # PR #61209 salvage (hygiene compression data loss)
    +    "[redacted:email]": "adambiggs",  # PR #43819 salvage (holographic shared SQLite connection)
         "[redacted:email]": "cypctlinux",  # PR #52403 salvage (Slack bot/workflow auth before no-user-id guard)
         "[email protected]": "YLChen-007",  # PR #26965 (approval remote command substitution)
         "[email protected]": "egilewski",  # co-author, PR #40663

  • tests/plugins/memory/test_holographic_store.py

    @@ -0,0 +1,243 @@
    +"""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
    +
    +
    [redacted:email](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}"
    +
    +
    [redacted:email]
    +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_symlinked_path_shares_connection(self, tmp_path):
    +        """A symlink to the same DB file must hit the same registry entry —
    +        otherwise two connections to one file silently reintroduce the
    +        multi-writer contention the registry exists to prevent."""
    +        real_dir = tmp_path / "real"
    +        real_dir.mkdir()
    +        link_dir = tmp_path / "link"
    +        link_dir.symlink_to(real_dir)
    +
    +        a = MemoryStore(real_dir / "memory_store.db")
    +        b = MemoryStore(link_dir / "memory_store.db")
    +        try:
    +            assert a._conn is b._conn
    +            assert len(MemoryStore._shared) == 1
    +        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 == {}