From 9ba82a1cc289ec679381f1fd7a4368b06abe6aea Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:00:02 -0500 Subject: [PATCH 1/3] fix: Admit foreground model ownership before database writers --- docs/resource-budgets.md | 38 ++ ...st-maintenance-foreground-admission-753.py | 491 ++++++++++++++++++ ...test-rebuild-foreground-publication-756.py | 6 +- tests/test-rebuild-source-streaming-756.py | 1 + tests/test_issue_511_write_path_opt.py | 20 +- truememory/engine.py | 20 +- truememory/vector_search.py | 25 +- 7 files changed, 579 insertions(+), 22 deletions(-) create mode 100644 tests/test-maintenance-foreground-admission-753.py diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index 3fe0675a..f01c02cd 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -647,3 +647,41 @@ resume. Synthetic SQLite tests cover precise REAL metadata rendered into separation text, signed zero, collations, storage types, rowid aliases, generated fields, repair failures and caller rollback. These tests do not measure native model behavior or add a process memory budget. + +### Foreground model admission before writer ownership + +Owned foreground add and incremental vector publication now acquire the model +identity fence before opening the SQLite write transaction. Add first owns its +engine write lock, so another mutation on that same engine cannot hold the +engine lock while waiting for a model fence held by the queued add. The model +fence validates the captured generation and remains held through commit or +rollback. There is one model acquisition, with no nested acquisition of the +nonreentrant model lock. + +This permits an owned foreground call to wait for a concurrent cluster source +snapshot without holding SQLite's writer or failing a valid vector publication. +Model loading and inference finish before owned writer admission. A model +generation change during loading, encoding or admission still rejects stale +publication; add and update propagate those typed errors instead of silently +falling back to a source-only write. Update retains its existing distinction: +capture failure precedes its source write, while a later vector-publication +failure can occur after the source update committed. + +Caller transactions use nonblocking model capture and publication admission. +Add and update classify the connection during model capture under their engine +write lock, so another same-engine mutation's transaction cannot be mistaken +for borrowed caller work. They release that lock before encoding. +Capture accepts only a cached model observed together with its identity under +the model lock. Busy ownership or an unloaded model produces a typed retry +error before any additional source write; the caller's transaction remains +intact. Cached-model inference can still run inside a caller-held transaction. +The publication fence protects the nested savepoint release, not a later outer +commit controlled by that caller. Streamed rebuilds retain nonblocking model +fences after writer admission and use the same cached-only borrowed capture. + +SQLite and event regressions cover actual clustering snapshot contention, +same-engine mutation ordering, stale generations, borrowed sentinels, atomic +add publication, and model release after admission/publication/commit failures. +This checkpoint changes admission ordering rather than bounding wait duration: +cluster snapshots may still hold the model lock while scanning the corpus. +Native integration and latency remain separate acceptance checks. diff --git a/tests/test-maintenance-foreground-admission-753.py b/tests/test-maintenance-foreground-admission-753.py new file mode 100644 index 00000000..d295387e --- /dev/null +++ b/tests/test-maintenance-foreground-admission-753.py @@ -0,0 +1,491 @@ +"""Foreground publication admission with real SQLite and synthetic models.""" + +import runpy +import sqlite3 +import threading +import types +import unittest +from pathlib import Path +from unittest.mock import patch + + +FOREGROUND = runpy.run_path(str(Path(__file__).with_name("test-rebuild-foreground-publication-756.py"))) +CLUSTERS = runpy.run_path(str(Path(__file__).with_name("test-cluster-publication-749.py"))) + + +class AdmissionFixture(unittest.TestCase): + setUp = FOREGROUND["TestForegroundPublication"].setUp + seed = FOREGROUND["TestForegroundPublication"].seed + ids = FOREGROUND["TestForegroundPublication"].ids + writer = FOREGROUND["TestForegroundPublication"].writer + engine = FOREGROUND["TestForegroundPublication"].engine + + def start(self, action, name="synthetic-worker"): + results, done = [], threading.Event() + + def run(): + try: + results.append(action()) + except BaseException as error: + results.append(error) + finally: + done.set() + + thread = threading.Thread(target=run, name=name, daemon=True) + thread.start() + self.addCleanup(thread.join, 3) + return thread, results, done + + def released(self): + self.assertTrue(self.vector._lock.acquire(blocking=False)) + self.vector._lock.release() + + def sentinel(self): + self.conn.execute("INSERT INTO metadata(key,value) VALUES ('synthetic-caller','pending')") + + def preserved(self): + self.assertTrue(self.conn.in_transaction) + self.assertEqual(self.conn.execute("SELECT value FROM metadata WHERE key='synthetic-caller'").fetchone()[0], "pending") + + +class TestForegroundAdmission(AdmissionFixture): + def test_style_computation_runs_outside_capture_and_publication_locks(self) -> None: + engine = self.engine() + engine._has_style_vec = True + events = [] + style = [0.25, 0.75] + + def compute(content: str) -> list[float]: + self.assertFalse(engine._write_lock.locked()) + self.assertFalse(self.vector._lock.locked()) + self.assertFalse(self.conn.in_transaction) + events.append("compute") + return style + + def publish(conn: sqlite3.Connection, sender: str, content: str, + _pre_computed_vec: list[float]) -> None: + self.assertTrue(engine._write_lock.locked()) + self.assertTrue(self.vector._lock.locked()) + self.assertTrue(conn.in_transaction) + self.assertIs(_pre_computed_vec, style) + events.append("publish") + + module = types.ModuleType("synthetic_style") + module.compute_style_vector = compute + with patch.dict(self.modules, personality_style_vec=module), \ + patch.dict(engine.add.__globals__, _update_style_vec=publish): + engine.add("synthetic styled content", sender="synthetic sender") + self.assertEqual(events, ["compute", "publish"]) + self.assertEqual(self.ids(), [1]) + self.assertEqual(self.ids("vec_messages_sep"), [1]) + + def test_actual_cluster_snapshot_delays_add_before_writer_then_add_commits_both_vectors(self): + engine = self.engine() + cluster, runtime, _native = CLUSTERS["load_clustering"]() + encoding, snapshot, admission, release = (threading.Event() for _ in range(4)) + underlying = self.vector._lock + fixture = self + + class ObservedLock: + def acquire(self, blocking=True): + if threading.current_thread().name == "synthetic-add" and snapshot.is_set(): + fixture.assertTrue(blocking) + fixture.assertFalse(engine.conn.in_transaction) + fixture.assertTrue(engine._write_lock.locked()) + admission.set() + return underlying.acquire(blocking=blocking) + + def release(self): + underlying.release() + + def __enter__(self): + self.acquire() + return self + + def __exit__(self, *_args): + self.release() + + self.vector._lock = runtime._lock = ObservedLock() + runtime.EMBEDDING_MODEL = self.vector.EMBEDDING_MODEL + cluster_conn, independent = self.writer(), self.writer() + original = cluster._get_all_embeddings + + def read(conn): + snapshot.set() + self.assertTrue(release.wait(3)) + return original(conn) + + def encode(texts): + if len(self.calls) == 1: + encoding.set() + self.assertTrue(snapshot.wait(3)) + return [[1., 0.] for _ in texts] + + def consolidate(): + self.assertTrue(encoding.wait(3)) + return cluster.cluster_messages(cluster_conn) + + self.on_encode = encode + with patch.object(cluster, "_get_all_embeddings", read): + cluster_thread, cluster_result, cluster_done = self.start(consolidate) + add_thread, result, done = self.start(lambda: engine.add("synthetic first add"), "synthetic-add") + try: + self.assertTrue(admission.wait(2)) + self.assertFalse(done.is_set()) + independent.execute("INSERT INTO metadata(key,value) VALUES ('synthetic-independent','committed')") + independent.commit() + self.assertFalse(self.conn.in_transaction) + finally: + release.set() + add_thread.join(3) + cluster_thread.join(3) + self.assertTrue(done.is_set()) + self.assertTrue(cluster_done.is_set()) + self.assertIsInstance(cluster_result[0], (int, RuntimeError), cluster_result) + self.assertIsInstance(result[0], dict, result) + self.assertEqual(self.ids(), [1]) + self.assertEqual(self.ids("vec_messages_sep"), [1]) + self.assertEqual(self.conn.execute("SELECT content FROM messages").fetchone()[0], "synthetic first add") + self.released() + + def test_writer_owner_defers_nonblocking_model_fence_and_foreground_finishes(self): + owner = self.writer() + owner.execute("BEGIN IMMEDIATE") + admission, attempted = threading.Event(), threading.Event() + underlying = self.vector._lock + + class AdmittedLock: + def acquire(self, blocking=True): + acquired = underlying.acquire(blocking=blocking) + if acquired and threading.current_thread().name == "synthetic-publish": + admission.set() + return acquired + + def release(self): + underlying.release() + + self.vector._lock = AdmittedLock() + identity = (0, self.vector.EMBEDDING_MODEL, 2) + + def publish(): + with self.vector._foreground_vector_publication(self.conn, identity, 1): + self.conn.execute("INSERT INTO metadata(key,value) VALUES ('synthetic-published','yes')") + + thread, result, done = self.start(publish, "synthetic-publish") + try: + self.assertTrue(admission.wait(2)) + with self.assertRaisesRegex(RuntimeError, "model is busy"): + with self.vector._rebuild_model_fence(identity): + attempted.set() + self.assertFalse(attempted.is_set()) + finally: + owner.rollback() + thread.join(3) + self.assertTrue(done.is_set()) + self.assertEqual(result, [None]) + self.released() + + def test_add_waiting_for_same_engine_update_does_not_own_model(self): + self.seed((1,)) + engine = self.engine() + update_ready, add_waiting = threading.Event(), threading.Event() + underlying = engine._write_lock + + class EngineLock: + def __enter__(self): + if threading.current_thread().name == "synthetic-add": + add_waiting.set() + underlying.acquire() + return self + + def __exit__(self, *_args): + underlying.release() + + engine._write_lock = EngineLock() + original = engine.update.__globals__["update_message"] + + def update(conn, memory_id, **fields): + result = original(conn, memory_id, **fields) + update_ready.set() + self.assertTrue(add_waiting.wait(3)) + self.released() + return result + + with patch.dict(engine.update.__globals__, update_message=update): + update_thread, updated, update_done = self.start(lambda: engine.update(1, "synthetic updated"), "synthetic-update") + self.assertTrue(update_ready.wait(2)) + add_thread, added, add_done = self.start(lambda: engine.add("synthetic added"), "synthetic-add") + update_thread.join(3) + add_thread.join(3) + self.assertTrue(update_done.is_set()) + self.assertTrue(add_done.is_set()) + self.assertIsInstance(updated[0], dict, updated) + self.assertIsInstance(added[0], dict, added) + self.assertEqual(self.ids(), [1, 2]) + self.assertEqual(self.ids("vec_messages_sep"), [1, 2]) + + def test_same_engine_writer_finishes_before_capture_classifies_transaction(self) -> None: + self.seed((1,)) + for operation in ("add", "update"): + with self.subTest(operation=operation): + engine = self.engine() + writer_ready, waiting, release = (threading.Event() for _ in range(3)) + underlying = engine._write_lock + + class EngineLock: + def __enter__(self) -> "EngineLock": + if threading.current_thread().name == "synthetic-capture": + waiting.set() + underlying.acquire() + return self + + def __exit__(self, *_args: object) -> None: + underlying.release() + + engine._write_lock = EngineLock() + self.vector._model = None + loads = [] + + def load() -> object: + self.assertFalse(self.conn.in_transaction) + self.assertTrue(underlying.locked()) + loads.append(True) + self.vector._model = self.model + return self.model + + def writer() -> None: + with engine._write_lock: + self.conn.execute("BEGIN IMMEDIATE") + self.conn.execute("INSERT INTO metadata(key,value) VALUES (?, 'pending')", + ("synthetic-owner-" + operation,)) + writer_ready.set() + self.assertTrue(release.wait(3)) + self.conn.commit() + + with patch.object(self.vector, "get_model", load): + writer_thread, written, writer_done = self.start(writer) + self.assertTrue(writer_ready.wait(2)) + action = (lambda: engine.add("synthetic added")) if operation == "add" else ( + lambda: engine.update(1, "synthetic updated") + ) + mutation_thread, result, done = self.start(action, "synthetic-capture") + try: + self.assertTrue(waiting.wait(2)) + self.assertFalse(done.is_set()) + self.assertEqual(loads, []) + finally: + release.set() + mutation_thread.join(3) + writer_thread.join(3) + self.assertTrue(writer_done.is_set()) + self.assertEqual(written, [None]) + self.assertTrue(done.is_set()) + self.assertIsInstance(result[0], dict, result) + self.assertEqual(loads, [True]) + self.assertFalse(self.conn.in_transaction) + self.released() + + def test_generation_change_during_model_load_rejects_add_and_update_before_source_write(self): + self.seed((1,)) + engine = self.engine() + + def changed(): + self.vector._model_generation += 1 + return self.model + + with patch.object(self.vector, "get_model", changed): + for action in (lambda: engine.add("synthetic rejected"), lambda: engine.update(1, "synthetic rejected")): + with self.assertRaisesRegex(self.vector.VectorPublicationChanged, "changed while loading"): + action() + self.assertEqual(self.conn.execute("SELECT id,content FROM messages").fetchall(), [(1, "synthetic-1")]) + self.assertEqual(self.ids(), []) + self.released() + + def test_generation_change_while_waiting_rejects_before_opening_transaction(self): + captured = (0, self.vector.EMBEDDING_MODEL, 2) + attempted = threading.Event() + underlying = self.vector._lock + + class WaitingLock: + def acquire(self, blocking=True): + attempted.set() + return underlying.acquire(blocking=blocking) + + def release(self): + underlying.release() + + self.vector._lock = WaitingLock() + underlying.acquire() + + def publish(): + with self.vector._foreground_vector_publication(self.conn, captured, 1): + self.fail("Changed generation entered publication") + + thread, results, done = self.start(publish) + try: + self.assertTrue(attempted.wait(2)) + self.assertFalse(done.is_set()) + self.assertFalse(self.conn.in_transaction) + self.vector._model_generation += 1 + finally: + underlying.release() + thread.join(3) + self.assertIsInstance(results[0], self.vector.VectorPublicationChanged) + self.assertIn("model changed", str(results[0])) + self.assertFalse(self.conn.in_transaction) + self.released() + + def test_add_failure_after_source_insert_rolls_back_source_and_vectors(self): + engine = self.engine() + with patch.object(self.vector, "_validate_foreground_vector_target", side_effect=self.vector.VectorPublicationChanged("synthetic target changed")): + with self.assertRaises(self.vector.VectorPublicationChanged): + engine.add("synthetic rollback") + self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(self.ids(), []) + self.assertEqual(self.ids("vec_messages_sep"), []) + self.assertFalse(self.conn.in_transaction) + self.released() + + def test_admission_begin_and_commit_failures_release_model(self): + real = self.conn + + class FailedConnection: + def __init__(self, phase): + self.phase = phase + + def __getattr__(self, name): + return getattr(real, name) + + def execute(self, sql, *args): + if self.phase == "begin" and sql == "BEGIN IMMEDIATE": + raise sqlite3.OperationalError("synthetic writer admission") + return real.execute(sql, *args) + + def commit(self): + if self.phase == "commit": + raise sqlite3.OperationalError("synthetic commit failure") + return real.commit() + + for phase in ("begin", "commit"): + with self.subTest(phase=phase): + with self.assertRaises(sqlite3.OperationalError): + self.engine(FailedConnection(phase)).add("synthetic rollback") + self.assertFalse(real.in_transaction) + self.assertEqual(real.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(self.ids(), []) + self.released() + + def test_owned_model_load_and_encoding_finish_before_writer(self): + def load(): + self.assertFalse(self.conn.in_transaction) + return self.model + + def encode(texts): + self.assertFalse(self.conn.in_transaction) + self.released() + return [[1., 0.] for _ in texts] + + self.on_encode = encode + with patch.object(self.vector, "get_model", load): + self.engine().add("synthetic outside writer") + self.assertEqual(self.ids(), [1]) + + +class TestBorrowedAdmission(AdmissionFixture): + def actions(self): + engine = self.engine() + return ( + ("add", lambda: engine.add("synthetic rejected")), + ("update", lambda: engine.update(1, "synthetic rejected")), + ("embed_single", lambda: self.vector.embed_single(self.conn, 1, "synthetic-1")), + ("completion", lambda: self.vector.build_vectors(self.conn)), + ("separation", lambda: self.vector.build_separation_vectors(self.conn)), + ) + + def test_busy_capture_defers_every_mutation_without_waiting_or_loading(self): + self.seed((1,)) + self.sentinel() + calls = [] + + class BusyLock: + def acquire(self, blocking=True): + calls.append(blocking) + if blocking: + raise AssertionError("Borrowed caller attempted a blocking model acquisition") + return False + + with patch.object(self.vector, "_lock", BusyLock()), patch.object(self.vector, "get_model", side_effect=AssertionError("Borrowed model load")): + for name, action in self.actions(): + with self.subTest(action=name): + with self.assertRaisesRegex(self.vector.VectorPublicationChanged, "model is busy"): + action() + self.preserved() + self.assertEqual(calls, [False] * 5) + self.assertEqual(self.calls, []) + self.assertEqual(self.conn.execute("SELECT id,content FROM messages").fetchall(), [(1, "synthetic-1")]) + self.assertEqual(self.ids(), []) + self.conn.rollback() + self.released() + + def test_unloaded_capture_defers_every_mutation_and_preserves_sentinel(self): + self.seed((1,)) + self.sentinel() + self.vector._model = None + with patch.object(self.vector, "get_model", side_effect=AssertionError("Borrowed model load")) as loader: + for name, action in self.actions(): + with self.subTest(action=name): + with self.assertRaisesRegex(self.vector.VectorPublicationChanged, "not loaded"): + action() + self.preserved() + self.released() + loader.assert_not_called() + self.assertEqual(self.calls, []) + self.assertEqual(self.conn.execute("SELECT id,content FROM messages").fetchall(), [(1, "synthetic-1")]) + self.conn.rollback() + + def test_borrowed_loaded_snapshot_skips_loader_and_add_remains_rollbackable(self): + self.sentinel() + with patch.object(self.vector, "get_model", side_effect=AssertionError("Borrowed model load")) as loader: + item = self.engine().add("synthetic borrowed") + loader.assert_not_called() + self.preserved() + self.assertEqual(self.ids(), [item["id"]]) + self.assertEqual(self.ids("vec_messages_sep"), [item["id"]]) + self.conn.rollback() + self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(self.ids(), []) + self.released() + + def test_borrowed_publication_contention_preserves_caller_transaction(self): + self.sentinel() + held, release = threading.Event(), threading.Event() + + def hold(): + with self.vector._lock: + held.set() + self.assertTrue(release.wait(3)) + + holder = [] + + def encode(texts): + if len(self.calls) == 2: + holder.append(self.start(hold)[0]) + self.assertTrue(held.wait(2)) + return [[1., 0.] for _ in texts] + + self.on_encode = encode + try: + with self.assertRaisesRegex(self.vector.VectorPublicationChanged, "model is busy"): + self.engine().add("synthetic rejected") + self.preserved() + self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + finally: + release.set() + for thread in holder: + thread.join(3) + self.conn.rollback() + self.released() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test-rebuild-foreground-publication-756.py b/tests/test-rebuild-foreground-publication-756.py index 6c71ac5e..5e8b0b7d 100644 --- a/tests/test-rebuild-foreground-publication-756.py +++ b/tests/test-rebuild-foreground-publication-756.py @@ -273,7 +273,7 @@ def __getattr__(self, name): def execute(self, sql, parameters=()): if sql == "BEGIN IMMEDIATE" or sql == "UPDATE messages SET id = id WHERE 0": - case.assertFalse(case.vector._lock.locked()) + case.assertTrue(case.vector._lock.locked()) boundaries.append("writer") if sql == "RELEASE SAVEPOINT rebuild_source": case.assertTrue(case.vector._lock.locked()) @@ -296,13 +296,13 @@ def commit(self): def test_embed_owned_commit_stays_inside_model_fence(self): self._assert_foreground_writer_and_terminal_fence("embed", False) - def test_embed_caller_writer_upgrade_precedes_fence_and_release_stays_inside(self): + def test_embed_caller_fence_precedes_writer_upgrade_and_contains_release(self): self._assert_foreground_writer_and_terminal_fence("embed", True) def test_add_owned_commit_stays_inside_model_fence(self): self._assert_foreground_writer_and_terminal_fence("add", False) - def test_add_caller_writer_upgrade_precedes_fence_and_release_stays_inside(self): + def test_add_caller_fence_precedes_writer_upgrade_and_contains_release(self): self._assert_foreground_writer_and_terminal_fence("add", True) def test_denied_add_savepoint_release_rolls_back_only_add(self): diff --git a/tests/test-rebuild-source-streaming-756.py b/tests/test-rebuild-source-streaming-756.py index 369304e5..cf3ca3e8 100644 --- a/tests/test-rebuild-source-streaming-756.py +++ b/tests/test-rebuild-source-streaming-756.py @@ -78,6 +78,7 @@ def setUp(self): self.addCleanup(self.conn.close) self.conn.executescript("CREATE TABLE vec_messages(embedding BLOB); CREATE TABLE vec_messages_sep(embedding BLOB);") self.model = object() + self.vector._model = self.model self.vector.get_model = lambda: self.model self.calls = [] self.on_encode = None diff --git a/tests/test_issue_511_write_path_opt.py b/tests/test_issue_511_write_path_opt.py index c2f1cbe4..0d1858a3 100644 --- a/tests/test_issue_511_write_path_opt.py +++ b/tests/test_issue_511_write_path_opt.py @@ -5,7 +5,9 @@ M16: Fallback embed_single inside write lock — removed in favor of skip-and-log """ +import ast import inspect +import textwrap import unittest @@ -30,13 +32,17 @@ class TestM15StyleVecOutsideLock(unittest.TestCase): def test_add_precomputes_style_vec(self): from truememory.engine import TrueMemoryEngine - source = inspect.getsource(TrueMemoryEngine.add) - lock_pos = source.find("self._write_lock") - precompute_pos = source.find("compute_style_vector") - self.assertGreater(precompute_pos, -1, - "add() should reference compute_style_vector for pre-computation") - self.assertLess(precompute_pos, lock_pos, - "compute_style_vector call should appear before _write_lock") + tree = ast.parse(textwrap.dedent(inspect.getsource(TrueMemoryEngine.add))) + calls = [node for node in ast.walk(tree) if isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) and node.func.id == "compute_style_vector"] + self.assertEqual(len(calls), 1) + for block in (node for node in ast.walk(tree) if isinstance(node, ast.With)): + self.assertNotIn(calls[0], list(ast.walk(block)), + "Style computation must run outside lock and transaction contexts") + publications = [node for node in ast.walk(tree) if isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) and node.func.id == "_update_style_vec"] + self.assertEqual(len(publications), 1) + self.assertLess(calls[0].lineno, publications[0].lineno) def test_style_vec_accepts_precomputed(self): from truememory.personality_style_vec import update_entity_style_vector_incremental diff --git a/truememory/engine.py b/truememory/engine.py index 0285b066..9e61f69f 100644 --- a/truememory/engine.py +++ b/truememory/engine.py @@ -724,14 +724,18 @@ def add( pre_embedding = None pre_sep_embedding = None if self._has_vectors: + from truememory.vector_search import VectorPublicationChanged try: from truememory.vector_search import ( _capture_rebuild_model, _encode_with_mps_fallback, _build_sep_text, ) - model, pre_identity = _capture_rebuild_model() + with self._write_lock: + model, pre_identity = _capture_rebuild_model(self.conn) pre_embedding = _encode_with_mps_fallback(model, [content])[0] sep_text = _build_sep_text(sender, recipient, timestamp, content) pre_sep_embedding = _encode_with_mps_fallback(model, [sep_text])[0] + except VectorPublicationChanged: + raise except Exception: logger.debug("Failed to pre-compute embedding during add()", exc_info=True) @@ -743,13 +747,15 @@ def add( except Exception: logger.debug("Failed to pre-compute style vector during add()", exc_info=True) + from contextlib import nullcontext from truememory.rebuild_source import rebuild_transaction - publication_fence = None if pre_embedding is not None: from truememory.vector_search import _foreground_model_fence - publication_fence = _foreground_model_fence(pre_identity) - with self._write_lock, rebuild_transaction(self.conn, write=True, publication_fence=publication_fence): + with self._write_lock, ( + _foreground_model_fence(pre_identity, blocking=not self.conn.in_transaction) + if pre_embedding is not None else nullcontext() + ), rebuild_transaction(self.conn, write=True): msg = { "content": content, "sender": sender, @@ -1176,11 +1182,13 @@ def update(self, memory_id: int, content: str | None = None, **fields) -> dict | pre_embedding = None pre_sep_embedding = None if content is not None and self._has_vectors: + from truememory.vector_search import VectorPublicationChanged try: from truememory.vector_search import ( _capture_rebuild_model, _encode_with_mps_fallback, _build_sep_text, ) - model, pre_identity = _capture_rebuild_model() + with self._write_lock: + model, pre_identity = _capture_rebuild_model(self.conn) pre_embedding = _encode_with_mps_fallback(model, [content])[0] row = self.conn.execute( "SELECT sender, recipient, timestamp FROM messages WHERE id = ?", @@ -1192,6 +1200,8 @@ def update(self, memory_id: int, content: str | None = None, **fields) -> dict | timestamp_val = fields.get("timestamp", row[2]) sep_text = _build_sep_text(sender_val, recipient_val, timestamp_val, content) pre_sep_embedding = _encode_with_mps_fallback(model, [sep_text])[0] + except VectorPublicationChanged: + raise except Exception: logger.debug("Failed to pre-compute embedding during update()", exc_info=True) diff --git a/truememory/vector_search.py b/truememory/vector_search.py index 507e4bb9..965ff0ab 100644 --- a/truememory/vector_search.py +++ b/truememory/vector_search.py @@ -739,13 +739,23 @@ def _encode_with_mps_fallback(model, texts, **kwargs): """ -def _capture_rebuild_model() -> tuple[object, tuple[int, str, int]]: +def _capture_rebuild_model(conn: sqlite3.Connection | None = None) -> tuple[object, tuple[int, str, int]]: + """Caller transactions can borrow a cached model, but cannot wait or load.""" + if conn is not None and conn.in_transaction: + if not _lock.acquire(blocking=False): + raise VectorPublicationChanged("Embedding model is busy; retry outside the caller transaction") + try: + if _model is None: + raise VectorPublicationChanged("Embedding model is not loaded; load it outside the caller transaction") + return _model, (_model_generation, EMBEDDING_MODEL, _embedding_dim) + finally: + _lock.release() with _lock: before = (_model_generation, EMBEDDING_MODEL) model = get_model() with _lock: if before != (_model_generation, EMBEDDING_MODEL): - raise RuntimeError("Embedding model changed while loading rebuild model") + raise VectorPublicationChanged("Embedding model changed while loading rebuild model") return model, (_model_generation, EMBEDDING_MODEL, _embedding_dim) @@ -767,8 +777,9 @@ class VectorPublicationChanged(RuntimeError): @contextmanager -def _foreground_model_fence(identity: tuple[int, str, int]) -> Iterator[None]: - if not _lock.acquire(blocking=False): +def _foreground_model_fence(identity: tuple[int, str, int], *, blocking: bool = False) -> Iterator[None]: + """Blocking admission is permitted only before an owned writer starts.""" + if not _lock.acquire(blocking=blocking): raise VectorPublicationChanged("Embedding model is busy; retry vector publication") try: if identity != (_model_generation, EMBEDDING_MODEL, _embedding_dim): @@ -815,7 +826,7 @@ def _foreground_vector_publication( ) -> Iterator[tuple[str, str]]: from truememory.rebuild_source import rebuild_transaction - with rebuild_transaction(conn, write=True, publication_fence=_foreground_model_fence(identity)): + with _foreground_model_fence(identity, blocking=not conn.in_transaction), rebuild_transaction(conn, write=True): yield _validate_foreground_vector_target(conn, identity, message_id) @@ -856,7 +867,7 @@ def _build_streamed_vectors( save_manifest(conn, key, empty) _clear_build_in_progress(conn, table) return 0 - model, identity = _capture_rebuild_model() + model, identity = _capture_rebuild_model(conn) with rebuild_transaction(conn, write=True, publication_fence=_rebuild_model_fence(identity)): manifest = load_manifest(conn, key) resumable = (manifest is not None and not manifest.complete @@ -1191,7 +1202,7 @@ def embed_single(conn: sqlite3.Connection, message_id: int, content: str) -> Non message_id: The ``messages.id`` of the row being embedded. content: The text to embed. """ - model, identity = _capture_rebuild_model() + model, identity = _capture_rebuild_model(conn) embedding = _encode_with_mps_fallback(model, [content])[0] normed = _normalize_for_cosine(embedding) if normed is None: From 2cfa2bd6ec59461f5df80b27c63ae5fd16d10c1c Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:20:29 -0500 Subject: [PATCH 2/3] fix: Coalesce database maintenance behind owned workers --- docs/resource-budgets.md | 73 ++ tests/test-maintenance-engine-routing-753.py | 783 +++++++++++++++++++ tests/test_consolidation_gaps.py | 194 ++--- tests/test_hardening.py | 12 +- tests/test_issue_498_auto_consolidate.py | 76 +- tests/test_issue_590_build_vectors_batch.py | 2 + tests/test_issue_647_vector_rebuild.py | 2 + truememory/engine.py | 394 ++++------ truememory/maintenance.py | 316 +++++++- 9 files changed, 1431 insertions(+), 421 deletions(-) create mode 100644 tests/test-maintenance-engine-routing-753.py diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md index 178e3e1e..662152a0 100644 --- a/docs/resource-budgets.md +++ b/docs/resource-budgets.md @@ -1032,3 +1032,76 @@ add publication, and model release after admission/publication/commit failures. This checkpoint changes admission ordering rather than bounding wait duration: cluster snapshots may still hold the model lock while scanning the corpus. Native integration and latency remain separate acceptance checks. + +### Engine maintenance routing + +Engine startup, committed mutations and later safe API boundaries now use the +same durable eight-layer planner. The default append trigger remains +`insert_count - attempted_insert_count >= 25`; 24 committed appends do not +qualify and the 25th does, including across engine instances and reopen. +`TRUEMEMORY_AUTO_CONSOLIDATE_EVERY` retains its positive-integer override. +Initial historical or empty input attempts once. Corrections, source epochs and +dependency changes follow their separate eligibility rules. A successful empty +cluster result no longer starts every layer again on each open. Source edits +committed through another handle become visible at the next safe API boundary; +there is no idle polling timer. Uncommitted caller changes do not schedule work. + +Foreground scheduling reads eight checkpoint records, source counters and small +dependency metadata. It does not scan source/vector rows, import native +frameworks, wait for the model lock or perform consolidation in `add()`. The +foreground write lock is acquired nonblocking for this probe. Package versions +are cached in three records and refreshed on connection setup and worker/manual +setup. One extension-failure record is bound to the probed model/table identity +and version snapshot. Actual worker extension loading supplies that evidence; +metadata alone is not proof of availability. Epoch checks prevent a late worker +from overwriting newer shared capability evidence, while that worker still uses +its own actual failure for its current attempt. A transient model-busy probe +cannot erase an existing extension-failure baseline. + +Each canonical file has at most one active owner and one pending notification +generation. A notification stores only a generation and threshold, not an +engine, caller connection or source content. Multiple foreground notifications +replace that single pending slot. A same-process owner's release services an +existing pending notification; it creates no notification of its own. A pending +coordinator behind a local owner is retained only until that exact owner's +release or cancellation, so closing the last engine does not lose the wake. +Cross-process contention reports busy/pending and needs a later API boundary; +this checkpoint has no automatic cross-process release notification. Deferred +model ownership and failed attempts do not schedule their own retry loops. + +Workers create and close their own file connection. `engine.close()` detaches +and closes only the foreground connection; it does not cancel or join a shared +worker. Cancellation clears queued notifications but retains ownership until +active computation reaches a cooperative boundary and its connection closes. +Thread-start failure closes an unclaimed owner descriptor; interruption after +the target started leaves teardown with that exact worker and cannot cancel a +later successor. Child processes discard inherited owner descriptors, pending +references and registry state. Daemon threads do not guarantee completion when +the process exits; durable abandoned-attempt recovery remains necessary. + +Explicit `consolidate()` forces one pass with the same eight result keys plus +the existing nonpersisted preference extraction result. A busy owner produces +nine explicit `BUSY` entries and does not queue a hidden forced pass. A file +call without an existing transaction uses a dedicated connection. An existing +caller transaction uses the original connection's protected savepoints and +reports `pending caller commit`; its writes and maintenance output remain +rollbackable together. `:memory:` automatic work reports pending without opening +an empty replacement database or creating a worker. Explicit manual maintenance +uses the original in-memory handle. Caller-held writer and later outer-commit +limitations from the borrowed-runner contract still apply. + +Successful manual summary/cluster publication restores the calling engine's +corresponding capability flags, including successful empty output. Failed or +unavailable layers never enable those flags or clear a previously usable +capability. The existing automatic-consolidation opt-out remains until explicit +manual repair. Cluster retrieval still requires live vector availability. +`get_stats()` exposes bounded coordinator and per-layer outcome, freshness, +coverage and categorical error state; it does not expose source text, paths or +exception messages. Unverified vector-generation and legacy-contact coverage +remain explicit limitations, rather than complete success claims. + +This routing checkpoint preserves builder algorithms and the A17 foreground +vector-publication guards. Ingest database replacement, profile/style getter +freshness and custom-schema source tracking remain separate work. Synthetic +SQLite and event tests establish scheduling and ownership behavior; native +latency, throughput and memory acceptance still require measurement. diff --git a/tests/test-maintenance-engine-routing-753.py b/tests/test-maintenance-engine-routing-753.py new file mode 100644 index 00000000..5c8c02ff --- /dev/null +++ b/tests/test-maintenance-engine-routing-753.py @@ -0,0 +1,783 @@ +"""Engine routing with real SQLite, event-controlled owners and synthetic models.""" + +import ast +import builtins +import gc +import logging +import os +import select +import signal +import runpy +import sqlite3 +import sys +import threading +import time +import types +import unittest +import warnings +import weakref +from contextlib import contextmanager +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] +ADAPTERS = runpy.run_path(str(Path(__file__).with_name("test-maintenance-adapters-753.py"))) +STORAGE, MAINTENANCE = ADAPTERS["STORAGE"], ADAPTERS["MAINTENANCE"] + + +def stdlib_module(name: str, modules: dict) -> types.ModuleType: + module = types.ModuleType("synthetic_routing_" + name) + + def safe_import(name, globals=None, locals=None, fromlist=(), level=0): + if name in modules: + return modules[name] + if name.startswith("truememory"): + raise ImportError("Synthetic boundary excludes native modules") + return builtins.__import__(name, globals, locals, fromlist, level) + + module.__dict__["__builtins__"] = dict(vars(builtins), __import__=safe_import) + path = ROOT / "truememory" / (name + ".py") + with patch.dict(sys.modules, {module.__name__: module}): + exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), module.__dict__) + return module + + +def load_engine(modules: dict) -> types.ModuleType: + module = types.ModuleType("synthetic_routing_engine") + + def safe_import(name, globals=None, locals=None, fromlist=(), level=0): + if name in modules: + return modules[name] + if name.startswith("truememory"): + raise ImportError("Synthetic boundary excludes native modules") + return builtins.__import__(name, globals, locals, fromlist, level) + + module.__dict__.update({"__builtins__": dict(vars(builtins), __import__=safe_import), + "__file__": str(ROOT / "truememory/engine.py"), "Path": Path, "sqlite3": sqlite3, + "threading": threading, "time": time, "os": os, "warnings": warnings, + "logger": logging.getLogger("synthetic-routing"), "MAX_CONTENT_LENGTH": 50_000, + "_env_int": lambda name, default, **kwargs: default}) + source = ast.parse((ROOT / "truememory/engine.py").read_text(encoding="utf-8")) + nodes = [ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0)] + for node in source.body: + if isinstance(node, (ast.FunctionDef, ast.ClassDef)): + nodes.append(node) + elif isinstance(node, ast.ImportFrom) and node.module == "truememory.storage": + module.__dict__.update({item.asname or item.name: getattr(STORAGE, item.name) for item in node.names}) + elif isinstance(node, ast.Assign): + for target in node.targets: + if isinstance(target, ast.Name): + if target.id.startswith("_HAS_"): + module.__dict__[target.id] = False + elif target.id in {"_ALLOWED_TABLES", "_ALLOWED_COLUMNS", "_ALL_VEC_TABLES", "_SQLITE_IN_CHUNK"}: + nodes.append(node) + module.__dict__["_HAS_CONSOLIDATION"] = True + exec(compile(ast.fix_missing_locations(ast.Module(body=nodes, type_ignores=[])), + str(ROOT / "truememory/engine.py"), "exec"), module.__dict__) + return module + + +class RoutingFixture(ADAPTERS["AdapterFixture"]): + def setUp(self) -> None: + style = stdlib_module("personality_style_vec", {}) + self.enterContext(patch.dict(sys.modules, {"truememory.personality_style_vec": style})) + super().setUp() + self.modules.update({"truememory.storage": STORAGE, "truememory.maintenance": MAINTENANCE, + "truememory.vector_search": self.vector, + "sqlite_vec": types.SimpleNamespace(load=lambda conn: None)}) + self.modules["truememory.rebuild_source"] = stdlib_module("rebuild_source", self.modules) + self.engine_module = load_engine(self.modules) + create = STORAGE.create_db + + class ExtensionBoundary: + def __init__(boundary, conn): + boundary.conn = conn + + def __getattr__(boundary, name): + return getattr(boundary.conn, name) + + def enable_load_extension(boundary, enabled): + # The fixture uses ordinary BLOB tables and a fake sqlite-vec + # loader, including hosts compiled without extension support. + self.assertIs(type(enabled), bool) + + self.enterContext(patch.object(MAINTENANCE, "create_db", lambda path: ExtensionBoundary(create(path)))) + self.enterContext(patch.object(self.engine_module, "create_db", lambda path: ExtensionBoundary(create(path)))) + self.engine = self.new_engine(self.conn) + self.coordinator = self.engine._get_maintenance_coordinator() + self.addCleanup(self.coordinator.wait, 3) + self.addCleanup(self.coordinator.cancel) + + def new_engine(self, conn: sqlite3.Connection | None = None): + engine = self.engine_module.TrueMemoryEngine(self.path) + engine.conn = conn or self.open_db() + engine.ready = True + engine._has_consolidation = False + self.addCleanup(engine.close) + return engine + + def completed(self) -> None: + self.assertTrue(self.coordinator.wait(3), "Synthetic maintenance did not terminate") + self.assertFalse(self.coordinator.snapshot()["active"]) + + def fake_report(self): + return MAINTENANCE.MaintenanceReport((), "SKIPPED (synthetic)") + + +class TestCoordinatorHandoff(RoutingFixture): + def test_local_owner_retains_one_pending_wake_after_last_engine_close(self) -> None: + path = self.path.with_name("synthetic-detached.sqlite") + engine = self.engine_module.TrueMemoryEngine(path) + engine.conn, engine.ready = STORAGE.create_db(path), True + coordinator = engine._get_maintenance_coordinator() + reference, engine_reference = weakref.ref(coordinator), weakref.ref(engine) + entered, release = threading.Event(), threading.Event() + observed = [] + + def work(conn, owner, **kwargs): + observed.append(kwargs["threshold"]) + entered.set() + self.assertTrue(release.wait(3)) + return self.fake_report() + + with patch.object(MAINTENANCE, "run_engine_maintenance", work): + try: + with MAINTENANCE.maintenance_owner(path): + coordinator.request_layers(threshold=31) + coordinator.request_layers(threshold=37) + self.assertEqual(len(MAINTENANCE._owner_waiters), 1) + engine.close() + del engine, coordinator + gc.collect() + self.assertIsNone(engine_reference()) + self.assertIsNotNone(reference()) + self.assertTrue(entered.wait(2)) + self.assertEqual(MAINTENANCE._owner_waiters, {}) + coordinator = reference() + self.assertIsNotNone(coordinator) + finally: + release.set() + if reference() is not None: + self.assertTrue(reference().wait(3)) + self.assertEqual(observed, [37]) + + def test_cancel_releases_local_waiter_and_admission_cannot_restore_it(self) -> None: + original = MAINTENANCE._acquire_owner + + def cancelled_admission(path, on_busy=None): + self.coordinator.cancel() + return original(path, on_busy=on_busy) + + with patch.object(MAINTENANCE, "run_engine_maintenance") as work: + with MAINTENANCE.maintenance_owner(self.path): + self.coordinator.request_layers() + self.assertEqual(len(MAINTENANCE._owner_waiters), 1) + self.coordinator.cancel() + self.assertEqual(MAINTENANCE._owner_waiters, {}) + with patch.object(MAINTENANCE, "_acquire_owner", cancelled_admission): + self.assertFalse(self.coordinator.request_layers()) + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0) + self.assertEqual(MAINTENANCE._owner_waiters, {}) + work.assert_not_called() + self.assertEqual(self.coordinator.status, ("cancelled", None)) + + def test_delayed_start_exception_cannot_cancel_successor_or_its_pending_wake(self) -> None: + start = threading.Thread.start + entered, release = threading.Event(), threading.Event() + attempts, observed, cancellation = [], [], [] + + def work(conn, coordinator, **kwargs): + observed.append(kwargs["threshold"]) + if kwargs["threshold"] == 31: + entered.set() + self.assertTrue(release.wait(3)) + cancellation.append(kwargs["cancel"].is_set()) + return self.fake_report() + + def late_failure(worker): + attempts.append(worker) + start(worker) + if len(attempts) == 1: + self.completed() + self.assertTrue(self.coordinator.request_layers(threshold=31)) + self.assertTrue(entered.wait(2)) + self.assertFalse(self.coordinator.request_layers(threshold=37)) + raise RuntimeError("synthetic delayed start failure") + + with patch.object(MAINTENANCE.threading.Thread, "start", late_failure), \ + patch.object(MAINTENANCE, "run_engine_maintenance", work): + try: + with self.assertRaisesRegex(RuntimeError, "synthetic delayed start failure"): + self.coordinator.request_layers() + self.assertFalse(self.coordinator._cancel.is_set()) + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1) + finally: + release.set() + self.completed() + self.assertEqual(observed, [25, 31, 37]) + self.assertEqual(cancellation, [False]) + + @unittest.skipUnless(hasattr(os, "fork"), "POSIX fork required") + def test_fork_discards_inherited_pending_worker_and_rejects_stale_coordinator(self) -> None: + entered, release = threading.Event(), threading.Event() + + def active(conn, cancel): + entered.set() + self.assertTrue(release.wait(3)) + + self.coordinator.request(active) + self.assertTrue(entered.wait(2)) + self.coordinator.request_layers() + waiting_path = self.path.with_name("synthetic-fork-waiter.sqlite") + self.enterContext(MAINTENANCE.maintenance_owner(waiting_path)) + waiting = MAINTENANCE.get_coordinator(waiting_path) + self.addCleanup(waiting.cancel) + waiting.request_layers() + self.assertEqual(len(MAINTENANCE._owner_waiters), 1) + read_fd, write_fd = os.pipe() + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + child = os.fork() + if child == 0: + os.close(read_fd) + try: + rejected = False + try: + self.coordinator.request_layers() + except RuntimeError: + rejected = True + fresh = MAINTENANCE.get_coordinator(self.path) + clean = (not fresh.snapshot()["active"] and fresh.snapshot()["pending_generations"] == 0 + and not MAINTENANCE._owner_waiters) + os.write(write_fd, b"PASS" if rejected and clean else b"FAIL") + finally: + os.close(write_fd) + os._exit(0) + os.close(write_fd) + try: + self.assertTrue(select.select([read_fd], [], [], 2)[0], "Synthetic child did not answer") + self.assertEqual(os.read(read_fd, 4), b"PASS") + finally: + os.close(read_fd) + if os.waitpid(child, os.WNOHANG)[0] == 0: + os.kill(child, signal.SIGKILL) + os.waitpid(child, 0) + self.coordinator.cancel() + release.set() + self.completed() + + def test_start_failure_services_only_a_real_later_notification(self) -> None: + start = threading.Thread.start + attempts, observed = [], [] + + def fail_once(worker): + attempts.append(True) + if len(attempts) == 1: + self.assertFalse(self.coordinator.request_layers(threshold=31)) + raise RuntimeError("synthetic start failure") + return start(worker) + + with patch.object(MAINTENANCE.threading.Thread, "start", fail_once), \ + patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(kwargs["threshold"]) or self.fake_report()): + with self.assertRaisesRegex(RuntimeError, "synthetic start failure"): + self.coordinator.request_layers() + self.completed() + self.assertEqual(observed, [31]) + self.assertEqual(len(attempts), 2) + + def test_start_failure_without_notification_has_no_retry(self) -> None: + with patch.object(MAINTENANCE.threading.Thread, "start", side_effect=RuntimeError("synthetic")) as start: + with self.assertRaises(RuntimeError): + self.coordinator.request_layers() + self.completed() + start.assert_called_once() + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0) + self.assertEqual(self.coordinator.status, ("failed", "worker_start")) + + def test_close_detaches_without_cancelling_or_retaining_engine(self) -> None: + entered, release = threading.Event(), threading.Event() + + def active(conn, cancel): + entered.set() + self.assertTrue(release.wait(2)) + self.assertFalse(cancel.is_set()) + self.assertEqual(conn.execute("SELECT 1").fetchone()[0], 1) + + with patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: self.fake_report()): + self.coordinator.request(active) + self.assertTrue(entered.wait(2)) + engine = self.engine_module.TrueMemoryEngine(self.path) + engine.conn, engine.ready = self.open_db(), True + reference = weakref.ref(engine) + engine._maybe_auto_consolidate() + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1) + engine.close() + del engine + gc.collect() + self.assertIsNone(reference()) + self.assertFalse(self.coordinator.wait(0)) + release.set() + self.completed() + self.assertEqual(self.coordinator.status, ("success", None)) + + def test_busy_admission_and_concurrent_owner_release_cannot_lose_wake(self) -> None: + held, release, leaving, ran = (threading.Event() for _ in range(4)) + original_acquire, original_release = MAINTENANCE._acquire_owner, MAINTENANCE._release_owner + + def owner(): + with MAINTENANCE.maintenance_owner(self.path): + held.set() + self.assertTrue(release.wait(2)) + + def release_owner(fd): + leaving.set() + original_release(fd) + + def acquire(path, on_busy=None): + def busy(): + on_busy() + release.set() + self.assertTrue(leaving.wait(2)) + return original_acquire(path, on_busy=busy if on_busy is not None else None) + + owner_thread = threading.Thread(target=owner) + with patch.object(MAINTENANCE, "_release_owner", release_owner), \ + patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: ran.set() or self.fake_report()): + owner_thread.start() + try: + self.assertTrue(held.wait(2)) + with patch.object(MAINTENANCE, "_acquire_owner", acquire): + self.assertFalse(self.coordinator.request_layers()) + self.assertTrue(ran.wait(2)) + self.completed() + finally: + release.set() + owner_thread.join(3) + self.assertFalse(owner_thread.is_alive()) + + def test_many_active_notifications_coalesce_to_one_static_successor(self) -> None: + entered, release = threading.Event(), threading.Event() + observed = [] + + def active(conn, cancel): + entered.set() + self.assertTrue(release.wait(2)) + + def successor(conn, coordinator, **kwargs): + observed.append((conn is self.conn, kwargs["threshold"])) + return self.fake_report() + + with patch.object(MAINTENANCE, "run_engine_maintenance", successor): + self.assertTrue(self.coordinator.request(active)) + self.assertTrue(entered.wait(2)) + for threshold in range(1, 101): + self.assertFalse(self.coordinator.request_layers(threshold=threshold)) + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1) + self.assertFalse(self.coordinator.wait(0)) + release.set() + self.completed() + self.assertEqual(observed, [(False, 100)]) + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0) + + def test_local_synchronous_owner_release_services_pending_wake(self) -> None: + observed = [] + with patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(True) or self.fake_report()): + with MAINTENANCE.maintenance_owner(self.path): + self.assertFalse(self.coordinator.request_layers()) + self.assertEqual(self.coordinator.status, ("busy", None)) + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1) + self.assertEqual(observed, []) + self.completed() + self.assertEqual(observed, [True]) + + def test_external_busy_requires_new_boundary_and_does_not_spin(self) -> None: + observed = [] + with patch.object(MAINTENANCE, "try_file_lock", return_value=None), \ + patch.object(MAINTENANCE.threading, "Thread") as thread: + self.assertFalse(self.coordinator.request_layers()) + thread.assert_not_called() + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 1) + self.assertEqual(MAINTENANCE._owner_waiters, {}) + with patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(True) or self.fake_report()): + self.assertTrue(self.coordinator.request_layers()) + self.completed() + self.assertEqual(observed, [True]) + + def test_wake_between_descriptor_release_and_idle_transition_is_not_lost(self) -> None: + released, finish = threading.Event(), threading.Event() + original_release = MAINTENANCE._release_owner + observed = [] + first = [] + + def pause(fd): + original_release(fd) + if not first: + first.append(True) + released.set() + self.assertTrue(finish.wait(2)) + + with patch.object(MAINTENANCE, "_release_owner", pause), \ + patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(True) or self.fake_report()): + self.assertTrue(self.coordinator.request(lambda conn, cancel: None)) + self.assertTrue(released.wait(2)) + self.assertFalse(self.coordinator.request_layers()) + self.assertFalse(self.coordinator.wait(0)) + finish.set() + self.completed() + self.assertEqual(observed, [True]) + + def test_model_deferral_has_no_self_generated_successor(self) -> None: + observed = [] + report = MAINTENANCE.MaintenanceReport((MAINTENANCE.LayerResult( + "clusters", "cluster_messages", "deferred", None, 0., "ModelBusy", False),), "SKIPPED") + with patch.object(MAINTENANCE, "run_engine_maintenance", lambda *args, **kwargs: observed.append(True) or report): + self.assertTrue(self.coordinator.request_layers()) + self.completed() + self.assertEqual(observed, [True]) + self.assertEqual(self.coordinator.status, ("deferred", "ModelBusy")) + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0) + + def test_cancellation_drops_pending_but_keeps_active_ownership(self) -> None: + entered, release = threading.Event(), threading.Event() + + def active(conn, cancel): + entered.set() + self.assertTrue(release.wait(2)) + + with patch.object(MAINTENANCE, "run_engine_maintenance") as successor: + self.coordinator.request(active) + self.assertTrue(entered.wait(2)) + self.coordinator.request_layers() + self.coordinator.cancel() + self.assertFalse(self.coordinator.wait(0)) + self.assertFalse(self.coordinator.request_layers()) + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0) + release.set() + self.completed() + successor.assert_not_called() + self.assertEqual(self.coordinator.status, ("cancelled", None)) + + +class TestEngineRouting(RoutingFixture): + def test_status_reports_categorical_layer_failure_and_caller_transaction(self) -> None: + self.add(1) + with patch.object(self.cluster, "cluster_messages", side_effect=RuntimeError("synthetic private sentinel")): + self.engine.consolidate() + state = self.engine.get_stats()["maintenance"] + self.assertEqual(len(state["layers"]), 8) + self.assertEqual(state["layers"]["clusters"]["outcome"], "failed") + self.assertEqual(state["layers"]["clusters"]["error_category"], "RuntimeError") + self.assertEqual(state["layers"]["clusters"]["freshness"], "untrusted") + self.assertNotIn("synthetic private sentinel", str(state)) + self.assertNotIn(str(self.path), str(state)) + self.conn.execute("UPDATE messages SET content='synthetic uncommitted status'") + self.assertTrue(self.engine.get_stats()["maintenance"]["pending_caller_commit"]) + self.assertTrue(self.conn.in_transaction) + self.conn.rollback() + + def test_manual_repair_restores_actual_cluster_search_and_automatic_eligibility(self) -> None: + self.add(5) + self.native.fit = lambda: setattr(self.native, "labels", [0, 0, 1, 1, 2]) + self.engine._has_clustering = False + self.engine._has_vectors = True + self.modules["truememory.agentic_search"] = types.SimpleNamespace(normalize_scores=lambda rows: rows) + with patch.object(self.coordinator, "request_layers") as request, \ + patch.object(self.engine, "search", return_value=[{"id": 1, "score": .5}]), \ + patch.object(self.engine, "_entity_focused_search", return_value=[]), \ + patch.object(self.engine, "_check_sufficiency", return_value=True), \ + patch.object(self.engine, "_apply_surprise_boost", side_effect=lambda rows: rows), \ + patch.object(self.engine, "_clean_results", side_effect=lambda rows, *_args, **_kwargs: rows), \ + patch.object(self.engine_module, "search_clustered", return_value=[{"id": 2, "score": .5}], create=True) as search: + self.engine._maybe_auto_consolidate() + request.assert_not_called() + self.engine.search_agentic("synthetic query", use_hyde=False, use_reranker=False) + search.assert_not_called() + result = self.engine.consolidate() + self.assertIn("3 clusters", result["cluster_messages"]) + self.assertTrue(self.engine._has_clustering) + self.assertTrue(self.engine._has_consolidation) + rows = self.engine.search_agentic("synthetic query", use_hyde=False, use_reranker=False) + search.assert_called_once() + self.assertEqual([row["id"] for row in rows], [1, 2]) + request.assert_not_called() + self.conn.execute("UPDATE messages SET content='synthetic repaired correction' WHERE id=1") + self.conn.commit() + self.engine._maybe_auto_consolidate() + request.assert_called_once_with(threshold=25) + + def test_failed_manual_repair_does_not_enable_capabilities_or_remove_prior_success(self) -> None: + self.engine._has_clustering = False + with patch.object(self.cluster, "cluster_messages", side_effect=RuntimeError("synthetic")), \ + patch.object(self.modules["truememory.consolidation"], "build_summaries", side_effect=RuntimeError("synthetic")): + result = self.engine.consolidate() + self.assertIn("ERROR", result["cluster_messages"]) + self.assertIn("ERROR", result["build_summaries"]) + self.assertFalse(self.engine._has_clustering) + self.assertFalse(self.engine._has_consolidation) + self.engine._has_clustering = self.engine._has_consolidation = True + self.engine.consolidate() + self.assertTrue(self.engine._has_clustering) + self.assertTrue(self.engine._has_consolidation) + + def test_unavailable_manual_cluster_repair_preserves_only_prior_capability(self) -> None: + self.engine._has_clustering = False + with patch.object(self.modules["sqlite_vec"], "load", side_effect=RuntimeError("synthetic")): + self.assertIn("UNAVAILABLE", self.engine.consolidate()["cluster_messages"]) + self.assertFalse(self.engine._has_clustering) + self.engine._has_clustering = True + self.assertIn("UNAVAILABLE", self.engine.consolidate()["cluster_messages"]) + self.assertTrue(self.engine._has_clustering) + + def test_first_open_manual_does_not_start_its_own_competitor(self) -> None: + self.add(1) + engine = self.engine_module.TrueMemoryEngine(self.path) + self.addCleanup(engine.close) + with patch.object(self.coordinator, "request_layers") as request: + result = engine.consolidate() + request.assert_not_called() + self.assertTrue(engine.ready) + self.assertEqual(set(result), set(MAINTENANCE._MAINTENANCE_RESULT_KEYS)) + + def test_current_empty_startup_and_rolled_back_correction_do_not_schedule(self) -> None: + self.engine.consolidate() + self.engine._has_consolidation = True + with patch.object(self.coordinator, "request_layers") as request: + self.engine._maybe_startup_consolidate() + self.engine._ensure_connection() + request.assert_not_called() + self.add(1) + self.engine.consolidate() + with patch.object(self.coordinator, "request_layers") as request: + self.conn.execute("UPDATE messages SET content='synthetic uncommitted'") + self.engine._ensure_connection() + self.assertEqual(self.engine._maintenance_pending_reason, "pending_caller_commit") + self.conn.rollback() + self.engine._ensure_connection() + request.assert_not_called() + self.conn.execute("UPDATE messages SET content='synthetic committed'") + self.conn.commit() + self.engine._ensure_connection() + request.assert_called_once_with(threshold=25) + + def test_add_probe_uses_no_corpus_scan_native_import_or_blocking_model_lock(self) -> None: + self.engine.consolidate() + self.engine._has_consolidation = True + operations = [] + underlying = self.vector._lock + + class ProbeLock: + def acquire(lock, blocking=True): + self.assertFalse(blocking) + return underlying.acquire(blocking=False) + + def release(lock): + underlying.release() + + def authorize(action, table, column, *_args): + if action == sqlite3.SQLITE_READ: + operations.append((table, column)) + return sqlite3.SQLITE_OK + + self.conn.set_authorizer(authorize) + try: + with patch.object(self.vector, "_lock", ProbeLock()), \ + patch.object(MAINTENANCE, "_installed_dependency", side_effect=AssertionError("per-add package scan")), \ + patch.object(self.coordinator, "request_layers", side_effect=AssertionError("unexpected work")): + self.engine._maybe_auto_consolidate() + self.assertEqual(self.engine._maintenance_pending_reason, "awaiting_eligibility") + finally: + self.conn.set_authorizer(lambda *_args: sqlite3.SQLITE_OK) + self.assertFalse({table for table, _ in operations}.intersection({"messages", "vec_messages_edge", "summaries", "cluster_centroids"})) + + def test_source_commit_followed_by_a17_vector_error_still_notifies(self) -> None: + self.add(1) + self.engine.consolidate() + self.engine._has_vectors = self.engine._has_consolidation = True + changed = type("VectorPublicationChanged", (RuntimeError,), {}) + + @contextmanager + def publication(*args): + raise changed("synthetic model replacement") + yield + + def capture(conn: sqlite3.Connection) -> tuple[object, str]: + self.assertIs(conn, self.engine.conn) + return object(), "synthetic-identity" + + fields = {"_capture_rebuild_model": capture, + "_encode_with_mps_fallback": lambda *args: [[1., 2.]], + "_build_sep_text": lambda *args: "synthetic separated text", + "_foreground_vector_publication": publication, + "serialize_f32": lambda values: b"synthetic-unused-vector", + "_write_embedder_metadata_no_commit": lambda conn: None, + "VectorPublicationChanged": changed} + with patch.dict(self.vector.__dict__, fields), patch.object(self.coordinator, "request_layers") as request: + with self.assertRaisesRegex(changed, "Source update committed"): + self.engine.update(1, content="synthetic corrected source") + request.assert_called_once_with(threshold=25) + self.assertFalse(self.conn.in_transaction) + self.assertFalse(self.engine._write_lock.locked()) + self.assertEqual(self.conn.execute("SELECT content FROM messages WHERE id=1").fetchone()[0], "synthetic corrected source") + + def test_manual_busy_is_truthful_and_does_not_queue_forced_work(self) -> None: + entered, release = threading.Event(), threading.Event() + + def active(conn, cancel): + entered.set() + self.assertTrue(release.wait(2)) + + self.coordinator.request(active) + self.assertTrue(entered.wait(2)) + try: + result = self.engine.consolidate() + self.assertEqual(set(result), set(MAINTENANCE._MAINTENANCE_RESULT_KEYS)) + self.assertTrue(all(value.startswith("BUSY") for value in result.values())) + self.assertEqual(self.coordinator.snapshot()["pending_generations"], 0) + finally: + release.set() + self.completed() + + def test_file_manual_uses_dedicated_connection_all_eight_and_preferences(self) -> None: + self.add(5) + seen = [] + original = MAINTENANCE.run_engine_maintenance + + def run(conn, coordinator, **kwargs): + seen.append(conn) + self.assertIsNot(conn, self.conn) + self.assertTrue(self.engine._write_lock.acquire(blocking=False)) + self.engine._write_lock.release() + return original(conn, coordinator, **kwargs) + + with patch.object(MAINTENANCE, "run_engine_maintenance", run), \ + patch.object(self.coordinator, "request_layers") as automatic: + result = self.engine.consolidate() + automatic.assert_not_called() + self.assertEqual(set(result), set(MAINTENANCE._MAINTENANCE_RESULT_KEYS)) + self.assertTrue(all("ERROR" not in value and "UNAVAILABLE" not in value for value in result.values()), result) + self.assertIn("vector_generation_unverified", result["cluster_messages"]) + with self.assertRaises(sqlite3.ProgrammingError): + seen[0].execute("SELECT 1") + + def test_24_then_25_committed_adds_across_engines_use_durable_counts(self) -> None: + self.engine.consolidate() + second = self.new_engine() + self.engine._has_consolidation = second._has_consolidation = True + with patch.object(self.coordinator, "request_layers", wraps=self.coordinator.request_layers) as request: + for index in range(24): + engine = self.engine if index % 2 else second + engine.add("synthetic append", sender="synthetic-sender") + request.assert_not_called() + second.add("synthetic twenty fifth", sender="synthetic-sender") + self.completed() + self.assertGreaterEqual(request.call_count, 1) + states = MAINTENANCE.read_layer_states(self.conn, self.specs()) + self.assertTrue(all(state.attempted_insert_count == 25 for state in states.values()), states) + + def test_borrowed_manual_preserves_caller_writes_and_rollback(self) -> None: + self.add(1, commit=False) + with patch.object(self.engine_module, "create_db", side_effect=AssertionError("new handle")): + result = self.engine.consolidate() + self.assertTrue(all("pending caller commit" in value for value in result.values()), result) + self.assertTrue(self.conn.in_transaction) + self.conn.rollback() + self.assertEqual(self.conn.execute("SELECT count(*) FROM messages").fetchone()[0], 0) + self.assertEqual(self.output(), ([], [])) + + def test_in_memory_auto_stays_pending_and_manual_uses_original_rows(self) -> None: + conn = STORAGE.create_db(":memory:") + self.addCleanup(conn.close) + engine = self.engine_module.TrueMemoryEngine(":memory:") + engine.conn, engine.ready = conn, True + self.addCleanup(engine.close) + conn.execute("INSERT INTO messages(content,timestamp) VALUES ('synthetic in memory','2026-01-01')") + conn.commit() + with patch.object(MAINTENANCE, "create_db", side_effect=AssertionError("new handle")), \ + patch.object(MAINTENANCE.threading, "Thread", side_effect=AssertionError("memory worker")): + engine._maybe_auto_consolidate() + self.assertEqual(engine._maintenance_coordinator.status, ("pending_in_memory", None)) + result = engine.consolidate() + self.assertIn("1 episodes", result["detect_episodes"]) + self.assertEqual(conn.execute("SELECT count(*) FROM messages").fetchone()[0], 1) + + +class TestCapabilityEvidence(RoutingFixture): + def test_transient_model_busy_cannot_clear_prior_extension_failure_baseline(self) -> None: + self.add(5) + with patch.object(self.modules["sqlite_vec"], "load", side_effect=RuntimeError("synthetic")): + self.engine.consolidate() + prior = self.coordinator.capability_snapshot()[2] + self.add(1) + worker = MAINTENANCE.create_db(self.path) + try: + with self.vector._lock: + MAINTENANCE._prepare_worker_extensions(worker, self.coordinator) + finally: + worker.close() + self.assertEqual(self.coordinator.capability_snapshot()[2], prior) + self.assertEqual(MAINTENANCE.plan_layers(self.conn, MAINTENANCE.engine_layer_specs(self.conn, self.coordinator)), ()) + + def test_reopen_during_worker_failure_preserves_current_run_local_evidence(self) -> None: + self.add(1) + + def fail_and_refresh(conn): + self.coordinator.refresh_capabilities() + raise RuntimeError("synthetic load failure after reopen") + + with patch.object(self.modules["sqlite_vec"], "load", fail_and_refresh): + result = self.engine.consolidate() + self.assertIn("VectorExtensionUnavailable", result["cluster_messages"]) + self.assertIsNone(self.coordinator.capability_snapshot()[2]) + + def test_worker_extension_failure_keeps_same_key_and_25_insert_baseline(self) -> None: + self.add(5) + loader = self.modules["sqlite_vec"] + with patch.object(loader, "load", side_effect=RuntimeError("synthetic private sentinel")): + result = self.engine.consolidate() + self.assertIn("VectorExtensionUnavailable", result["cluster_messages"]) + self.engine._has_consolidation = True + with patch.object(self.coordinator, "request_layers") as request: + self.engine.add("synthetic sixth") + self.engine._ensure_connection() + request.assert_not_called() + state = MAINTENANCE.read_layer_states(self.conn, MAINTENANCE.engine_layer_specs(self.conn, self.coordinator))["clusters"] + self.assertEqual(state.attempted_insert_count, 5) + self.assertEqual(state.dependency.key, state.attempted_dependency) + repaired = self.engine.consolidate() + self.assertNotIn("UNAVAILABLE", repaired["cluster_messages"]) + self.assertIsNone(self.coordinator.capability_snapshot()[2]) + + def test_reopen_refreshes_static_evidence_and_new_model_invalidates_failure_key(self) -> None: + self.add(1) + with patch.object(self.modules["sqlite_vec"], "load", side_effect=RuntimeError("synthetic")): + self.engine.consolidate() + epoch, versions, failure = self.coordinator.capability_snapshot() + self.assertEqual(len(versions), 3) + self.assertIsNotNone(failure) + self.vector.EMBEDDING_MODEL = "synthetic-model-c" + dependency = MAINTENANCE.engine_layer_specs(self.conn, self.coordinator)[0].resolve_dependency() + self.assertTrue(dependency.available) + engine = self.new_engine() + self.assertIs(engine._get_maintenance_coordinator(), self.coordinator) + self.assertGreater(self.coordinator.capability_snapshot()[0], epoch) + self.assertIsNone(self.coordinator.capability_snapshot()[2]) + + def test_old_worker_cannot_publish_capability_error_after_refresh(self) -> None: + epoch, _, _ = self.coordinator.capability_snapshot() + failure = MAINTENANCE.make_layer_dependency(1, available=False, error_category="VectorExtensionUnavailable") + self.coordinator.refresh_capabilities() + self.coordinator._publish_extension_evidence(epoch, failure) + self.assertIsNone(self.coordinator.capability_snapshot()[2]) + + def test_actual_connection_unavailability_is_not_overridden_by_cached_success(self) -> None: + dependency = self.spec().resolve_dependency() + unavailable = MAINTENANCE.make_layer_dependency(1, MAINTENANCE._dependency_parameters(dependency), + available=False, error_category="VectorExtensionUnavailable") + with patch.object(MAINTENANCE, "_cluster_schedule_dependency", return_value=unavailable): + observed = MAINTENANCE.engine_layer_specs(self.conn, self.coordinator)[0].resolve_dependency() + self.assertEqual(observed, unavailable) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_consolidation_gaps.py b/tests/test_consolidation_gaps.py index 3c92e12c..5344444d 100644 --- a/tests/test_consolidation_gaps.py +++ b/tests/test_consolidation_gaps.py @@ -1,134 +1,60 @@ -"""Regression tests for consolidation gaps (clustering + preferences). - -Verifies that consolidate() calls cluster_messages and extract_preferences — -features that were previously only available via ingest(). -""" -from __future__ import annotations - -import tempfile -import os -from unittest.mock import patch, MagicMock - - -from truememory.engine import TrueMemoryEngine -from truememory.storage import create_db - - -def _make_engine_with_messages(n=20): - """Create an engine with messages in a temp DB (no vectors needed). - - Disables startup consolidation to prevent background threads from - accessing SQLite cross-thread (causes segfault on Python 3.14). - """ - td = tempfile.mkdtemp() - db = os.path.join(td, "test.db") - conn = create_db(db) - for i in range(n): - conn.execute( - "INSERT INTO messages (content, sender, recipient, timestamp, category, modality) " - "VALUES (?, ?, '', '', '', '')", - (f"Message {i} about topic {i % 5}", "alice" if i % 2 == 0 else "bob"), - ) - conn.commit() - conn.close() - - eng = TrueMemoryEngine(db_path=db) - eng._has_consolidation = False - eng._ensure_connection() - return eng, td - - -def test_consolidate_calls_cluster_messages(): - """consolidate() must call cluster_messages when clustering is available.""" - eng, td = _make_engine_with_messages() - eng._has_vectors = True - - with patch("truememory.engine._HAS_CLUSTERING", True), \ - patch("truememory.clustering.cluster_messages", return_value=3) as mock_cm: - result = eng.consolidate() - - assert "cluster_messages" in result - assert "ERROR" not in result["cluster_messages"] - assert "3 clusters" in result["cluster_messages"] - mock_cm.assert_called_once_with(eng.conn) - eng.close() - - -def test_consolidate_calls_extract_preferences(): - """consolidate() must call extract_preferences when personality is available.""" - eng, td = _make_engine_with_messages() - - with patch("truememory.engine._HAS_PERSONALITY", True), \ - patch("truememory.personality.extract_preferences", return_value={}) as mock_ep: - result = eng.consolidate() - - assert "extract_preferences" in result - assert "ERROR" not in result["extract_preferences"] - mock_ep.assert_called_once_with(eng.conn) - eng.close() - - -def test_consolidate_skips_clustering_without_vectors(): - """consolidate() must skip clustering when vectors are unavailable.""" - eng, td = _make_engine_with_messages() - eng._has_vectors = False - - result = eng.consolidate() - assert "cluster_messages" not in result - eng.close() - - -def test_consolidate_handles_cluster_error_gracefully(): - """consolidate() must catch and report clustering errors.""" - eng, td = _make_engine_with_messages() - eng._has_vectors = True - - with patch("truememory.engine._HAS_CLUSTERING", True), \ - patch("truememory.clustering.cluster_messages", side_effect=RuntimeError("boom")): - result = eng.consolidate() - - assert "cluster_messages" in result - assert "ERROR" in result["cluster_messages"] - eng.close() - - -def test_auto_consolidation_threshold_lowered(): - """Default auto-consolidation threshold is 25 (was 100).""" - eng, td = _make_engine_with_messages() - assert eng._auto_consolidate_threshold == 25 - eng.close() - - -def test_startup_consolidation_triggers_when_clusters_empty(): - """On startup, if messages exist but clusters are empty, consolidate fires.""" - eng, td = _make_engine_with_messages(n=30) - eng._has_consolidation = True - - import truememory.engine as _eng_mod - with patch.object(_eng_mod.threading, "Thread") as mock_thread_cls: - mock_thread = MagicMock() - mock_thread_cls.return_value = mock_thread - eng._maybe_startup_consolidate() - - mock_thread_cls.assert_called_once() - assert mock_thread_cls.call_args.kwargs["name"] == "startup-consolidate" - mock_thread.start.assert_called_once() - eng.close() - - -def test_startup_consolidation_skips_when_few_messages(): - """Don't consolidate on startup if fewer messages than threshold.""" - eng, td = _make_engine_with_messages(n=5) - - consolidate_called = False - orig = eng._bg_consolidate - def spy(): - nonlocal consolidate_called - consolidate_called = True - return orig() - - eng._bg_consolidate = spy - eng._has_consolidation = True - eng._maybe_startup_consolidate() - assert not consolidate_called - eng.close() +"""Behavioral coverage for all maintenance layers and startup provenance.""" + +import runpy +from pathlib import Path +from unittest.mock import patch + + +_ROUTING = runpy.run_path(str(Path(__file__).with_name("test-maintenance-engine-routing-753.py"))) + + +class TestConsolidationGaps(_ROUTING["RoutingFixture"]): + def test_clustering_count_matches_real_published_centroids(self): + self.add(5) + self.native.fit = lambda: setattr(self.native, "labels", [0, 0, 1, 1, 2]) + result = self.engine.consolidate() + self.assertIn("3 clusters", result["cluster_messages"]) + self.assertEqual(self.conn.execute("SELECT count(*) FROM cluster_centroids").fetchone()[0], 3) + + def test_preferences_runs_on_the_worker_connection(self): + function = self.modules["truememory.personality"].extract_preferences + with patch.object(self.modules["truememory.personality"], "extract_preferences", wraps=function) as preference: + result = self.engine.consolidate() + self.assertIn("extract_preferences", result) + self.assertNotIn("ERROR", result["extract_preferences"]) + preference.assert_called_once() + self.assertIsNot(preference.call_args.args[0], self.conn) + + def test_missing_vector_index_reports_unavailable_without_skipping_siblings(self): + self.conn.execute("DROP TABLE vec_messages_edge") + self.conn.commit() + result = self.engine.consolidate() + self.assertIn("UNAVAILABLE", result["cluster_messages"]) + self.assertIn("build_summaries", result) + self.assertNotIn("ERROR", result["build_summaries"]) + + def test_cluster_error_is_categorical_and_siblings_still_run(self): + with patch.object(self.cluster, "cluster_messages", side_effect=RuntimeError("synthetic private sentinel")): + result = self.engine.consolidate() + self.assertEqual(result["cluster_messages"], "ERROR (RuntimeError)") + self.assertNotIn("synthetic private sentinel", str(result)) + self.assertNotIn("ERROR", result["detect_episodes"]) + + def test_startup_attempts_initial_small_corpus_once(self): + self.add(5) + self.engine._has_consolidation = True + self.engine._maybe_startup_consolidate() + self.completed() + self.assertEqual(self.state().outcome, "success_empty") + with patch.object(self.coordinator, "request_layers") as request: + self.engine._maybe_startup_consolidate() + request.assert_not_called() + + def test_successful_empty_clusters_do_not_retrigger_all_layers(self): + self.add(30) + self.engine.consolidate() + self.assertEqual(self.conn.execute("SELECT count(*) FROM cluster_centroids").fetchone()[0], 0) + self.engine._has_consolidation = True + with patch.object(self.coordinator, "request_layers") as request: + self.engine._maybe_startup_consolidate() + request.assert_not_called() diff --git a/tests/test_hardening.py b/tests/test_hardening.py index 654845ed..a8950826 100644 --- a/tests/test_hardening.py +++ b/tests/test_hardening.py @@ -188,15 +188,19 @@ class TestAutoConsolidation: def test_auto_consolidation_doesnt_fire_below_threshold(self): eng, td = _make_engine(n_messages=0) + # Establish the initial successful/empty or unavailable attempt. The + # threshold applies after that durable baseline, including on reopen. + eng.consolidate() eng._has_consolidation = True eng._auto_consolidate_threshold = 25 + eng.conn.executemany("INSERT INTO messages(content) VALUES (?)", [("synthetic append",)] * 24) + eng.conn.commit() - import truememory.engine as _eng_mod - with patch.object(_eng_mod.threading, "Thread") as mock_thread_cls: - for i in range(24): + with patch.object(eng._maintenance_coordinator, "request_layers") as request: + for _ in range(3): eng._maybe_auto_consolidate() - mock_thread_cls.assert_not_called() + request.assert_not_called() eng.close() def test_startup_consolidation_no_crash_on_closed_conn(self): diff --git a/tests/test_issue_498_auto_consolidate.py b/tests/test_issue_498_auto_consolidate.py index 901d7f73..86fd1238 100644 --- a/tests/test_issue_498_auto_consolidate.py +++ b/tests/test_issue_498_auto_consolidate.py @@ -1,37 +1,39 @@ -"""Regression test for M1 #498: auto-consolidation trigger. - -Consolidation should run automatically after a configurable number of -add() calls, not require manual truememory_consolidate invocation. -""" - -import inspect -import unittest - - -class TestAutoConsolidation(unittest.TestCase): - - def test_auto_consolidate_threshold_exists(self): - from truememory.engine import TrueMemoryEngine - source = inspect.getsource(TrueMemoryEngine.__init__) - self.assertIn("_auto_consolidate_threshold", source) - self.assertIn("TRUEMEMORY_AUTO_CONSOLIDATE_EVERY", source) - - def test_add_calls_maybe_auto_consolidate(self): - from truememory.engine import TrueMemoryEngine - source = inspect.getsource(TrueMemoryEngine.add) - self.assertIn("_maybe_auto_consolidate", source, - "add() should call _maybe_auto_consolidate()") - - def test_maybe_auto_consolidate_checks_threshold(self): - from truememory.engine import TrueMemoryEngine - source = inspect.getsource(TrueMemoryEngine._maybe_auto_consolidate) - self.assertIn("_auto_consolidate_threshold", source) - self.assertIn("_has_consolidation", source) - - def test_bg_consolidate_exists(self): - from truememory.engine import TrueMemoryEngine - self.assertTrue(hasattr(TrueMemoryEngine, "_bg_consolidate")) - - -if __name__ == "__main__": - unittest.main() +"""Durable automatic maintenance thresholds across engine instances.""" + +import runpy +from pathlib import Path +from unittest.mock import patch + + +_ROUTING = runpy.run_path(str(Path(__file__).with_name("test-maintenance-engine-routing-753.py"))) + + +class TestAutoConsolidation(_ROUTING["RoutingFixture"]): + def test_default_threshold_is_twenty_five(self): + self.assertEqual(self.engine._auto_consolidate_threshold, 25) + + def test_configured_threshold_uses_committed_source_inserts(self): + self.engine.consolidate() + self.engine._auto_consolidate_threshold = 3 + self.engine._has_consolidation = True + with patch.object(self.coordinator, "request_layers", wraps=self.coordinator.request_layers) as request: + for _ in range(2): + self.engine.add("synthetic committed append") + request.assert_not_called() + self.engine.add("synthetic threshold append") + self.completed() + self.assertTrue(request.called) + self.assertTrue(all(call.kwargs["threshold"] == 3 for call in request.call_args_list)) + + def test_close_and_reopen_does_not_reset_threshold_progress(self): + self.engine.consolidate() + self.add(24) + self.engine.close() + reopened = self.new_engine() + reopened._has_consolidation = True + with patch.object(self.coordinator, "request_layers", wraps=self.coordinator.request_layers) as request: + reopened._maybe_auto_consolidate() + request.assert_not_called() + reopened.add("synthetic twenty fifth") + self.completed() + self.assertTrue(request.called) diff --git a/tests/test_issue_590_build_vectors_batch.py b/tests/test_issue_590_build_vectors_batch.py index a678b1c8..1f11dc2a 100644 --- a/tests/test_issue_590_build_vectors_batch.py +++ b/tests/test_issue_590_build_vectors_batch.py @@ -73,6 +73,8 @@ def _make_server(monkeypatch, tmp_path): def _seed_messages(ms, n: int) -> None: """Add *n* messages to the database via the server module.""" m = ms._get_memory() + # These focused tests own explicit rebuild maintenance. + m._engine._has_consolidation = False for i in range(n): m.add(f"memory number {i}", user_id="test") diff --git a/tests/test_issue_647_vector_rebuild.py b/tests/test_issue_647_vector_rebuild.py index 3a168421..a85b8530 100644 --- a/tests/test_issue_647_vector_rebuild.py +++ b/tests/test_issue_647_vector_rebuild.py @@ -81,6 +81,8 @@ def _make_server(monkeypatch, tmp_path): def _seed_messages(ms, n: int) -> None: m = ms._get_memory() + # These focused tests own explicit rebuild maintenance. + m._engine._has_consolidation = False for i in range(n): m.add(f"memory number {i}", user_id="test") diff --git a/truememory/engine.py b/truememory/engine.py index 9e61f69f..37e7b4d8 100644 --- a/truememory/engine.py +++ b/truememory/engine.py @@ -31,6 +31,10 @@ import time import sqlite3 from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from truememory.maintenance import LayerResult, MaintenanceCoordinator # ─────────────────────────────────────────────────────────────────────────── # Core modules (always available) @@ -358,18 +362,18 @@ def __init__(self, db_path: str | Path = None, # Coordination flag for background NaN re-embed thread (#485). self._nan_migration_in_progress = False - # Auto-consolidation: run L5 consolidation every N adds (#498) - self._adds_since_consolidation = 0 self._auto_consolidate_threshold = _env_int( "TRUEMEMORY_AUTO_CONSOLIDATE_EVERY", 25, lo=1 ) - self._consolidation_thread: threading.Thread | None = None + self._maintenance_coordinator = None + self._maintenance_handle = None + self._maintenance_pending_reason = "not_connected" # ────────────────────────────────────────────────────────────────────── # Auto-connect (production API) # ────────────────────────────────────────────────────────────────────── - def _ensure_connection(self) -> None: + def _ensure_connection(self, *, _suppress_maintenance: bool = False) -> None: """Open (or create) the database and load extensions if needed. Called automatically by :meth:`add`, :meth:`search`, and other @@ -385,6 +389,8 @@ def _ensure_connection(self) -> None: # through to reconnect so the server self-heals once the FS recovers. try: self.conn.execute("PRAGMA schema_version") + if not _suppress_maintenance: + self._maybe_auto_consolidate() return except sqlite3.Error: logger.warning( @@ -400,6 +406,8 @@ def _ensure_connection(self) -> None: with self._init_lock: if self.conn is not None: + if not _suppress_maintenance: + self._maybe_auto_consolidate() return # Create parent directory if using a real path @@ -581,7 +589,8 @@ def _bg_wrapper(db_path): self._purge_legacy_entity_profile_summaries() self.ready = True - self._maybe_startup_consolidate() + if not _suppress_maintenance: + self._maybe_startup_consolidate() def _purge_legacy_entity_profile_summaries(self) -> None: """Delete legacy ``period='entity_profile'`` summary rows once. @@ -656,25 +665,8 @@ def _purge_legacy_entity_profile_summaries(self) -> None: ) def _maybe_startup_consolidate(self) -> None: - """Trigger background consolidation on startup if data looks stale.""" - if not self._has_consolidation or self.conn is None: - return - try: - msg_count = self.conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0] - if msg_count < self._auto_consolidate_threshold: - return - cluster_count = self.conn.execute( - "SELECT COUNT(*) FROM message_clusters" - ).fetchone()[0] - if cluster_count == 0: - self._consolidation_thread = threading.Thread( - target=self._bg_consolidate, - daemon=True, - name="startup-consolidate", - ) - self._consolidation_thread.start() - except Exception: - pass + """Use durable per-layer provenance, including successful empty work.""" + self._maybe_auto_consolidate() # ────────────────────────────────────────────────────────────────────── # Production CRUD API @@ -822,30 +814,54 @@ def add( "metadata": metadata or {}, } + def _get_maintenance_coordinator(self) -> MaintenanceCoordinator: + from truememory.maintenance import get_coordinator + coordinator = self._maintenance_coordinator + if coordinator is not None: + try: + coordinator.status + except RuntimeError: + coordinator = None + if coordinator is None or self._maintenance_handle is not self.conn: + coordinator = get_coordinator(self.db_path) + coordinator.refresh_capabilities() + self._maintenance_coordinator = coordinator + self._maintenance_handle = self.conn + return coordinator + def _maybe_auto_consolidate(self) -> None: - """Trigger background consolidation after threshold adds.""" - self._adds_since_consolidation += 1 - if self._adds_since_consolidation < self._auto_consolidate_threshold: - return - if not self._has_consolidation: + """Probe fixed metadata without waiting on foreground or model work.""" + if not self._has_consolidation or self.conn is None: return - if (self._consolidation_thread is not None - and self._consolidation_thread.is_alive()): + if not self._write_lock.acquire(blocking=False): + self._maintenance_pending_reason = "foreground_busy" return - self._adds_since_consolidation = 0 - self._consolidation_thread = threading.Thread( - target=self._bg_consolidate, - daemon=True, - name="auto-consolidate", - ) - self._consolidation_thread.start() + coordinator = None + eligible = () + try: + if self.conn is None: + return + if self.conn.in_transaction: + self._maintenance_pending_reason = "pending_caller_commit" + return + from truememory.maintenance import engine_layer_specs, plan_layers + coordinator = self._get_maintenance_coordinator() + eligible = plan_layers(self.conn, engine_layer_specs(self.conn, coordinator), + threshold=self._auto_consolidate_threshold) + self._maintenance_pending_reason = "eligible" if eligible else "awaiting_eligibility" + except Exception as error: + self._maintenance_pending_reason = type(error).__name__[:64] + finally: + self._write_lock.release() + if eligible and coordinator is not None: + try: + coordinator.request_layers(threshold=self._auto_consolidate_threshold) + except Exception as error: + self._maintenance_pending_reason = type(error).__name__[:64] def _bg_consolidate(self) -> None: - """Run consolidation in a background thread with its own connection.""" - try: - self.consolidate() - except Exception: - logger.debug("Auto-consolidation failed", exc_info=True) + """Compatibility entry point; the coordinator owns the worker handle.""" + self._maybe_auto_consolidate() def delete(self, memory_id: int) -> bool: """Delete a memory by ID. @@ -854,7 +870,10 @@ def delete(self, memory_id: int) -> bool: """ self._ensure_connection() with self._write_lock: - return delete_message(self.conn, memory_id) + deleted = delete_message(self.conn, memory_id) + if deleted: + self._maybe_auto_consolidate() + return deleted def delete_all(self, user_id: str | None = None) -> bool: """Delete all memories, optionally filtered by user. @@ -1021,147 +1040,51 @@ def delete_all(self, user_id: str | None = None) -> bool: logger.warning("Failed to rebuild FTS index during delete_all", exc_info=True) self.conn.commit() - return deleted + if deleted: + self._maybe_auto_consolidate() + return deleted + + def _apply_manual_maintenance_capabilities(self, results: tuple[LayerResult, ...]) -> None: + completed = {result.layer for result in results + if result.outcome in {"success", "success_empty"}} + if "summaries" in completed: + self._has_consolidation = True + if "clusters" in completed: + self._has_clustering = True def consolidate(self) -> dict[str, str]: - """Run all consolidation layers (L0-L5) under the write lock. - - Returns timing stats for each step. - """ - import time as _time - - self._ensure_connection() - stats: dict[str, str] = {} - - try: - from truememory.consolidation import ( - build_summaries, - detect_contradictions, - build_structured_facts, - ) - except (ImportError, ModuleNotFoundError): - stats["consolidation"] = "SKIPPED (module not available)" - return stats - - try: - from truememory.predictive import build_surprise_index - except (ImportError, ModuleNotFoundError): - build_surprise_index = None - - try: - from truememory.temporal import detect_episodes, detect_landmark_events - except (ImportError, ModuleNotFoundError): - detect_episodes = None - detect_landmark_events = None - - try: - from truememory.personality import build_dunbar_hierarchy - except (ImportError, ModuleNotFoundError): - build_dunbar_hierarchy = None - - _cluster_messages = None - if _HAS_CLUSTERING and self._has_vectors: - try: - from truememory.clustering import cluster_messages as _cm - _cluster_messages = _cm - except (ImportError, ModuleNotFoundError): - pass - - _extract_preferences = None - if _HAS_PERSONALITY: - try: - from truememory.personality import extract_preferences as _ep - _extract_preferences = _ep - except (ImportError, ModuleNotFoundError): - pass - + """Force one pass, preserving caller transactions and shared ownership.""" + from truememory.maintenance import ( + MaintenanceBusyError, format_maintenance_report, maintenance_busy_result, + maintenance_owner, run_engine_maintenance, + ) + self._ensure_connection(_suppress_maintenance=True) with self._write_lock: - if _cluster_messages: - try: - t0 = _time.time() - n = _cluster_messages(self.conn) - stats["cluster_messages"] = f"{n} clusters in {_time.time() - t0:.3f}s" - self._has_clustering = True - except Exception as exc: - stats["cluster_messages"] = f"ERROR: {exc}" - - if _extract_preferences: - try: - t0 = _time.time() - _extract_preferences(self.conn) - stats["extract_preferences"] = f"{_time.time() - t0:.3f}s" - except Exception as exc: - stats["extract_preferences"] = f"ERROR: {exc}" - - try: - t0 = _time.time() - build_summaries(self.conn) - stats["build_summaries"] = f"{_time.time() - t0:.3f}s" - except Exception as exc: - stats["build_summaries"] = f"ERROR: {exc}" - - try: - t0 = _time.time() - detect_contradictions(self.conn) - stats["detect_contradictions"] = f"{_time.time() - t0:.3f}s" - except Exception as exc: - stats["detect_contradictions"] = f"ERROR: {exc}" - - try: - t0 = _time.time() - n = build_structured_facts(self.conn) - stats["structured_facts"] = f"{n} facts in {_time.time() - t0:.3f}s" - except Exception as exc: - stats["structured_facts"] = f"ERROR: {exc}" - - if build_surprise_index: - try: - t0 = _time.time() - build_surprise_index(self.conn) - stats["build_surprise_index"] = f"{_time.time() - t0:.3f}s" - except Exception as exc: - stats["build_surprise_index"] = f"ERROR: {exc}" - - if detect_episodes: - try: - t0 = _time.time() - ep = detect_episodes(self.conn) - stats["detect_episodes"] = f"{ep} episodes in {_time.time() - t0:.3f}s" - except Exception as exc: - stats["detect_episodes"] = f"ERROR: {exc}" - - if detect_landmark_events: + coordinator = self._get_maintenance_coordinator() + coordinator.refresh_capabilities() + borrowed = self.conn.in_transaction + if borrowed or coordinator.path is None: try: - t0 = _time.time() - lm = detect_landmark_events(self.conn) - stats["detect_landmarks"] = f"{lm} events in {_time.time() - t0:.3f}s" - except Exception as exc: - stats["detect_landmarks"] = f"ERROR: {exc}" - - if build_dunbar_hierarchy: + report = run_engine_maintenance(self.conn, coordinator, force=True, + threshold=self._auto_consolidate_threshold, prepare_extensions=False, + allow_caller_transaction=borrowed) + except MaintenanceBusyError: + return maintenance_busy_result() + self._apply_manual_maintenance_capabilities(report.results) + return format_maintenance_report(report) + try: + with maintenance_owner(coordinator.path): + connection = create_db(coordinator.path) try: - t0 = _time.time() - primary = None - try: - row = self.conn.execute( - "SELECT sender, COUNT(*) as cnt FROM messages " - "WHERE sender != '' AND sender IS NOT NULL " - "GROUP BY sender ORDER BY cnt DESC LIMIT 1" - ).fetchone() - if row and row[0] and row[0].strip(): - primary = row[0] - except Exception: - pass - result = build_dunbar_hierarchy(self.conn, primary_entity=primary) - n_rel = len(result) if isinstance(result, dict) else result - stats["dunbar_hierarchy"] = f"{n_rel} relationships in {_time.time() - t0:.3f}s" - except Exception as exc: - stats["dunbar_hierarchy"] = f"ERROR: {exc}" - - self.conn.commit() - - self._has_consolidation = True - return stats + report = run_engine_maintenance(connection, coordinator, force=True, + threshold=self._auto_consolidate_threshold) + finally: + connection.close() + except MaintenanceBusyError: + return maintenance_busy_result() + with self._write_lock: + self._apply_manual_maintenance_capabilities(report.results) + return format_maintenance_report(report) def update(self, memory_id: int, content: str | None = None, **fields) -> dict | None: """Update a memory. @@ -1205,41 +1128,47 @@ def update(self, memory_id: int, content: str | None = None, **fields) -> dict | except Exception: logger.debug("Failed to pre-compute embedding during update()", exc_info=True) - with self._write_lock: - if content is not None: - fields["content"] = content + source_committed = False + try: + with self._write_lock: + if content is not None: + fields["content"] = content - ok = update_message(self.conn, memory_id, **fields) - if not ok: - return None + ok = update_message(self.conn, memory_id, **fields) + source_committed = ok and not self.conn.in_transaction + if not ok: + return None - if pre_embedding is not None: - from truememory.vector_search import VectorPublicationChanged, _foreground_vector_publication - try: - from truememory.vector_search import serialize_f32, _write_embedder_metadata_no_commit - with _foreground_vector_publication(self.conn, pre_identity, memory_id) as (vec_tbl, sep_tbl): - try: - self.conn.execute(f"DELETE FROM {vec_tbl} WHERE rowid = ?", (memory_id,)) - except Exception: - logger.debug("Failed to delete old vector embedding for message %d", memory_id, exc_info=True) - self.conn.execute( - f"INSERT INTO {vec_tbl}(rowid, embedding) VALUES (?, ?)", - (memory_id, serialize_f32(pre_embedding)), - ) - if pre_sep_embedding is not None: + if pre_embedding is not None: + from truememory.vector_search import VectorPublicationChanged, _foreground_vector_publication + try: + from truememory.vector_search import serialize_f32, _write_embedder_metadata_no_commit + with _foreground_vector_publication(self.conn, pre_identity, memory_id) as (vec_tbl, sep_tbl): try: - self.conn.execute(f"DELETE FROM {sep_tbl} WHERE rowid = ?", (memory_id,)) + self.conn.execute(f"DELETE FROM {vec_tbl} WHERE rowid = ?", (memory_id,)) except Exception: - logger.debug("Failed to delete old sep vector for message %d", memory_id, exc_info=True) + logger.debug("Failed to delete old vector embedding for message %d", memory_id, exc_info=True) self.conn.execute( - f"INSERT INTO {sep_tbl}(rowid, embedding) VALUES (?, ?)", - (memory_id, serialize_f32(pre_sep_embedding)), + f"INSERT INTO {vec_tbl}(rowid, embedding) VALUES (?, ?)", + (memory_id, serialize_f32(pre_embedding)), ) - _write_embedder_metadata_no_commit(self.conn) - except VectorPublicationChanged as exc: - raise VectorPublicationChanged("Source update committed; retry update to publish vectors") from exc - except Exception: - logger.warning("Vector embedding failed for message %d", memory_id, exc_info=True) + if pre_sep_embedding is not None: + try: + self.conn.execute(f"DELETE FROM {sep_tbl} WHERE rowid = ?", (memory_id,)) + except Exception: + logger.debug("Failed to delete old sep vector for message %d", memory_id, exc_info=True) + self.conn.execute( + f"INSERT INTO {sep_tbl}(rowid, embedding) VALUES (?, ?)", + (memory_id, serialize_f32(pre_sep_embedding)), + ) + _write_embedder_metadata_no_commit(self.conn) + except VectorPublicationChanged as exc: + raise VectorPublicationChanged("Source update committed; retry update to publish vectors") from exc + except Exception: + logger.warning("Vector embedding failed for message %d", memory_id, exc_info=True) + finally: + if source_committed: + self._maybe_auto_consolidate() return self.get(memory_id) @@ -2567,6 +2496,28 @@ def get_stats(self) -> dict: """Return ingestion and search statistics.""" self._ensure_connection() stats = dict(self.stats) + if self._write_lock.acquire(blocking=False): + try: + from truememory.maintenance import engine_layer_specs, layer_freshness, read_layer_states + coordinator = self._get_maintenance_coordinator() + maintenance = coordinator.snapshot() + maintenance["last_results"] = [result._asdict() for result in maintenance["last_results"]] + maintenance["pending_reason"] = self._maintenance_pending_reason + maintenance["pending_caller_commit"] = self.conn.in_transaction + states = read_layer_states(self.conn, engine_layer_specs(self.conn, coordinator)) + maintenance["layers"] = {name: { + "outcome": state.outcome, "freshness": layer_freshness(state.source, state, state.dependency), + "coverage": state.successful_coverage, "error_category": state.error_category, + "dependency_error": state.dependency.error_category, "output_count": state.output_count, + "pending_inserts": state.source.insert_count - (state.attempted_insert_count or 0), + } for name, state in states.items()} + stats["maintenance"] = maintenance + except Exception as error: + stats["maintenance"] = {"status": "unavailable", "error_category": type(error).__name__[:64]} + finally: + self._write_lock.release() + else: + stats["maintenance"] = {"status": "pending", "pending_reason": "foreground_busy"} # Add live DB stats if connected. if self.conn: @@ -2589,19 +2540,18 @@ def get_stats(self) -> dict: return stats def close(self): - """Close database connection and join background threads.""" - if self._consolidation_thread is not None and self._consolidation_thread.is_alive(): - try: - self._consolidation_thread.join(timeout=5) - except Exception: - pass - if self.conn: - try: - self.conn.close() - except Exception: - logger.debug("Failed to close database connection", exc_info=True) - self.conn = None - self.ready = False + """Detach this engine; a shared worker retains its own connection.""" + with self._init_lock, self._write_lock: + self._maintenance_coordinator = None + self._maintenance_handle = None + self._maintenance_pending_reason = "not_connected" + if self.conn: + try: + self.conn.close() + except sqlite3.Error: + logger.debug("Failed to close database connection", exc_info=True) + self.conn = None + self.ready = False def __enter__(self): return self diff --git a/truememory/maintenance.py b/truememory/maintenance.py index 325ac2c8..670f0202 100644 --- a/truememory/maintenance.py +++ b/truememory/maintenance.py @@ -1,7 +1,6 @@ -"""Database maintenance ownership and committed source revision primitives. +"""Database maintenance ownership, source revisions and durable layer scheduling. -The engine scheduler is not routed here yet. No models, polling loop or automatic -builder invocation is introduced by importing this module. +Importing this module loads no models and starts no worker or polling loop. """ import hashlib @@ -104,27 +103,41 @@ def connection_database_path(conn: sqlite3.Connection) -> Path | None: _coordinators: weakref.WeakValueDictionary = weakref.WeakValueDictionary() _held_fds: set[int] = set() _held_paths: dict[Path, int] = {} +_owner_waiters: dict[Path, tuple[int, int, "MaintenanceCoordinator"]] = {} _thread_owners = threading.local() -def _acquire_owner(path: Path) -> int | None: +def _acquire_owner(path: Path, on_busy: Callable[[], None] | None = None) -> int | None: with _registry_lock: if path in _held_paths: + if on_busy is not None: + on_busy() return None fd = try_file_lock(Path(str(path) + ".maintenance.lock")) if fd is not None: _held_fds.add(fd) _held_paths[path] = fd + elif on_busy is not None: + on_busy() return fd def _release_owner(fd: int) -> None: + notify = None with _registry_lock: os.close(fd) _held_fds.discard(fd) for path, claimed_fd in tuple(_held_paths.items()): if claimed_fd == fd: del _held_paths[path] + waiter = _owner_waiters.get(path) + if waiter is not None and waiter[0] == fd: + del _owner_waiters[path] + notify = waiter[2] + else: + notify = _coordinators.get(path) + if notify is not None: + notify._owner_released() @contextmanager @@ -169,13 +182,14 @@ def maintenance_owner(db_path: str | os.PathLike[str] | None) -> Iterator[Mainte def _after_fork() -> None: - global _registry_lock, _coordinators, _held_fds, _held_paths, _thread_owners + global _registry_lock, _coordinators, _held_fds, _held_paths, _owner_waiters, _thread_owners # The child must not prolong its parent's ownership by retaining an # inherited descriptor. Inherited coordinator objects reject child use. for fd in _held_fds: os.close(fd) _held_fds = set() _held_paths = {} + _owner_waiters = {} _thread_owners = threading.local() _coordinators = weakref.WeakValueDictionary() _registry_lock = threading.Lock() @@ -203,6 +217,12 @@ def __init__(self, path: Path | None) -> None: self._done.set() self._status = "pending" self._error_category: str | None = None + self._notification_generation = 0 + self._pending: tuple[int, int] | None = None + self._last_report = None + self._capability_epoch = 0 + self._capability_versions: tuple[tuple[str, str | None], ...] | None = None + self._extension_failure: LayerDependency | None = None def _check_process(self) -> None: if self._pid != os.getpid(): @@ -214,6 +234,83 @@ def status(self) -> tuple[str, str | None]: with self._mutex: return self._status, self._error_category + def snapshot(self) -> dict: + self._check_process() + with self._mutex: + return {"status": self._status, "error_category": self._error_category, + "active": self._active, "pending_generations": int(self._pending is not None), + "capability_epoch": self._capability_epoch, + "last_results": self._last_report.results if self._last_report is not None else ()} + + def refresh_capabilities(self, *, reset_extension: bool = True) -> None: + """Refresh three package records at open/manual setup, never per add.""" + self._check_process() + versions = tuple((module, _installed_dependency(module, distribution)) for module, distribution in ( + ("numpy", "numpy"), ("hdbscan", "hdbscan"), ("sqlite_vec", "sqlite-vec"), + )) + with self._mutex: + self._capability_epoch += 1 + if reset_extension or versions != self._capability_versions: + self._extension_failure = None + self._capability_versions = versions + + def capability_snapshot(self) -> tuple[int, tuple, "LayerDependency | None"]: + self._check_process() + with self._mutex: + return self._capability_epoch, self._capability_versions or (), self._extension_failure + + def _publish_extension_evidence(self, epoch: int, dependency: "LayerDependency | None") -> None: + with self._mutex: + if epoch == self._capability_epoch: + self._extension_failure = dependency + + def _reserve_locked(self) -> None: + self._active = True + self._done.clear() + self._cancel.clear() + self._status = "starting" + self._error_category = None + + def request_layers(self, *, threshold: int = 25) -> bool: + """Coalesce a real foreground wake; no callback captures its engine.""" + self._check_process() + if type(threshold) is not int or threshold < 1: + raise ValueError("Maintenance threshold must be a positive integer") + with self._mutex: + if self.path is None: + self._status = "pending_in_memory" + return False + if self._active and self._cancel.is_set(): + return False + self._notification_generation += 1 + self._pending = (self._notification_generation, threshold) + pending = self._take_pending_locked() + return self._launch_layers(pending) if pending is not None else False + + def _take_pending_locked(self) -> tuple[int, int] | None: + if self._active or self._pending is None: + return None + pending, self._pending = self._pending, None + self._reserve_locked() + return pending + + def _launch_layers(self, pending: tuple[int, int]) -> bool: + def work(conn: sqlite3.Connection, cancel: threading.Event) -> "MaintenanceReport": + return run_engine_maintenance(conn, self, threshold=pending[1], cancel=cancel) + return self._launch(work, pending) + + def _owner_released(self) -> None: + self._check_process() + with self._mutex: + pending = self._take_pending_locked() + if pending is not None: + try: + self._launch_layers(pending) + except Exception: + # The launch path records a bounded failure. Notification must + # not replace a synchronous owner's unrelated operation error. + pass + def request(self, work: Callable[[sqlite3.Connection, threading.Event], None]) -> bool: """Start one owned worker, or report busy/in-memory pending without retry.""" self._check_process() @@ -223,11 +320,21 @@ def request(self, work: Callable[[sqlite3.Connection, threading.Event], None]) - return False if self._active: return False - self._active = True - self._done.clear() - self._cancel.clear() - self._status = "starting" - self._error_category = None + self._reserve_locked() + return self._launch(work) + + def _launch(self, work: Callable, pending: tuple[int, int] | None = None) -> bool: + def busy() -> None: + # Registry -> coordinator ordering closes the release/admission + # race. No coordinator-mutex holder acquires the registry lock. + with self._mutex: + self._active = False + self._status = "cancelled" if self._cancel.is_set() else "busy" + if pending is not None and self._pending is None and not self._cancel.is_set(): + self._pending = pending + if self._pending is not None and self.path in _held_paths: + _owner_waiters[self.path] = (_held_paths[self.path], self._pending[0], self) + self._done.set() fd = None handoff = None @@ -235,12 +342,8 @@ def request(self, work: Callable[[sqlite3.Connection, threading.Event], None]) - try: # Serializing descriptor acquisition with fork registration makes # every inherited owner descriptor visible to the child cleanup. - fd = _acquire_owner(self.path) + fd = _acquire_owner(self.path, on_busy=busy) if fd is None: - with self._mutex: - self._active = False - self._status = "busy" - self._done.set() return False handoff = {"fd": fd, "started": False} @@ -266,7 +369,10 @@ def run() -> None: if started: # start() can be interrupted after launching the target. That # worker owns cleanup until it really finishes, even on error. - self._cancel.set() + with self._mutex: + if self._active and self._thread is worker: + self._pending = None + self._cancel.set() raise if fd is not None: _release_owner(fd) @@ -275,7 +381,14 @@ def run() -> None: self._thread = None self._status = "failed" self._error_category = "worker_start" - self._done.set() + successor = self._take_pending_locked() + if successor is None: + self._done.set() + if successor is not None: + try: + self._launch_layers(successor) + except Exception: + pass raise def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], None]) -> None: @@ -287,11 +400,15 @@ def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], No try: if not self._cancel.is_set(): conn = create_db(self.path) - work(conn, self._cancel) + report = work(conn, self._cancel) if conn.in_transaction: conn.rollback() raise RuntimeError("Maintenance work left an unfinished transaction") outcome = "cancelled" if self._cancel.is_set() else "success" + if isinstance(report, MaintenanceReport): + outcome, error_category = maintenance_report_status(report, self._cancel) + with self._mutex: + self._last_report = report finally: try: if conn is not None: @@ -307,14 +424,29 @@ def _run(self, fd: int, work: Callable[[sqlite3.Connection, threading.Event], No _release_owner(fd) with self._mutex: self._active = False + self._thread = None self._status = outcome self._error_category = error_category - self._done.set() + if self._cancel.is_set(): + self._pending = None + pending = self._take_pending_locked() + if pending is None: + self._done.set() + if pending is not None: + try: + self._launch_layers(pending) + except Exception: + pass # Launch already recorded the categorical failure. def cancel(self) -> None: """Signal a phase boundary; never release ownership of active work.""" self._check_process() - self._cancel.set() + with _registry_lock, self._mutex: + self._pending = None + self._cancel.set() + waiter = _owner_waiters.get(self.path) + if waiter is not None and waiter[2] is self: + del _owner_waiters[self.path] def wait(self, timeout: float | None = None) -> bool: """Wait for this worker if requested; report whether teardown completed.""" @@ -860,12 +992,15 @@ def _installed_dependency(module: str, distribution: str) -> str | None: return None -def _cluster_schedule_dependency(conn: sqlite3.Connection) -> LayerDependency: +def _cluster_schedule_dependency( + conn: sqlite3.Connection, *, versions: dict[str, str | None] | None = None, +) -> LayerDependency: parameters = {"min_cluster_size": 10, "min_samples": 5, "metric": "euclidean", "cluster_selection_method": "eom"} - versions = {module: _installed_dependency(module, distribution) for module, distribution in ( - ("numpy", "numpy"), ("hdbscan", "hdbscan"), ("sqlite_vec", "sqlite-vec"), - )} + if versions is None: + versions = {module: _installed_dependency(module, distribution) for module, distribution in ( + ("numpy", "numpy"), ("hdbscan", "hdbscan"), ("sqlite_vec", "sqlite-vec"), + )} parameters["versions"] = versions def unavailable(category: str) -> LayerDependency: @@ -1042,3 +1177,136 @@ def dunbar_dependency() -> LayerDependency: return tuple(specs[name] for name in ( "clusters", "summaries", "contradictions", "structured_facts", "surprise", "episodes", "landmarks", "dunbar", )) + + +class MaintenanceReport(NamedTuple): + results: tuple[LayerResult, ...] + preferences: str + + +def _dependency_parameters(dependency: LayerDependency) -> dict | None: + if dependency.deferred: + return None + return json.loads(dependency.key)["parameters"] + + +def engine_layer_specs( + conn: sqlite3.Connection, coordinator: MaintenanceCoordinator, *, evidence: tuple | None = None, +) -> tuple[LayerSpec, ...]: + """Small live probes plus one bounded worker-capability evidence record.""" + _, versions, failure = evidence if evidence is not None else coordinator.capability_snapshot() + + def resolve() -> LayerDependency: + dependency = _cluster_schedule_dependency(conn, versions=dict(versions)) + if (failure is not None and not dependency.deferred + and _dependency_parameters(dependency) == _dependency_parameters(failure)): + return failure + return dependency + + return tuple(spec._replace(resolve_dependency=resolve) if spec.layer == "clusters" else spec + for spec in all_layer_specs(conn)) + + +def _prepare_worker_extensions(conn: sqlite3.Connection, coordinator: MaintenanceCoordinator) -> tuple: + coordinator.refresh_capabilities(reset_extension=False) + epoch, versions, previous_failure = coordinator.capability_snapshot() + failed = False + enabled = False + try: + if dict(versions).get("sqlite_vec") is None: + failed = True + else: + module = importlib.import_module("sqlite_vec") + conn.enable_load_extension(True) + enabled = True + module.load(conn) + except Exception: + failed = True + finally: + if enabled: + conn.enable_load_extension(False) + dependency = _cluster_schedule_dependency(conn, versions=dict(versions)) + failure = previous_failure if failed else None + if failed and not dependency.deferred: + parameters = _dependency_parameters(dependency) + if parameters is not None and "identity_sha256" in parameters: + failure = make_layer_dependency(dependency.builder_version, parameters, + available=False, error_category="VectorExtensionUnavailable") + elif dependency.error_category == "VectorExtensionUnavailable": + failure = dependency + coordinator._publish_extension_evidence(epoch, failure) + return epoch, versions, failure + + +def run_engine_maintenance( + conn: sqlite3.Connection, coordinator: MaintenanceCoordinator, *, threshold: int = 25, + force: bool = False, cancel: threading.Event | None = None, + prepare_extensions: bool = True, allow_caller_transaction: bool = False, +) -> MaintenanceReport: + """Use only the supplied owned worker/original explicitly borrowed handle.""" + evidence = _prepare_worker_extensions(conn, coordinator) if prepare_extensions else None + specs = engine_layer_specs(conn, coordinator, evidence=evidence) + results = run_layers(conn, specs, threshold=threshold, force=force, cancel=cancel, + allow_caller_transaction=allow_caller_transaction) + preferences = "SKIPPED (no maintenance attempt)" + if cancel is not None and cancel.is_set(): + preferences = "CANCELLED" + elif force or any(result.attempted for result in results): + started = time.monotonic() + try: + importlib.import_module("truememory.personality").extract_preferences(conn) + except (ImportError, AttributeError): + preferences = "UNAVAILABLE (DependencyMissing)" + except Exception as error: + preferences = "ERROR (" + type(error).__name__[:64] + ")" + else: + preferences = f"{time.monotonic() - started:.3f}s" + if conn.in_transaction and allow_caller_transaction: + preferences += " (pending caller commit)" + return MaintenanceReport(results, preferences) + + +def maintenance_report_status(report: MaintenanceReport, cancel: threading.Event | None = None) -> tuple[str, str | None]: + if cancel is not None and cancel.is_set(): + return "cancelled", "Cancelled" + for outcome in ("failed", "unavailable", "deferred", "abandoned"): + for result in report.results: + if result.outcome == outcome: + return outcome, result.error_category + if report.preferences.startswith(("ERROR", "UNAVAILABLE")): + return "failed", "PreferencesUnavailable" + if any(result.outcome not in {"success", "success_empty", "current"} for result in report.results): + return "pending", None + if any(result.coverage != "complete" for result in report.results): + return "completed_with_limits", None + return "success", None + + +_MAINTENANCE_RESULT_KEYS = ( + "cluster_messages", "build_summaries", "detect_contradictions", "structured_facts", + "build_surprise_index", "detect_episodes", "detect_landmarks", "dunbar_hierarchy", "extract_preferences", +) + + +def maintenance_busy_result() -> dict[str, str]: + return dict.fromkeys(_MAINTENANCE_RESULT_KEYS, "BUSY (maintenance owner active)") + + +def format_maintenance_report(report: MaintenanceReport) -> dict[str, str]: + counts = {"clusters": "clusters", "structured_facts": "facts", "episodes": "episodes", + "landmarks": "events", "dunbar": "relationships"} + output = {} + for result in report.results: + if result.outcome in {"success", "success_empty"}: + prefix = f"{result.output_count} {counts[result.layer]} in " if result.layer in counts else "" + value = prefix + f"{result.elapsed_seconds:.3f}s" + if result.coverage != "complete": + value += " (" + result.coverage + ")" + else: + label = "ERROR" if result.outcome == "failed" else result.outcome.upper() + value = label + (" (" + result.error_category + ")" if result.error_category else "") + if result.pending_caller_commit: + value += " (pending caller commit)" + output[result.result_key] = value + output["extract_preferences"] = report.preferences + return output From 4431787bfe18ece1f3609ef4d827917976b8095b Mon Sep 17 00:00:00 2001 From: buildingjoshbetter <236675935+buildingjoshbetter@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:55:28 -0500 Subject: [PATCH 3/3] test: Isolate maintenance fixtures from detached workers --- tests/test-maintenance-adapters-753.py | 6 +- .../test-maintenance-fixture-isolation-753.py | 220 ++++++++++++++++++ tests/test_issue_498_auto_consolidate.py | 10 + 3 files changed, 235 insertions(+), 1 deletion(-) create mode 100644 tests/test-maintenance-fixture-isolation-753.py diff --git a/tests/test-maintenance-adapters-753.py b/tests/test-maintenance-adapters-753.py index a0454f79..ebc0601c 100644 --- a/tests/test-maintenance-adapters-753.py +++ b/tests/test-maintenance-adapters-753.py @@ -57,7 +57,11 @@ def setUp(self) -> None: ("vec_source_v1:vec_messages_edge", '{"version":1,"generation":"synthetic-a","cursor":0,"complete":true}'), ]) self.conn.commit() - self.enterContext(patch.dict(sys.modules, {"truememory.vector_search": self.vector})) + # Detached production workers must not discover this fixture's model + # or hold its lock through the process-wide module registry. + self.enterContext(patch.object(MAINTENANCE, "sys", types.SimpleNamespace( + modules={"truememory.vector_search": self.vector}, version_info=sys.version_info, + ))) self.enterContext(patch.object(MAINTENANCE, "importlib", types.SimpleNamespace( import_module=lambda name: self.modules[name], metadata=importlib.metadata, util=importlib.util, ))) diff --git a/tests/test-maintenance-fixture-isolation-753.py b/tests/test-maintenance-fixture-isolation-753.py new file mode 100644 index 00000000..f52397c5 --- /dev/null +++ b/tests/test-maintenance-fixture-isolation-753.py @@ -0,0 +1,220 @@ +"""Cross-fixture worker isolation using actual fixtures and SQLite in memory.""" + +import ast +import builtins +import sqlite3 +import sys +import threading +import types +import unittest +import uuid +from collections.abc import Callable +from pathlib import Path +from unittest.mock import patch + + +ROOT = Path(__file__).resolve().parents[1] + + +def fixture_loader(process_sys: object) -> Callable[[str], dict]: + """Keep fixture imports and module registries inside one synthetic process.""" + loaded = {} + + def safe_import(name: str, globals: dict | None = None, locals: dict | None = None, + fromlist: tuple = (), level: int = 0) -> object: + if name == "sys": + return process_sys + if name == "runpy": + return types.SimpleNamespace(run_path=load) + if name == "truememory.personality_style_vec": + return process_sys.modules[name] + if name.startswith("truememory") or name.split(".")[0] in {"torch", "numpy", "hdbscan", "sqlite_vec"}: + raise AssertionError("Unexpected application/native import: " + name) + return builtins.__import__(name, globals, locals, fromlist, level) + + def safe_exec(code: object, namespace: dict) -> None: + namespace.setdefault("__builtins__", fixture_builtins) + exec(code, namespace) + + fixture_builtins = dict(vars(builtins), __import__=safe_import, exec=safe_exec) + + def load(path: str) -> dict: + path = Path(path) + if path.name in loaded: + return loaded[path.name] + tree = ast.parse(path.read_text(encoding="utf-8")) + # Existing fixture loaders also read production source. Pin their + # decoding in this harness on Python 3.10/Windows without editing it. + for node in ast.walk(tree): + if (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + and node.func.attr == "read_text" and not node.args + and not any(keyword.arg == "encoding" for keyword in node.keywords)): + node.keywords.append(ast.keyword(arg="encoding", value=ast.Constant(value="utf-8"))) + namespace = {"__file__": str(path), "__name__": "synthetic_fixture_" + path.stem, + "__builtins__": fixture_builtins} + exec(compile(ast.fix_missing_locations(tree), str(path), "exec"), namespace) + loaded[path.name] = namespace + return namespace + + return load + + +class TestFixtureIsolation(unittest.TestCase): + def setUp(self) -> None: + values = vars(sys).copy() + values["modules"] = {} + self.process_sys = types.SimpleNamespace(**values) + load = fixture_loader(self.process_sys) + self.routing = load(str(ROOT / "tests/test-maintenance-engine-routing-753.py")) + self.adapters = self.routing["ADAPTERS"] + self.storage = self.routing["STORAGE"] + self.maintenance = self.routing["MAINTENANCE"] + _, self.foreign_vector, _ = self.adapters["CLUSTERS"]["load_clustering"]() + self.process_sys.modules["truememory.vector_search"] = self.foreign_vector + load_stdlib = self.routing["stdlib_module"] + style = load_stdlib("personality_style_vec", {}) + self.process_sys.modules["truememory.personality_style_vec"] = style + self.enter_context(patch.dict(self.routing, {"stdlib_module": lambda name, modules: + style if name == "personality_style_vec" else load_stdlib(name, modules)})) + self.registry_before = self.process_sys.modules.copy() + # Primitive loaders have their own import guard. Bind their sys view + # explicitly so both modules start in the same synthetic process. + self.enter_context(patch.object(self.maintenance, "sys", self.process_sys)) + self.foreign = types.ModuleType("synthetic_foreign_maintenance") + self.foreign.__dict__.update(self.maintenance.__dict__) + # Functions need their own globals to represent an unrelated worker. + source = (ROOT / "truememory/maintenance.py").read_text(encoding="utf-8") + exec(compile(source, "synthetic-foreign-maintenance", "exec"), self.foreign.__dict__) + self.foreign.sys = self.process_sys + + self.uri = "file:synthetic-fixture-" + uuid.uuid4().hex + "?mode=memory&cache=shared" + self.connect = sqlite3.connect + self.keeper = self.connect(self.uri, uri=True, check_same_thread=False) + self.addCleanup(self.keeper.close) + self.create = self.storage.create_db + self.coordinator = self.maintenance.MaintenanceCoordinator(None) + fixture_test = self + + class MemoryDirectory: + name = ":memory:" + + def cleanup(self) -> None: + pass + + class MemoryFixture(self.routing["RoutingFixture"]): + def open_db(self) -> sqlite3.Connection: + conn = fixture_test.open_memory() + self.addCleanup(conn.close) + return conn + + self.enter_context(patch.dict(self.adapters, {"tempfile": types.SimpleNamespace( + TemporaryDirectory=lambda **kwargs: MemoryDirectory())})) + self.enter_context(patch.object(self.storage, "create_db", self.open_memory)) + self.enter_context(patch.object(self.maintenance, "get_coordinator", lambda path: self.coordinator)) + self.fixture = MemoryFixture() + self.addCleanup(self.fixture.doCleanups) + self.fixture.setUp() + + def enter_context(self, context: object) -> object: + result = context.__enter__() + self.addCleanup(context.__exit__, None, None, None) + return result + + def open_memory(self, path: object = None) -> sqlite3.Connection: + sqlite_view = types.SimpleNamespace(**dict(vars(sqlite3), connect=lambda *args, **kwargs: + self.connect(self.uri, uri=True, check_same_thread=False))) + with patch.object(self.storage, "sqlite3", sqlite_view): + return self.create(":memory:") + + def states(self, conn: sqlite3.Connection | None = None) -> dict: + conn = conn or self.fixture.engine.conn + specs = self.maintenance.engine_layer_specs(conn, self.coordinator) + return self.maintenance.read_layer_states(conn, specs) + + def reopen(self) -> object: + self.fixture.engine.close() + reopened = self.fixture.new_engine() + reopened._has_consolidation = True + return reopened + + def test_fixture_does_not_publish_its_vector_into_process_registry(self) -> None: + self.assertIs(self.process_sys.modules["truememory.vector_search"], self.foreign_vector) + self.assertIs(self.maintenance.sys.modules["truememory.vector_search"], self.fixture.vector) + self.assertIsNot(self.maintenance.sys.modules, self.process_sys.modules) + self.assertEqual(self.process_sys.modules, self.registry_before) + + def test_interloper_cannot_defer_bootstrap_or_reset_reopen_threshold(self) -> None: + foreign_conn = self.connect(":memory:", check_same_thread=False) + self.fixture.conn.backup(foreign_conn) + entered, release = threading.Event(), threading.Event() + observed = [] + + class BlockedSnapshot: + def __getattr__(boundary, name: str) -> object: + return getattr(foreign_conn, name) + + def execute(boundary, sql: str, *args: object) -> sqlite3.Cursor: + if sql.startswith("SELECT type,name,rootpage,sql"): + entered.set() + if not release.wait(3): + raise AssertionError("Synthetic worker was not released") + return foreign_conn.execute(sql, *args) + + worker_coordinator = self.foreign.MaintenanceCoordinator(None) + + def work(conn: sqlite3.Connection, cancel: threading.Event) -> object: + observed.append(self.foreign.sys.modules["truememory.vector_search"]) + self.foreign._cluster_schedule_dependency(conn, versions=dict(self.fixture.versions)) + return self.foreign.MaintenanceReport((), "SKIPPED (synthetic)") + + with patch.object(self.foreign, "create_db", lambda path: BlockedSnapshot()), \ + patch.object(self.foreign, "_release_owner", lambda fd: None): + worker = threading.Thread(target=worker_coordinator._run, args=(-1, work)) + worker.start() + try: + self.assertTrue(entered.wait(3)) + result = self.fixture.engine.consolidate() + states = self.states() + finally: + release.set() + worker.join(3) + foreign_conn.close() + self.assertFalse(worker.is_alive()) + self.assertEqual(worker_coordinator.status, ("success", None)) + self.assertNotIn("DEFERRED", result["cluster_messages"]) + self.assertEqual(observed, [self.foreign_vector]) + self.assertEqual(len(states), 8) + self.assertTrue(all(state.attempted_insert_count == 0 for state in states.values()), states) + self.assertTrue(all(state.outcome == "success_empty" for state in states.values()), states) + self.fixture.add(24) + before = self.maintenance.read_source_revision(self.fixture.conn) + versions = self.coordinator.capability_snapshot()[1] + reopened = self.reopen() + with patch.object(self.coordinator, "request_layers") as request: + reopened._maybe_auto_consolidate() + request.assert_not_called() + self.assertEqual(self.maintenance.read_source_revision(reopened.conn), before) + self.assertEqual(self.coordinator.capability_snapshot()[1], versions) + reopened.add("synthetic twenty fifth") + request.assert_called_once_with(threshold=25) + + def test_intentionally_deferred_bootstrap_still_recovers_at_twenty_four(self) -> None: + with self.fixture.vector._lock: + result = self.fixture.engine.consolidate() + self.assertEqual(result["cluster_messages"], "DEFERRED (ModelBusy)") + self.assertIsNone(self.states()["clusters"].attempted_insert_count) + self.fixture.add(24) + reopened = self.reopen() + with patch.object(self.coordinator, "request_layers") as request: + reopened._maybe_auto_consolidate() + request.assert_called_once_with(threshold=25) + report = self.maintenance.run_engine_maintenance( + reopened.conn, self.coordinator, threshold=25, prepare_extensions=False) + self.assertEqual([item.layer for item in report.results if item.attempted], ["clusters"]) + states = self.states(reopened.conn) + self.assertEqual(states["clusters"].attempted_insert_count, 24) + self.assertTrue(all(state.attempted_insert_count == 0 for name, state in states.items() if name != "clusters")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_issue_498_auto_consolidate.py b/tests/test_issue_498_auto_consolidate.py index 86fd1238..4295082f 100644 --- a/tests/test_issue_498_auto_consolidate.py +++ b/tests/test_issue_498_auto_consolidate.py @@ -9,11 +9,20 @@ class TestAutoConsolidation(_ROUTING["RoutingFixture"]): + def assert_empty_bootstrap(self) -> None: + maintenance = _ROUTING["MAINTENANCE"] + states = maintenance.read_layer_states( + self.conn, maintenance.engine_layer_specs(self.conn, self.coordinator)) + self.assertEqual(len(states), 8) + self.assertTrue(all(state.outcome == "success_empty" for state in states.values()), states) + self.assertTrue(all(state.attempted_insert_count == 0 for state in states.values()), states) + def test_default_threshold_is_twenty_five(self): self.assertEqual(self.engine._auto_consolidate_threshold, 25) def test_configured_threshold_uses_committed_source_inserts(self): self.engine.consolidate() + self.assert_empty_bootstrap() self.engine._auto_consolidate_threshold = 3 self.engine._has_consolidation = True with patch.object(self.coordinator, "request_layers", wraps=self.coordinator.request_layers) as request: @@ -27,6 +36,7 @@ def test_configured_threshold_uses_committed_source_inserts(self): def test_close_and_reopen_does_not_reset_threshold_progress(self): self.engine.consolidate() + self.assert_empty_bootstrap() self.add(24) self.engine.close() reopened = self.new_engine()