feat: let the operator pause and resume a root's indexing (MNP 6.2)

A whole-root walk now checks its root before every file: removed or
ejected, it stops instead of reading the drive to the end; paused, it
finishes from the hash cache alone. A paused root is not watched, not
swept by reconcile and survives a restart (roster), its indexed files
stay served. Resume runs the reconcile sweep over the root, so entries
and their enrichment are kept.

Signed root_pause / root_resume over MNP, loopback routes, CLI
`root pause|resume`, a pause button on the indexing dock and
pause/resume in the roots table of Group settings.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 313 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_root_pause.py b/packages/meshbay-node/tests/test_root_pause.py
new file mode 100644
index 0000000..d9a4fcb
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_pause.py
@@ -0,0 +1,313 @@
+"""
+Pausing a root's indexing, and the walk that has to notice it.
+
+A scan of a large drive runs for hours, and nothing could stop one: the hashing
+loop never looked at the root again once it had the file list. Removing the
+directory let it read the whole drive anyway and threw the result away at the
+end; ejecting it went on reading a drive the operator was about to unplug.
+
+A pause has to outlast the walk it stops. Reconcile indexes every file the index
+lacks, so a walk that was merely cancelled was redone at the next tick; the flag
+is what reconcile, the watcher and a restart all read. And it keeps what was
+indexed: the index is rebuilt at every start, so a paused root is listed from the
+hash cache rather than left empty.
+"""
+
+import asyncio
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_node import ops
+from meshbay_node.config import GroupConfig, RootSpec
+from meshbay_node.indexer import indexer as indexer_mod
+from meshbay_node.indexer.cache import IndexCache
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import RootSet
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+GID = "g" * 32
+
+
+def _films(tmp_path: Path, n: int = 6) -> Path:
+    films = tmp_path / "Films"
+    films.mkdir()
+    for i in range(n):
+        (films / f"f{i}.mkv").write_bytes(f"film {i}".encode())
+    return films
+
+
+def _indexer(roots: RootSet, **kw) -> DirectoryIndexer:
+    return DirectoryIndexer(roots=roots, group_id=GID,
+                            sk_node=Ed25519PrivateKey.generate(), gek=None, **kw)
+
+
+def _names(idx: DirectoryIndexer) -> set[str]:
+    return {e.name for e in idx.index.entries}
+
+
+def _count_reads(monkeypatch, after=None, at: int = 0) -> list[str]:
+    """Record every file actually read; run `after` once `at` have been."""
+    read: list[str] = []
+    real = indexer_mod._scan_file
+
+    def counting(root, path):
+        read.append(path.name)
+        if after is not None and len(read) == at:
+            after()
+        return real(root, path)
+
+    monkeypatch.setattr(indexer_mod, "_scan_file", counting)
+    return read
+
+
+# ── The walk notices ─────────────────────────────────────────────────────────
+
+async def test_a_pause_stops_the_walk_at_the_next_file(tmp_path, monkeypatch):
+    roots = RootSet.build([{"path": str(_films(tmp_path))}])
+    idx = _indexer(roots)
+    read = _count_reads(monkeypatch, lambda: setattr(roots.roots[0], "paused", True), at=2)
+
+    await idx.initial_scan()
+
+    assert len(read) == 2, f"read {len(read)} files after the pause"
+    assert len(_names(idx)) == 2, "the files hashed before the pause stay listed"
+    assert idx.progress.scanning is False
+
+
+async def test_a_root_removed_mid_walk_is_not_read_to_the_end(tmp_path, monkeypatch):
+    films = _films(tmp_path)
+    other = tmp_path / "Other"
+    other.mkdir()
+    roots = RootSet.build([{"path": str(films)}])
+    idx = _indexer(roots)
+
+    def remove():
+        idx.roots = RootSet.build([{"path": str(other)}])
+
+    read = _count_reads(monkeypatch, remove, at=1)
+    await idx.initial_scan()
+
+    assert len(read) == 1, f"went on reading a removed root: {read}"
+
+
+async def test_a_root_ejected_mid_walk_is_not_read_to_the_end(tmp_path, monkeypatch):
+    roots = RootSet.build([{"path": str(_films(tmp_path)), "removable": True}])
+    idx = _indexer(roots)
+    read = _count_reads(monkeypatch, lambda: idx.eject_root("Films"), at=1)
+
+    await idx.initial_scan()
+
+    assert len(read) == 1, f"went on reading an ejected drive: {read}"
+
+
+async def test_a_retarget_does_not_hide_the_pause_from_a_running_walk(
+        tmp_path, monkeypatch):
+    """The walk holds the Root it started with; a reload replaces every one."""
+    films = _films(tmp_path)
+    roots = RootSet.build([{"path": str(films)}])
+    idx = _indexer(roots)
+
+    def reload_then_pause():
+        idx.roots = RootSet.build([{"path": str(films)}])
+        idx.roots.roots[0].paused = True
+
+    read = _count_reads(monkeypatch, reload_then_pause, at=1)
+    await idx.initial_scan()
+
+    assert len(read) == 1
+
+
+# ── The pause outlasts the walk ──────────────────────────────────────────────
+
+async def test_reconcile_leaves_a_paused_root_alone(tmp_path):
+    films = _films(tmp_path, n=1)
+    roots = RootSet.build([{"path": str(films)}])
+    idx = _indexer(roots)
+    await idx.initial_scan()
+    await idx.pause_root("Films")
+
+    (films / "new.mkv").write_bytes(b"new film")
+    await idx.reconcile()
+
+    assert "new.mkv" not in _names(idx), "reconcile redid the work the pause stopped"
+
+
+async def test_the_watcher_leaves_a_paused_root_alone(tmp_path):
+    films = _films(tmp_path, n=1)
+    idx = _indexer(RootSet.build([{"path": str(films)}]))
+    await idx.initial_scan()
+    await idx.pause_root("Films")
+
+    new = films / "new.mkv"
+    new.write_bytes(b"new film")
+    await idx._update_entry(new, deleted=False)
+
+    assert "new.mkv" not in _names(idx)
+
+
+async def test_a_paused_root_is_listed_from_the_cache_and_nothing_is_read(
+        tmp_path, monkeypatch):
+    """
+    After a restart the index is empty until the walk refills it. Skipping a
+    paused root would empty it for members; reading it would undo the pause.
+    """
+    films = _films(tmp_path, n=3)
+    cache = IndexCache(db_path=tmp_path / "cache.db")
+    await cache.open()
+    try:
+        await _indexer(RootSet.build([{"path": str(films)}]), cache=cache).initial_scan()
+        (films / "unhashed.mkv").write_bytes(b"never hashed")
+
+        roots = RootSet.build([{"path": str(films), "paused": True}])
+        idx = _indexer(roots, cache=cache)
+        read = _count_reads(monkeypatch)
+        await idx.initial_scan()
+
+        assert read == []
+        assert _names(idx) == {"f0.mkv", "f1.mkv", "f2.mkv"}
+        assert idx.progress.scanning is False
+    finally:
+        await cache.close()
+
+
+async def test_a_pause_mid_rescan_keeps_the_entries_it_had(tmp_path, monkeypatch):
+    """A rescan drops the root's entries before it walks; stopping it there
+    would leave the operator with half a library."""
+    films = _films(tmp_path, n=4)
+    cache = IndexCache(db_path=tmp_path / "cache.db")
+    await cache.open()
+    try:
+        roots = RootSet.build([{"path": str(films)}])
+        idx = _indexer(roots, cache=cache)
+        await idx.initial_scan()
+        for f in films.iterdir():
+            f.write_bytes(f.read_bytes() + b" edited")  # every cache row stale
+
+        read = _count_reads(monkeypatch, lambda: setattr(roots.roots[0], "paused", True),
+                            at=1)
+        await idx._rescan_root(roots.roots[0])
+
+        assert len(read) == 1
+        assert len(_names(idx)) == 1
+    finally:
+        await cache.close()
+
+
+# ── Resume ───────────────────────────────────────────────────────────────────
+
+async def test_resume_indexes_what_is_missing_and_keeps_what_was_enriched(tmp_path):
+    films = _films(tmp_path, n=2)
+    idx = _indexer(RootSet.build([{"path": str(films)}]))
+    await idx.initial_scan()
+    for e in idx.index.entries:
+        e.artist = "Some Artist"
+    await idx.pause_root("Films")
+    (films / "new.mkv").write_bytes(b"new film")
+
+    await idx.resume_root("Films")
+    await asyncio.gather(*idx._scan_tasks)
+
+    assert "new.mkv" in _names(idx)
+    kept = [e for e in idx.index.entries if e.name != "new.mkv"]
+    assert all(e.artist == "Some Artist" for e in kept), \
+        "resume replaced entries it should have left alone"
+
+
+async def test_resume_returns_before_the_walk_ends(tmp_path):
+    films = _films(tmp_path, n=1)
+    roots = RootSet.build([{"path": str(films)}])
+    idx = _indexer(roots)
+    await idx.initial_scan()
+    await idx.pause_root("Films")
+
+    async with idx._scan_lock:
+        await asyncio.wait_for(idx.resume_root("Films"), timeout=1)
+        assert idx.progress.queued == ["Films"]
+    await asyncio.gather(*idx._scan_tasks)
+    assert idx.progress.queued == []
+
+
+# ── The flag is the operator's, and survives ─────────────────────────────────
+
+def _state(tmp_path: Path, roots: RootSet, roster: Roster) -> dict:
+    index = GroupIndex(group_id=GID, sk_node=Ed25519PrivateKey.generate())
+    return {
+        "groups_ctx": {GID: {"index": index, "roots": roots, "gek": None}},
+        "config": SimpleNamespace(groups=[SimpleNamespace(id=GID, roots=[])]),
+        "roster": roster,
+        "node_user_id": "operator",
+    }
+
+
+async def test_pause_and_resume_persist_through_the_roster(tmp_path):
+    films = _films(tmp_path, n=1)
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        state = _state(tmp_path, RootSet.build([{"path": str(films)}]), roster)
+
+        out = await ops.pause_root(state, GID, "films")
+        assert out["status"] == "paused"
+        assert out["roots"][0]["paused"] is True
+        assert out["roots"][0]["available"] is True, "a paused root is still served"
+        assert await roster.paused_roots(GID) == {"films"}
+        assert (await ops.pause_root(state, GID, "Films"))["status"] == "already_paused"
+
+        out = await ops.resume_root(state, GID, "Films")
+        assert out["status"] == "resumed"
+        assert await roster.paused_roots(GID) == set()
+    finally:
+        await roster.close()
+
+
+async def test_removing_a_root_forgets_its_pause(tmp_path):
+    """A directory added back later under the same name starts indexed."""
+    films = _films(tmp_path, n=1)
+    keep = tmp_path / "Keep"
+    keep.mkdir()
+    conf = tmp_path / "node.toml"
+    conf.write_text(
+        f'[[groups]]\nid = "{GID}"\nname = "g"\n\n'
+        f'[[groups.roots]]\npath = "{films.as_posix()}"\n\n'
+        f'[[groups.roots]]\npath = "{keep.as_posix()}"\n')
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        state = _state(tmp_path, RootSet.build(
+            [{"path": str(films)}, {"path": str(keep)}]), roster)
+        state["config_path"] = str(conf)
+        state["config"] = SimpleNamespace(groups=[GroupConfig(
+            id=GID, roots=[RootSpec(path=str(films)), RootSpec(path=str(keep))])])
+        await ops.pause_root(state, GID, "Films")
+
+        await ops.remove_root(state, GID, "Films")
+
+        assert await roster.paused_roots(GID) == set()
+    finally:
+        await roster.close()
+
+
+async def test_a_new_group_can_be_paused_during_its_first_scan(tmp_path):
+    """A group added while the node runs has an indexer, and no context until
+    that first scan ends — the scan most worth pausing."""
+    films = _films(tmp_path, n=1)
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        roots = RootSet.build([{"path": str(films)}])
+        idx = _indexer(roots)
+        state = _state(tmp_path, roots, roster)
+        state["groups_ctx"] = {}
+        state["indexers"] = {GID: idx}
+
+        out = await ops.pause_root(state, GID, "Films")
+
+        assert out["status"] == "paused"
+        assert roots.roots[0].paused is True
+    finally:
+        await roster.close()