feat(node): persistent index cache, visible scan progress, adaptive reconcile, and delta sync

Indexer performance work, in four parts:

- Persistent (path, size, mtime) -> hash cache (indexer/cache.py) so a node
  restart no longer re-hashes every file — measured at 23 minutes for a
  114 GB library on a slow disk before this, near-instant after. Hashing
  is deliberately kept sequential (max_workers=1): it was never actually
  concurrent despite the pool size, and two interleaved reads seek-thrash
  a spinning disk instead of going faster.

- Byte-based scan progress (IndexProgress), surfaced via the loopback
  index-status route, the handshake ack, and a periodic INDEX_PROGRESS
  push to connected peers — drives a progress bar in the Create Group
  wizard and "add a directory" in Settings, and an animated presence dot.
  Guaranteed to settle back to idle via try/finally and a final push on
  the scanning->false transition.

- The reconcile backstop's directory walks now run in the executor
  instead of blocking the daemon's event loop; its interval defaults to
  10 min (was 60s) with adaptive backoff to 2h when nothing changes,
  reset on a real change or a peer connecting, and is now a per-group
  operator setting (signed op + group Settings UI).

- INDEX_DELTA wired up (protocol support existed, nothing called it):
  _on_index_change now sends additions/deletions instead of rebuilding
  the full entries list, coalesced over a short window so a burst of
  file events produces one push, and the hub swarm registration for
  public groups only (re-)registers newly added hashes.

Also fixes several bugs found while testing the above against real
libraries (a 114 GB and a 100+ GB group on a USB HDD):

- /api/reload blocked until the reload — including a brand-new group's
  full initial scan — finished, which the Electron bridge's fixed 30s
  call timeout turned into a hard failure on any real library. The route
  now fires the reload without waiting (ops.start_reload), matching
  add_root/remove_root's existing pattern; the wizard's own step order
  was fixed to wait for the group to actually appear hosted before the
  steps that need it (extra roots, GEK), with retries for the residual
  race between that and the daemon's own bookkeeping.
- transport.js's hand-rolled msgpack codec had no case for uint64/int64
  (0xcf/0xd3) and crashed decoding any message containing one — hit by
  IndexProgress.scanned_bytes/total_bytes for any group over ~4.3 GB.
  Verified against real msgpack-encoded bytes from the Python side.
- chat_hist_resp, and this change's own index_progress and
  set_scan_settings_ack pushes, were not routed by message type and
  could be handed to an unrelated pending request by the transport's
  "oldest pending" fallback, stalling it until its own 30s timeout and
  corrupting whatever received the wrong reply in its place.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016SF6RKNBKg9qejmoMJ9ybA

1 files changed, 329 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py
new file mode 100644
index 0000000..7cb74cb
--- /dev/null
+++ b/packages/meshbay-node/tests/test_hot_reload_survives_client_close.py
@@ -0,0 +1,329 @@
+"""
+Adding a group (Create Group wizard, or "add a directory" to an existing
+one) fires `_reload_config()` without awaiting it (`asyncio.ensure_future`,
+ui/app.py) — the request handler, and whatever browser tab triggered it,
+return immediately. This is deliberate: the initial scan behind it can take
+a very long time (measured at 23 minutes for a 114 GB library on a slow
+disk), and none of that work belongs to the HTTP request or the WebRTC
+session that happened to start it.
+
+This test proves the scan is genuinely independent of its caller: it starts
+the reload the same way the real endpoint does — schedules it and does not
+await it, standing in for "the browser tab that made the call was closed" —
+then does something else, and only afterwards checks that the reload
+finished and the new group became available on its own.
+"""
+
+import asyncio
+import base64
+import os
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from meshbay_node import ops
+from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
+from meshbay_node.daemon import NodeDaemon
+import meshbay_node.indexer.indexer as indexer_mod
+
+
+def _free_port() -> int:
+    import socket
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+def _toml(data_dir, first_group_dir, second_group_id=None, second_group_dir=None) -> str:
+    # data_dir MUST come before any [section] header — TOML has no notion of
+    # "back to top-level" once a table is open, so a bare `key = value` line
+    # placed after [node] becomes node.data_dir, not the top-level data_dir
+    # load_config() actually reads. Silently falls back to the real default
+    # (~/.local/share/meshbay) instead of erroring, which is exactly how this
+    # test once ran a whole daemon — including _shutdown()'s unlink of
+    # ui-token — against the developer's real, already-running node.
+    text = f"""
+data_dir = "{data_dir}"
+
+[hub]
+url = "http://localhost:9999"
+username = "testuser"
+
+[node]
+quic_port = {_free_port()}
+ui_port = {_free_port()}
+
+[[groups]]
+id = "{"a" * 32}"
+name = "first"
+shared_dir = "{first_group_dir}"
+visibility = "private"
+"""
+    if second_group_id:
+        text += f"""
+[[groups]]
+id = "{second_group_id}"
+name = "slow-new-group"
+shared_dir = "{second_group_dir}"
+visibility = "private"
+"""
+    return text
+
+
+def _mock_keystore_keys(sk_ed):
+    sk_x = X25519PrivateKey.generate()
+    pk_x_raw = sk_x.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    mock_keys = MagicMock()
+    mock_keys.sk_ed25519 = sk_ed
+    mock_keys.pk_ed25519_b64 = "test"
+    mock_keys.sk_x25519 = sk_x
+    mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode()
+    return mock_keys
+
+
+@pytest.mark.asyncio
+async def test_hot_loaded_group_finishes_scanning_without_anyone_awaiting_the_reload(
+        tmp_path):
+    first_dir = tmp_path / "first"
+    first_dir.mkdir()
+    (first_dir / "readme.txt").write_bytes(b"hello")
+
+    second_dir = tmp_path / "second"
+    second_dir.mkdir()
+    for i in range(3):
+        (second_dir / f"file{i}.bin").write_bytes(os.urandom(64))
+    second_group_id = "b" * 32
+
+    data_dir = tmp_path / "data"
+    config_path = tmp_path / "node.toml"
+    config_path.write_text(_toml(data_dir, first_dir))
+
+    from meshbay_node.config import load_config
+    daemon = NodeDaemon(load_config(config_path), config_path=config_path)
+
+    sk_node = Ed25519PrivateKey.generate()
+    mock_keys = _mock_keystore_keys(sk_node)
+    mock_session = MagicMock()
+    mock_session.node_id = "node123"
+    mock_session.user_id = "user123"
+    mock_session.hub_pk_pem = sk_node.public_key().public_bytes(
+        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+
+    # Slow the second group's hashing down (a stand-in for a large/slow
+    # library) so there is a real window in which "nobody is awaiting this"
+    # actually means something, without needing a genuinely huge file.
+    real_scan_file = indexer_mod._scan_file
+
+    def slow_scan_file(root, path):
+        import time
+        time.sleep(0.15)
+        return real_scan_file(root, path)
+
+    with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \
+         patch("meshbay_node.daemon.HubClient") as MockHub, \
+         patch.object(indexer_mod, "_scan_file", slow_scan_file):
+
+        hub_instance = AsyncMock()
+        hub_instance.startup = AsyncMock(return_value=mock_session)
+        hub_instance.send_ws = AsyncMock()
+        hub_instance._ws = None
+        hub_instance.close = AsyncMock()
+        hub_instance.__aenter__ = AsyncMock(return_value=hub_instance)
+        hub_instance.__aexit__ = AsyncMock(return_value=False)
+        MockHub.return_value = hub_instance
+
+        shutdown_event = asyncio.Event()
+
+        async def mock_maintain_ws(**kwargs):
+            await shutdown_event.wait()
+        hub_instance.maintain_ws = mock_maintain_ws
+
+        async def run_daemon():
+            with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15):
+                try:
+                    await asyncio.wait_for(daemon.run(), timeout=15)
+                except (asyncio.TimeoutError, Exception):
+                    pass
+
+        run_task = asyncio.create_task(run_daemon())
+        try:
+            for _ in range(50):
+                if daemon._state.get("status") == "running":
+                    break
+                await asyncio.sleep(0.05)
+            assert daemon._state["status"] == "running"
+            assert second_group_id not in daemon._state.get("groups_ctx", {})
+
+            # Add the second group to the config on disk, the way the wizard's
+            # attach + /api/reload would leave it, then fire the reload exactly
+            # as ui/app.py does: scheduled, NOT awaited.
+            config_path.write_text(_toml(data_dir, first_dir,
+                                         second_group_id, second_dir))
+            reload_task = asyncio.ensure_future(daemon._reload_config())
+
+            # Stand in for "the browser tab is gone": do something completely
+            # unrelated to the reload, and explicitly do not await it here.
+            await asyncio.sleep(0.01)
+            assert second_group_id not in daemon._state.get("groups_ctx", {}), \
+                "the scan (3 files x 0.15s) cannot have finished yet"
+
+            # Only now catch up with the background work, from a place that
+            # has no relationship to whatever originally triggered it.
+            await asyncio.wait_for(reload_task, timeout=5)
+
+            assert second_group_id in daemon._state["groups_ctx"], \
+                "the new group must be usable once its scan finishes, " \
+                "regardless of whether anything was still watching the reload"
+            new_indexer = daemon._state["indexers"][second_group_id]
+            assert new_indexer.index.count == 3
+            assert new_indexer.progress.scanning is False
+        finally:
+            shutdown_event.set()
+            await daemon._shutdown()
+            run_task.cancel()
+            try:
+                await run_task
+            except (asyncio.CancelledError, Exception):
+                pass
+
+
+@pytest.mark.asyncio
+async def test_group_scoped_ops_404_until_listed_then_succeed(tmp_path):
+    """
+    The wizard's own sequence, reproduced against the real ops layer: attach
+    a brand-new group, fire the reload the way /api/reload now does
+    (ops.start_reload — scheduled, not awaited), and hit the group-scoped
+    calls that come right after in the UI (add a root, init the GEK) while
+    the scan is still running.
+
+    Found live: "Attaching to node" no longer times out (ops.start_reload
+    returns immediately), but the very next wizard step then failed with
+    "Group not configured on this node" / "Group not hosted on this node" —
+    the group is not in daemon._state["config"].groups or ["groups_ctx"]
+    until _reload_config_inner() finishes, scan included, which is *after*
+    ops.start_reload has already returned. This locks in both halves: the
+    404 while the scan runs, and success once ops.list_groups() actually
+    lists the group — the exact condition the wizard's own wait
+    (platform.waitForGroupHosted, app.js) polls for.
+    """
+    first_dir = tmp_path / "first"
+    first_dir.mkdir()
+    (first_dir / "readme.txt").write_bytes(b"hello")
+
+    second_dir = tmp_path / "second"
+    second_dir.mkdir()
+    for i in range(3):
+        (second_dir / f"file{i}.bin").write_bytes(os.urandom(64))
+    second_group_id = "c" * 32
+    extra_root_dir = tmp_path / "extra_root"
+    extra_root_dir.mkdir()
+
+    data_dir = tmp_path / "data2"
+    config_path = tmp_path / "node2.toml"
+    config_path.write_text(_toml(data_dir, first_dir))
+
+    from meshbay_node.config import load_config
+    daemon = NodeDaemon(load_config(config_path), config_path=config_path)
+
+    sk_node = Ed25519PrivateKey.generate()
+    mock_keys = _mock_keystore_keys(sk_node)
+    mock_session = MagicMock()
+    mock_session.node_id = "node123"
+    mock_session.user_id = "user123"
+    mock_session.hub_pk_pem = sk_node.public_key().public_bytes(
+        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+
+    real_scan_file = indexer_mod._scan_file
+
+    def slow_scan_file(root, path):
+        import time
+        time.sleep(0.15)
+        return real_scan_file(root, path)
+
+    with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \
+         patch("meshbay_node.daemon.HubClient") as MockHub, \
+         patch.object(indexer_mod, "_scan_file", slow_scan_file):
+
+        hub_instance = AsyncMock()
+        hub_instance.startup = AsyncMock(return_value=mock_session)
+        hub_instance.send_ws = AsyncMock()
+        hub_instance._ws = None
+        hub_instance.close = AsyncMock()
+        hub_instance.__aenter__ = AsyncMock(return_value=hub_instance)
+        hub_instance.__aexit__ = AsyncMock(return_value=False)
+        MockHub.return_value = hub_instance
+
+        shutdown_event = asyncio.Event()
+
+        async def mock_maintain_ws(**kwargs):
+            await shutdown_event.wait()
+        hub_instance.maintain_ws = mock_maintain_ws
+
+        async def run_daemon():
+            with patch("signal.SIGINT", 2), patch("signal.SIGTERM", 15):
+                try:
+                    await asyncio.wait_for(daemon.run(), timeout=15)
+                except (asyncio.TimeoutError, Exception):
+                    pass
+
+        run_task = asyncio.create_task(run_daemon())
+        try:
+            for _ in range(50):
+                if daemon._state.get("status") == "running":
+                    break
+                await asyncio.sleep(0.05)
+            assert daemon._state["status"] == "running"
+
+            # The same file write ops.attach_group does (a raw text append),
+            # then the same fire-and-forget reload /api/reload now does.
+            config_path.write_text(_toml(data_dir, first_dir,
+                                         second_group_id, second_dir))
+            reload_task = asyncio.ensure_future(ops.start_reload(daemon._state))
+
+            await asyncio.sleep(0.01)
+            listing = await ops.list_groups(daemon._state)
+            assert second_group_id not in [g["id"] for g in listing["groups"]], \
+                "the scan (3 files x 0.15s) cannot have finished this fast"
+
+            # Exactly the wizard's next two steps, hit mid-scan.
+            with pytest.raises(ops.OpError) as add_root_exc:
+                await ops.add_root(daemon._state, second_group_id,
+                                   str(extra_root_dir))
+            assert add_root_exc.value.status == 404
+
+            with pytest.raises(ops.OpError) as gek_exc:
+                await ops.set_gek(daemon._state, second_group_id)
+            assert gek_exc.value.status == 404
+
+            # Now wait the way platform.waitForGroupHosted (app.js) does:
+            # poll list_groups(), not index-status, until the group is
+            # actually there.
+            for _ in range(100):
+                listing = await ops.list_groups(daemon._state)
+                if second_group_id in [g["id"] for g in listing["groups"]]:
+                    break
+                await asyncio.sleep(0.05)
+            else:
+                pytest.fail("group never appeared in list_groups()")
+
+            await asyncio.wait_for(reload_task, timeout=5)
+
+            # Both calls that 404'd above must now succeed.
+            add_result = await ops.add_root(daemon._state, second_group_id,
+                                            str(extra_root_dir))
+            assert add_result["status"] == "added"
+
+            gek_result = await ops.set_gek(daemon._state, second_group_id)
+            assert gek_result["status"] == "ok"
+        finally:
+            shutdown_event.set()
+            await daemon._shutdown()
+            run_task.cancel()
+            try:
+                await run_task
+            except (asyncio.CancelledError, Exception):
+                pass