diff --git a/CHANGELOG.md b/CHANGELOG.md index b857532..ced2cd6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,9 @@ A write path, through an inbox you approve. - An unchanged note whose mtime or size moved (an identical save, a sync tool) now has them updated without being embedded again, so `puremind_recent` orders it correctly. - `puremind_propose` holds a lock across the duplicate check, the 100-proposal cap and the file creation, so two agents proposing at once can no longer pass the cap or save the same proposal twice. - The MCP server's instructions and `puremind connect` now say that the server cannot write to the vault but accepts proposals. The test that listed exactly three tools now lists four. +- Index runs hold a process lock beside the database until they finish. A `puremind index` that overlaps another run is refused with a clear message, and a `puremind watch` run waits for the other run to finish and then checks the model again, so a watch on an old model can no longer write its vectors into an index rebuilt for another model. A read-only index is still reported as read-only, and the lock file is never opened through a symlink. +- Rebuilding for another vault, or for a vault that moved, first has the model embed a test passage, so a model that cannot run leaves the index as it was. It then clears the old rows and records the new root in one transaction before indexing the notes. If embedding fails after that, searches no longer return the old vault's text labelled as the new vault; results stay empty until new notes are indexed. +- Replacing or deleting a note records the cleanup owed in the same transaction. A run that crashes leaves that record for the next run, and compaction keeps it until the WAL checkpoint succeeds, including when a reader holds the log open. ## 0.1.6 (2026-10-22) diff --git a/README.md b/README.md index 8e1a6c5..db791bf 100644 --- a/README.md +++ b/README.md @@ -93,7 +93,7 @@ Limits: a title of at most 200 characters, content of at most 20,000, and 100 pr - **Store.** One SQLite file, `~/.puremind/index.sqlite`, readable only by you: files and hashes, chunks, an FTS5 keyword index (used by `puremind search --lexical-only`), and float16 vectors. Vectors are loaded into a numpy matrix for dense search. - **Retrieval.** BM25 over English-stemmed words (top 50) and Qwen3-Embedding-0.6B cosine search (top 50, 1024 dimensions, with the query instruction prefix) are fused with reciprocal rank fusion (k=60). The bge-reranker-v2-m3 cross-encoder then re-scores the 50 fused candidates, reading the first 800 characters of each. `--no-rerank` skips that last step. - **Languages.** The embedder is multilingual, so search by meaning works in most languages. Keyword matching uses an English stemmer. In hybrid search, Chinese, Japanese and Korean text is matched by single characters and overlapping pairs of characters; `search --lexical-only` does not do this, and treats an unspaced run of those characters as one word. Thai, Lao, Khmer and Burmese are not split into words for keyword matching. In scripts that need more than one token per character, such as Telugu, Burmese, Khmer and Ethiopic, a full-length chunk is longer than the 2,048 tokens the embedder reads, so search by meaning misses its end (keyword search does not); raise `embed_max_seq` to about 3,200 if most of your notes are in such a script, then run `puremind index --rebuild`. -- **Watcher.** `puremind watch` indexes once, then again each time the vault has been quiet for `--debounce` seconds (default 2) after a change to a visible note or folder. Reads, hidden folders and other file types never trigger a run, and runs never overlap. It follows a vault directory that is replaced or made again (for example by a remount or a sync tool) or that cannot be read for a while, and if nothing has happened for 10 minutes it scans anyway, in case a change was missed. A run that fails is reported on one line and watching goes on, except when no later run could succeed (a vault symlink that now names another directory than the index's, or a system limit such as the inotify watch limit): then `puremind watch` stops with an error. +- **Watcher.** `puremind watch` indexes once, then again each time the vault has been quiet for `--debounce` seconds (default 2) after a change to a visible note or folder. Reads, hidden folders and other file types never trigger a run, and runs never overlap, also not with other processes: a `puremind index` started during a run on the same index stops with an error and changes nothing, and a `puremind watch` run waits for a `puremind index` that is still running, then checks the model again. It follows a vault directory that is replaced or made again (for example by a remount or a sync tool) or that cannot be read for a while, and if nothing has happened for 10 minutes it scans anyway, in case a change was missed. A run that fails is reported on one line and watching goes on, except when no later run could succeed (a vault symlink that now names another directory than the index's, or a system limit such as the inotify watch limit): then `puremind watch` stops with an error. ## Memory and speed @@ -130,7 +130,7 @@ The reranker costs about 0.2 seconds per candidate on a CPU, so with the default - **Model revisions.** The default models are pinned to `Qwen/Qwen3-Embedding-0.6B` at `97b0c614be4d77ee51c0cef4e5f07c00f9eb65b3` and `BAAI/bge-reranker-v2-m3` at `953dc6f6f85a1b2dbfca4c34a2796e7dde08d41e`. Another model uses its repository's default branch as it was at the first download, and is not refreshed afterwards unless you set a revision or clear the cache. - **Model names.** A name is a Hugging Face model id unless it starts with `/`, `./`, `../` or `~`, which mark a local directory. A directory in your working directory is never picked up by accident. Models load with `trust_remote_code` off. -- **Changing the embedding model**, `embedder_revision` or `embed_max_seq` takes `puremind index --rebuild`, since vectors from different models are not comparable, and a new window changes the vectors of long chunks. Until then `puremind index` and `puremind watch` stop with an error and change nothing, and searches are refused. A `puremind watch` keeps the model and window it started with, so restart it after the change; one left running stops instead of wiping the rebuilt index. A running `puremind serve` keeps them too, and refuses searches until it is restarted. With `--rebuild`, the new model must first load and embed a test passage, so a mistyped name leaves the existing index untouched. The index is then emptied, and searches find nothing until the run finishes. `puremind index --rebuild` with the same model, by contrast, replaces one note at a time, so searches keep working. +- **Changing the embedding model**, `embedder_revision` or `embed_max_seq` takes `puremind index --rebuild`, since vectors from different models are not comparable, and a new window changes the vectors of long chunks. Until then `puremind index` and `puremind watch` stop with an error and change nothing, and searches are refused. A `puremind watch` keeps the model and window it started with, so restart it after the change; one left running stops instead of wiping the rebuilt index. A running `puremind serve` keeps them too, and refuses searches until it is restarted. With `--rebuild`, the new model must first load and embed a test passage, so a mistyped name leaves the existing index untouched. The index is then emptied, and searches find nothing until the run finishes. `puremind index --rebuild` with the same model, by contrast, replaces one note at a time, so searches keep working. The exception is a vault at another path (one that moved, or another vault), which the index cannot tell apart: its rebuild also checks the model first, then empties the index. - **The reranker window.** 800 characters is the setting the benchmark below was measured with. Raising `rerank_chars` lets the reranker see more of each passage, at the cost of slower searches. - **Indexes from 0.1.1 or earlier** are brought up to date by the next `puremind index` or `puremind watch` run, which cuts and embeds every note again. It takes about as long as the first index did, and searches keep working meanwhile. diff --git a/src/puremind_mac/cli.py b/src/puremind_mac/cli.py index fa77acb..9d7a02f 100644 --- a/src/puremind_mac/cli.py +++ b/src/puremind_mac/cli.py @@ -139,7 +139,8 @@ def run() -> None: # always called on this thread, one run at a time purge = first[0] or (owed_since[0] is not None and now - owed_since[0] >= PURGE_AT_LEAST_EVERY) first[0] = False try: - stats = index_vault(vault, svc.store, svc.embedder, purge=purge) + # A `puremind index` still running is waited for; this run then checks the model again. + stats = index_vault(vault, svc.store, svc.embedder, purge=purge, wait=True) except (VaultMismatch, ModelMismatch) as exc: # A symlinked vault that now names another directory, or an index rebuilt for another # model since this watch read its config: every later run would fail the same way. diff --git a/src/puremind_mac/indexer.py b/src/puremind_mac/indexer.py index 2448e02..382484a 100644 --- a/src/puremind_mac/indexer.py +++ b/src/puremind_mac/indexer.py @@ -3,6 +3,7 @@ from __future__ import annotations import errno +import fcntl import hashlib import logging import os @@ -10,6 +11,7 @@ import sqlite3 import stat from collections.abc import Callable, Sequence +from contextlib import contextmanager from dataclasses import dataclass, field from pathlib import Path @@ -46,7 +48,7 @@ class EmbedderError(RuntimeError): - """The embedding model could not be loaded or failed. The index is left as it was.""" + """The embedding model could not be loaded or failed.""" class VaultMismatch(ValueError): @@ -454,6 +456,43 @@ def _same_directory(known: str, vault: Path) -> bool: return False +def _flock(fd: int, path: str, *, block: bool) -> bool: + """Take the exclusive lock on ``fd``; False if another run holds it and ``block`` is off.""" + try: + fcntl.flock(fd, fcntl.LOCK_EX if block else fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + return False + except OSError as exc: # e.g. a file system without locks: running unlocked could interleave two runs + raise StoreError(f"cannot lock {path}: {_one_line(exc)}") from exc + return True + + +@contextmanager +def _index_lock(store: Store, *, wait: bool = False): + """Hold ``.lock`` for one run. Another run is refused, or with ``wait`` waited for. + + The kernel drops the lock when its holder exits, however that happens, so none is left stale. + """ + if store.path == ":memory:": + yield + return + store._check_writable() # a read-only index says so, instead of failing on its lock file + path = str(Path(store.path).resolve()) + ".lock" + try: + fd = os.open(path, os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0), 0o600) + except OSError as exc: + raise StoreError(f"cannot open the index lock {path}: {_one_line(exc)}") from exc + try: + if not _flock(fd, path, block=False): + if not wait: + raise StoreError(f"another index run holds {path}; try again after it finishes") + log.warning("another index run holds %s; waiting for it to finish", path) + _flock(fd, path, block=True) + yield + finally: + os.close(fd) # keep the sidecar: unlinking it would let writers lock different inodes + + def index_vault( vault: Path, store: Store, @@ -462,29 +501,49 @@ def index_vault( *, should_stop: Callable[[], bool] | None = None, purge: bool = True, + wait: bool = False, ) -> IndexStats: """Bring the index in line with the vault. ``rebuild`` embeds every note again, replacing each one's rows in turn, so searches keep finding the old version meanwhile. A new embedding model, revision or configuration is refused with ``ModelMismatch`` unless ``rebuild`` is set, and then discards the index first, since - vectors from two models cannot be searched together. Notes cut by an earlier chunker are cut + vectors from two models cannot be searched together. A vault at another path (another vault, + or the same one moved) is refused with ``VaultMismatch`` unless ``rebuild`` is set, and then + also discards the old rows, recording the new root in the same transaction. Either way the + model must first embed a test passage, so one that cannot run leaves the index as it was. + One run at a time works on a database: another run is refused with ``StoreError``, or with + ``wait`` waited for. Notes cut by an earlier chunker are cut and embedded again. ``should_stop`` is polled between notes; once it returns True the run ends early without removing anything. Text a run removes or replaces leaves the tables at once. Its words leave the keyword index when the index is compacted (``Store.purge``), which takes about half a second at 50,000 chunks. With ``purge`` the run compacts at its end, also paying a compaction an earlier run - put off; without it, the run only records that one is owed (``stats.purge_deferred``) and the + put off; without it, the debt recorded with each write stays (``stats.purge_deferred``) and the caller pays it later, as ``puremind watch`` does. """ + with _index_lock(store, wait=wait): + return _index_vault(vault, store, embedder, rebuild, should_stop=should_stop, purge=purge) + + +def _index_vault( + vault: Path, + store: Store, + embedder: Embedder, + rebuild: bool, + *, + should_stop: Callable[[], bool] | None, + purge: bool, +) -> IndexStats: vault = vault.expanduser().resolve() if not vault.is_dir(): raise NotADirectoryError(f"vault directory not found: {vault}") stats = IndexStats() known_root = store.get_meta("vault_root") - if not rebuild and known_root and not _same_directory(known_root, vault) and store.stats()["files"]: + root_changed = known_root is not None and not _same_directory(known_root, vault) + if not rebuild and root_changed and store.stats()["files"]: raise VaultMismatch( f"this index belongs to {known_root}, not {vault}; use a different PUREMIND_DB " "(or db = in config.toml) for this vault, or run `puremind index --rebuild` if the vault moved" @@ -499,8 +558,8 @@ def index_vault( ) config_changed = known_model is not None and known_config is not None and known_config != config changed = model_changed or config_changed - if changed and store.stats()["files"]: - if not rebuild: # else a `watch` left on the old config and a fresh `index` wipe each other in turn + if (changed or root_changed) and store.stats()["files"]: # the rows are about to be discarded + if changed and not rebuild: # else a `watch` left on the old config and a fresh `index` wipe each other in turn if model_changed: # the larger change; a new window alongside it needs the same rebuild built = model_label(known_model) + (f" at revision {known_rev}" if known_rev else "") wanted = model_label(embedder.name) + (f" at revision {revision}" if revision else "") @@ -516,7 +575,8 @@ def index_vault( "embedding model or configuration changed (%s%s -> %s%s); rebuilding the index", known_model, f" at {known_rev}" if known_rev else "", embedder.name, f" at {revision}" if revision else "", ) - store.clear() + if changed or root_changed: + store.clear(vault_root=str(vault)) store.set_meta("embedder", model_key(embedder.name)) store.set_meta("embedder_revision", revision) store.set_meta("embedding_config", config) @@ -548,14 +608,12 @@ def index_vault( store.delete_file(rel) stats.removed += 1 emptied = True - if stats.removed or stats.updated: # clear() purges by itself + if store.purge_pending(): if purge: emptied = store.purge() else: emptied = store.defer_purge() stats.purge_deferred = True - elif purge and store.purge_pending(): # owed by a `puremind watch` that stopped before paying it - emptied = store.purge() if not emptied: log.warning( "another process was reading the index, so text deleted or replaced in this run stays in " diff --git a/src/puremind_mac/store.py b/src/puremind_mac/store.py index 7e182d8..b9c788d 100644 --- a/src/puremind_mac/store.py +++ b/src/puremind_mac/store.py @@ -370,13 +370,26 @@ def purge(self) -> bool: FTS5 only marks a deleted row's terms as deleted and drops them when it merges the parts of its index, so the parts are merged now; secure_delete zeroes the pages this frees. - Returns what ``checkpoint`` returns. + Returns what ``checkpoint`` returns. Clearing the debt afterwards may leave metadata + in the -wal file; a concurrent write keeps the debt for the next purge. """ self._check_writable() with self.db: self.db.execute("INSERT INTO chunks_fts(chunks_fts) VALUES ('optimize')") - self.db.execute("DELETE FROM meta WHERE key = 'purge_pending'") - return self.checkpoint() + generation = self.generation() + if not self.checkpoint(): + return False + with self.db: + # A writer after compaction owes another purge, even if the checkpoint included its pages. + self.db.execute( + "DELETE FROM meta WHERE key = 'purge_pending' " + "AND COALESCE((SELECT value FROM meta WHERE key = 'generation'), '0') = ?", + (str(generation),), + ) + return True + + def _owe_purge(self) -> None: + self.db.execute("INSERT INTO meta(key, value) VALUES ('purge_pending', '1') ON CONFLICT(key) DO NOTHING") def defer_purge(self) -> bool: """Note in the index that ``purge`` is still owed, so that a process that stops before @@ -385,7 +398,7 @@ def defer_purge(self) -> bool: holds their words. Returns what ``checkpoint`` returns.""" self._check_writable() with self.db: - self.db.execute("INSERT INTO meta(key, value) VALUES ('purge_pending', '1') ON CONFLICT(key) DO NOTHING") + self._owe_purge() return self.checkpoint() def purge_pending(self) -> bool: @@ -453,7 +466,8 @@ def replace_file( if chunks and (vectors is None or len(vectors) != len(chunks)): raise ValueError("vectors must have one row per chunk") with self.db: - self.db.execute("DELETE FROM files WHERE path = ?", (path,)) + if self.db.execute("DELETE FROM files WHERE path = ?", (path,)).rowcount: + self._owe_purge() self.db.execute( "INSERT INTO files(path, sha256, mtime, size, title, indexed_at, chunker) VALUES (?,?,?,?,?,?,?)", (path, sha256, mtime, size, title, time.time(), chunker), @@ -482,21 +496,28 @@ def update_file_metadata(self, path: str, sha256: str, mtime: float, size: int) def delete_file(self, path: str) -> None: self._check_writable() with self.db: - self.db.execute("DELETE FROM files WHERE path = ?", (path,)) + if self.db.execute("DELETE FROM files WHERE path = ?", (path,)).rowcount: + self._owe_purge() self._bump() - def clear(self) -> None: - """Remove every note, and every trace of their text from the file. Searches meanwhile find nothing.""" + def clear(self, *, vault_root: str | None = None) -> None: + """Remove every note and adopt ``vault_root`` in one transaction, then purge their text.""" self._check_writable() with self.db: self.db.execute("DELETE FROM files") self.db.execute("INSERT INTO chunks_fts(chunks_fts) VALUES ('rebuild')") # empties the keyword index outright - self.db.execute("DELETE FROM meta WHERE key = 'purge_pending'") + self._owe_purge() + if vault_root is not None: + self.db.execute( + "INSERT INTO meta(key, value) VALUES ('vault_root', ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (vault_root,), + ) self._bump() if self.path != ":memory:": with self.db: # no transaction is open here, which VACUUM needs self.db.execute("VACUUM") # also drops pages an earlier version freed without zeroing - self.checkpoint() + self.purge() def recent(self, since: float, limit: int) -> list[FileRow]: rows = self.db.execute( diff --git a/tests/test_indexer.py b/tests/test_indexer.py index 69e8ef4..6b8984a 100644 --- a/tests/test_indexer.py +++ b/tests/test_indexer.py @@ -1,6 +1,8 @@ import errno +import fcntl import hashlib import logging +import multiprocessing import os import sqlite3 import sys @@ -15,13 +17,14 @@ from puremind_mac.indexer import ( EMBED_BATCH, MAX_PATH_CHARS, + PROBE_TEXT, EmbedderError, ModelMismatch, discover, index_vault, title_of, ) -from puremind_mac.store import Store +from puremind_mac.store import Store, StoreError def test_discover_skips_hidden_dirs_and_non_markdown(vault): @@ -51,6 +54,115 @@ def test_second_index_is_a_no_op_and_embeds_nothing(vault, store, embedder): assert embedder.texts_embedded == embedded +def _hold_index_run(vault, db, ready): + from fakes import FakeEmbedder + + def hold(texts): + ready.send("embedding") + assert ready.recv() == "finish" + + with Store(db) as writer: + index_vault(vault, writer, _Wrapped(FakeEmbedder(), hold, name="model-b"), rebuild=True) + + +@pytest.mark.parametrize("terminate", [False, True]) +def test_index_runs_exclude_other_processes(vault, store, embedder, terminate): + ctx = multiprocessing.get_context("spawn") + ready, child = ctx.Pipe() + process = ctx.Process(target=_hold_index_run, args=(vault, store.path, child)) + process.start() + child.close() + try: + assert ready.poll(20), "the writer never reached embedding" + assert ready.recv() == "embedding" + with pytest.raises(StoreError, match="another index run holds .*try again"): + index_vault(vault, store, embedder, rebuild=True) + assert embedder.doc_calls == 0 + if terminate: + process.terminate() + else: + ready.send("finish") + process.join(20) + assert not process.is_alive() + if not terminate: + assert process.exitcode == 0 + assert store.get_meta("embedder") == "model-b" and store.stats()["files"] == 3 + with pytest.raises(ModelMismatch): + index_vault(vault, store, embedder) + assert index_vault(vault, store, embedder, rebuild=True).added == 3 + finally: + if process.is_alive(): + process.terminate() + process.join(20) + ready.close() + process.close() + + +def test_index_lock_is_released_after_an_exception(vault, store, embedder): + with pytest.raises(EmbedderError): + index_vault(vault, store, _Wrapped(embedder, _boom)) + assert index_vault(vault, store, embedder).added == 3 + + +def _lock_file(store): + return Path(str(Path(store.path).resolve()) + ".lock") + + +def test_a_waiting_run_starts_once_the_other_run_ends(vault, store, embedder): + other = os.open(_lock_file(store), os.O_RDWR | os.O_CREAT, 0o600) # another run's lock (flock is per open file) + timers = [] + + class ReleaseSoon(logging.Handler): + def emit(self, record): + if "waiting for it to finish" in record.getMessage(): + timers.append(threading.Timer(0.3, fcntl.flock, (other, fcntl.LOCK_UN))) + timers[-1].start() + + logger, handler = logging.getLogger("puremind.index"), ReleaseSoon() + logger.addHandler(handler) + try: + fcntl.flock(other, fcntl.LOCK_EX) + with pytest.raises(StoreError, match="another index run holds"): + index_vault(vault, store, embedder) + assert not timers and embedder.doc_calls == 0 + start = time.monotonic() + assert index_vault(vault, store, embedder, wait=True).added == 3 + assert len(timers) == 1 and time.monotonic() - start >= 0.25 + finally: + logger.removeHandler(handler) + for timer in timers: + timer.join(5) + os.close(other) + + +@pytest.mark.skipif(not hasattr(os, "geteuid") or os.geteuid() == 0, reason="root ignores file permissions") +def test_a_read_only_index_is_reported_as_such_and_gets_no_lock_file(vault, embedder, tmp_path): + home = tmp_path / "frozen" + with Store(home / "index.sqlite") as writer: + index_vault(vault, writer, embedder) + lock = _lock_file(writer) + lock.unlink() + (home / "index.sqlite").chmod(0o400) + home.chmod(0o500) + try: + with Store(home / "index.sqlite") as frozen: + assert frozen.read_only + with pytest.raises(StoreError, match="it is read-only"): + index_vault(vault, frozen, embedder) + assert not lock.exists() + finally: + home.chmod(0o700) + (home / "index.sqlite").chmod(0o600) + + +def test_the_index_lock_is_never_opened_through_a_symlink(vault, store, embedder, tmp_path): + planted = tmp_path / "planted" + _lock_file(store).symlink_to(planted) + with pytest.raises(StoreError, match="cannot open the index lock"): + index_vault(vault, store, embedder) + assert not planted.exists() and store.stats()["files"] == 0 + + def test_touching_without_changing_content_refreshes_recent_without_reembedding(vault, store, embedder, service): index_vault(vault, store, embedder) before = embedder.texts_embedded @@ -691,10 +803,44 @@ def test_rebuild_repoints_an_index_at_a_moved_vault(vault, store, embedder, tmp_ vault.rename(moved) with pytest.raises(ValueError, match="--rebuild"): index_vault(moved, store, embedder) - assert index_vault(moved, store, embedder, rebuild=True).updated == 3 + assert index_vault(moved, store, embedder, rebuild=True).added == 3 assert store.get_meta("vault_root") == str(moved.resolve()) +def test_a_failed_vault_switch_leaves_no_old_content(vault, store, embedder, tmp_path): + index_vault(vault, store, embedder) + other = tmp_path / "other" + other.mkdir() + (other / "new.md").write_text("# New vault\n\nnew content\n") + + def fail(texts): + if texts == [PROBE_TEXT]: # the model runs; it fails later, on the new vault's notes + return + assert store.get_meta("vault_root") == str(other.resolve()) + assert store.stats()["files"] == 0 and not store.fts_hits("tomatoes") + _boom(texts) + + with pytest.raises(EmbedderError): + index_vault(other, store, _Wrapped(embedder, fail), rebuild=True) + assert store.get_meta("vault_root") == str(other.resolve()) + assert store.stats() == {"files": 0, "chunks": 0, "vectors": 0} + assert index_vault(other, store, embedder).added == 1 + assert set(store.file_hashes()) == {"new.md"} + + +def test_a_model_that_cannot_run_does_not_cost_a_moved_vault_its_index(vault, store, embedder, tmp_path): + index_vault(vault, store, embedder) + before, root = store.stats(), store.get_meta("vault_root") + moved = tmp_path / "moved" + vault.rename(moved) + with pytest.raises(EmbedderError, match="could not load the model"): + index_vault(moved, store, _Wrapped(embedder, _boom), rebuild=True) + assert store.stats() == before and store.fts_hits("tomatoes") + assert store.get_meta("vault_root") == root # still the old root, so a run without --rebuild is still refused + with pytest.raises(ValueError, match="belongs to"): + index_vault(moved, store, embedder) + + def test_no_probe_is_embedded_when_nothing_is_rebuilt(vault, store, embedder): index_vault(vault, store, embedder) calls = embedder.doc_calls @@ -850,7 +996,7 @@ def test_a_run_purges_deleted_and_replaced_text_from_the_files(vault, private_in stats = index_vault(vault, private_index, embedder) assert (stats.removed, stats.updated) == (1, 1) assert _copies(home, b"zqxcanary7f3a9") == 0 and _copies(home, b"wqvcanary2b8c") == 0 - assert (home / "index.sqlite-wal").stat().st_size == 0 + assert not private_index.purge_pending() # clearing the marker may itself leave metadata in the WAL @pytest.mark.parametrize("how", ["rebuild", "new model"]) @@ -881,7 +1027,35 @@ def test_a_reader_holding_the_log_gets_a_warning_instead_of_a_stall(vault, priva reader.close() assert stats.removed == 1 and took < 10 # the checkpoint does not wait out the 30 s lock timeout assert any("until the next run" in r.getMessage() for r in caplog.records) - assert private_index.checkpoint() # with the reader gone, the log is emptied + assert private_index.purge_pending() + assert index_vault(vault, private_index, embedder).unchanged == 2 + assert not private_index.purge_pending() + + +@pytest.mark.parametrize("operation", ["replace_file", "delete_file"]) +def test_an_interrupted_run_leaves_cleanup_debt(vault, private_index, embedder, monkeypatch, operation): + note = vault / "secret.md" + note.write_text("# Secret\n\nzqxcanary7f3a9\n") + index_vault(vault, private_index, embedder) + if operation == "replace_file": + note.write_text("# Public\n\nnew content\n") + else: + note.unlink() + original = getattr(private_index, operation) + + def interrupt(*args, **kwargs): + original(*args, **kwargs) + raise KeyboardInterrupt + + with monkeypatch.context() as patch: + patch.setattr(private_index, operation, interrupt) + with pytest.raises(KeyboardInterrupt): + index_vault(vault, private_index, embedder, purge=False) + assert private_index.purge_pending() + stats = index_vault(vault, private_index, embedder) + assert stats.updated == stats.removed == 0 + assert not private_index.purge_pending() + assert _copies(Path(private_index.path).parent, b"zqxcanary7f3a9") == 0 def test_a_run_can_leave_the_keyword_index_compaction_to_the_next_purging_run(vault, private_index, embedder): diff --git a/tests/test_store.py b/tests/test_store.py index e833a5a..aface13 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -737,3 +737,127 @@ def test_a_deferred_purge_is_recorded_until_a_purge_or_clear(store): store.defer_purge() store.clear() assert not store.purge_pending() + + +@pytest.mark.parametrize("operation", ["replace", "delete", "clear"]) +def test_deleted_text_debt_commits_with_the_change(store, monkeypatch, operation): + put(store) + assert not store.purge_pending() + monkeypatch.setattr(store, "purge", lambda: False) # stop clear() before it pays the debt + if operation == "replace": + put(store, sha="h2", chunks=(("H", "new text"),)) + elif operation == "delete": + store.delete_file("a.md") + else: + store.clear() + with Store(store.path) as reopened: + assert reopened.purge_pending() + assert reopened.file_hashes() == ({"a.md": "h2"} if operation == "replace" else {}) + + +@pytest.mark.parametrize("operation", ["replace", "delete", "clear"]) +def test_failed_deletion_rolls_back_rows_and_debt(store, monkeypatch, operation): + put(store) + store.set_meta("vault_root", "/old") + generation = store.generation() + + def fail(): + raise sqlite3.OperationalError("disk is full") + + monkeypatch.setattr(store, "_bump", fail) + with pytest.raises(sqlite3.OperationalError, match="disk is full"): + if operation == "replace": + put(store, sha="h2") + elif operation == "delete": + store.delete_file("a.md") + else: + store.clear(vault_root="/new") + assert store.file_hashes() == {"a.md": "h1"} + assert not store.purge_pending() and store.generation() == generation + assert store.get_meta("vault_root") == "/old" + + +def test_clear_changes_the_vault_root_and_rows_atomically(store, monkeypatch): + put(store) + store.set_meta("vault_root", "/old") + reader = sqlite3.connect(store.path) + original = store._bump + + def snapshot(): + return reader.execute( + "SELECT (SELECT value FROM meta WHERE key = 'vault_root'), (SELECT count(*) FROM files)" + ).fetchone() + + def observe(): + assert snapshot() == ("/old", 1) + original() + + monkeypatch.setattr(store, "_bump", observe) + try: + store.clear(vault_root="/new") + assert snapshot() == ("/new", 0) + finally: + reader.close() + + +@pytest.mark.parametrize("failure", ["busy", "checkpoint error", "compaction error"]) +def test_purge_keeps_debt_until_compaction_and_checkpoint_succeed(store, monkeypatch, failure): + put(store) + store.delete_file("a.md") + execute = store.db.execute + + def checkpoint(): + assert store.purge_pending() + if failure == "checkpoint error": + raise sqlite3.OperationalError("checkpoint failed") + return False + + def fail_compaction(sql, *args): + if "'optimize'" in sql: + raise sqlite3.OperationalError("compaction failed") + return execute(sql, *args) + + with monkeypatch.context() as patch: + if failure == "compaction error": + patch.setattr(store.db, "execute", fail_compaction) + else: + patch.setattr(store, "checkpoint", checkpoint) + if failure == "busy": + assert not store.purge() + else: + with pytest.raises(sqlite3.OperationalError, match="failed"): + store.purge() + with Store(store.path) as reopened: + assert reopened.purge_pending() + assert reopened.purge() + assert not reopened.purge_pending() + + +def test_clear_keeps_debt_when_its_checkpoint_is_busy(store, monkeypatch): + put(store) + with monkeypatch.context() as patch: + patch.setattr(store, "checkpoint", lambda: False) + store.clear() + assert not store.file_hashes() and store.purge_pending() + assert store.purge() and not store.purge_pending() + + +@pytest.mark.parametrize("when", ["before checkpoint", "after checkpoint"]) +def test_purge_does_not_clear_a_concurrent_writers_debt(store, monkeypatch, when): + put(store) + store.defer_purge() + checkpoint = store.checkpoint + with Store(store.path) as writer: + def write_during_purge(): + if when == "before checkpoint": + writer.delete_file("a.md") + assert checkpoint() + if when == "after checkpoint": + writer.delete_file("a.md") + return True + + with monkeypatch.context() as patch: + patch.setattr(store, "checkpoint", write_during_purge) + assert store.purge() + assert store.purge_pending() + assert store.purge() and not store.purge_pending() diff --git a/tests/test_watcher.py b/tests/test_watcher.py index 6398470..346b821 100644 --- a/tests/test_watcher.py +++ b/tests/test_watcher.py @@ -585,6 +585,42 @@ def fake_watch(vault_, run_index, delay, stop, quiet, quiet_after): assert seen == [True, False] # the second save is past the limit, so its run compacts +def test_puremind_watch_waits_for_an_index_run_that_holds_the_lock(vault, tmp_path, monkeypatch): + import fcntl + + from puremind_mac import cli + from puremind_mac.store import Store + + home = tmp_path / "home" + monkeypatch.setenv("PUREMIND_HOME", str(home)) + monkeypatch.delenv("PUREMIND_DB", raising=False) + _stub_models(monkeypatch) + assert cli.main(["index", str(vault)]) == 0 + db, note, took = home / "index.sqlite", vault / "projects" / "garden.md", [] + other = os.open(str(db.resolve()) + ".lock", os.O_RDWR) # a `puremind index` still running + fcntl.flock(other, fcntl.LOCK_EX) + release = threading.Timer(0.5, fcntl.flock, (other, fcntl.LOCK_UN)) + + def fake_watch(vault_, run_index, delay, stop, **kw): + note.write_text("# Garden plan\n\nPlant zucchini in June.\n") # saved while the other run is busy + start = time.monotonic() + release.start() + run_index() # waits, where a refused first run would end the watch + took.append(time.monotonic() - start) + assert not stop.is_set() + + monkeypatch.setattr("puremind_mac.watcher.watch", fake_watch) + try: + assert cli.main(["watch", str(vault)]) == 0 + finally: + if release.ident is not None: + release.join(5) + os.close(other) + assert took and took[0] >= 0.45 + with Store(db) as store: + assert [c.path for c, _ in store.fts_hits("zucchini")] == ["projects/garden.md"] + + def test_quiet_work_runs_once_after_each_quiet_spell(tmp_path): vault = tmp_path / "vault" vault.mkdir()