perf(node): list a group's directories off the event loop

Every full index walked all roots on the loop, and a node with several
large roots stopped answering for seconds. Walk directories only, on the
roots' disk thread; index_sync is spawned and still answers on failure.

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

3 files changed, 89 insertions, 93 deletions

diff --git a/packages/meshbay-node/tests/golden/dispatch.json b/packages/meshbay-node/tests/golden/dispatch.json
index 61bbc45..44d030a 100644
--- a/packages/meshbay-node/tests/golden/dispatch.json
+++ b/packages/meshbay-node/tests/golden/dispatch.json
@@ -9167,115 +9167,67 @@
   },
   "index_sync | member | bare": {
    "audit": [],
-   "log": [
-    "ERROR Error handling %s on DataChannel: %s"
-   ],
-   "sent": [
-    {
-     "detail": "Request failed",
-     "req_id": 4242,
-     "type": "error"
-    }
-   ],
-   "spawned": []
+   "log": [],
+   "sent": [],
+   "spawned": [
+    "_do_index_sync"
+   ]
   },
   "index_sync | member | lists": {
    "audit": [],
-   "log": [
-    "ERROR Error handling %s on DataChannel: %s"
-   ],
-   "sent": [
-    {
-     "detail": "Request failed",
-     "req_id": 4242,
-     "type": "error"
-    }
-   ],
-   "spawned": []
+   "log": [],
+   "sent": [],
+   "spawned": [
+    "_do_index_sync"
+   ]
   },
   "index_sync | member | numbers": {
    "audit": [],
-   "log": [
-    "ERROR Error handling %s on DataChannel: %s"
-   ],
-   "sent": [
-    {
-     "detail": "Request failed",
-     "req_id": 4242,
-     "type": "error"
-    }
-   ],
-   "spawned": []
+   "log": [],
+   "sent": [],
+   "spawned": [
+    "_do_index_sync"
+   ]
   },
   "index_sync | member | strings": {
    "audit": [],
-   "log": [
-    "ERROR Error handling %s on DataChannel: %s"
-   ],
-   "sent": [
-    {
-     "detail": "Request failed",
-     "req_id": 4242,
-     "type": "error"
-    }
-   ],
-   "spawned": []
+   "log": [],
+   "sent": [],
+   "spawned": [
+    "_do_index_sync"
+   ]
   },
   "index_sync | operator | bare": {
    "audit": [],
-   "log": [
-    "ERROR Error handling %s on DataChannel: %s"
-   ],
-   "sent": [
-    {
-     "detail": "Request failed",
-     "req_id": 4242,
-     "type": "error"
-    }
-   ],
-   "spawned": []
+   "log": [],
+   "sent": [],
+   "spawned": [
+    "_do_index_sync"
+   ]
   },
   "index_sync | operator | lists": {
    "audit": [],
-   "log": [
-    "ERROR Error handling %s on DataChannel: %s"
-   ],
-   "sent": [
-    {
-     "detail": "Request failed",
-     "req_id": 4242,
-     "type": "error"
-    }
-   ],
-   "spawned": []
+   "log": [],
+   "sent": [],
+   "spawned": [
+    "_do_index_sync"
+   ]
   },
   "index_sync | operator | numbers": {
    "audit": [],
-   "log": [
-    "ERROR Error handling %s on DataChannel: %s"
-   ],
-   "sent": [
-    {
-     "detail": "Request failed",
-     "req_id": 4242,
-     "type": "error"
-    }
-   ],
-   "spawned": []
+   "log": [],
+   "sent": [],
+   "spawned": [
+    "_do_index_sync"
+   ]
   },
   "index_sync | operator | strings": {
    "audit": [],
-   "log": [
-    "ERROR Error handling %s on DataChannel: %s"
-   ],
-   "sent": [
-    {
-     "detail": "Request failed",
-     "req_id": 4242,
-     "type": "error"
-    }
-   ],
-   "spawned": []
+   "log": [],
+   "sent": [],
+   "spawned": [
+    "_do_index_sync"
+   ]
   },
   "invite_cancel | challenged | bare": {
    "audit": [],
diff --git a/packages/meshbay-node/tests/test_content_blocklist.py b/packages/meshbay-node/tests/test_content_blocklist.py
index 0e07787..8e0b790 100644
--- a/packages/meshbay-node/tests/test_content_blocklist.py
+++ b/packages/meshbay-node/tests/test_content_blocklist.py
@@ -129,7 +129,7 @@ async def test_a_public_group_refuses_a_blocked_file_and_its_thumbnail(tmp_path)
 @pytest.mark.asyncio
 async def test_a_public_group_does_not_list_a_blocked_file(tmp_path):
     s, gek = _session("public", tmp_path)
-    s._do_index_sync()
+    await s._do_index_sync()
     payload = unseal(gek, PURPOSE_INDEX, "index_sync", GROUP, s._channel.sent[-1])
     assert [e["id"] for e in payload["entries"]] == [KEPT]
 
@@ -137,7 +137,7 @@ async def test_a_public_group_does_not_list_a_blocked_file(tmp_path):
 @pytest.mark.asyncio
 async def test_a_private_group_is_untouched(tmp_path):
     s, gek = _session("private", tmp_path)
-    s._do_index_sync()
+    await s._do_index_sync()
     payload = unseal(gek, PURPOSE_INDEX, "index_sync", GROUP, s._channel.sent[-1])
     assert sorted(e["id"] for e in payload["entries"]) == [BLOCKED, KEPT]
     assert not s._refuse_blocked(BLOCKED)
@@ -218,14 +218,15 @@ def _daemon(tmp_path, visibility: str):
     return daemon, session, gek
 
 
-def test_a_pushed_block_resends_the_index_without_the_file(tmp_path):
+@pytest.mark.asyncio
+async def test_a_pushed_block_resends_the_index_without_the_file(tmp_path):
     daemon, session, gek = _daemon(tmp_path, "public")
-    daemon._on_blocklist_update([BLOCKED], [])
+    await daemon._on_blocklist_update([BLOCKED], [])
     msg = session._send.call_args[0][0]
     ids = [e["id"] for e in unseal(gek, PURPOSE_INDEX, "index_sync", GROUP, msg)["entries"]]
     assert ids == [KEPT]
 
-    daemon._on_blocklist_update([], [BLOCKED])
+    await daemon._on_blocklist_update([], [BLOCKED])
     msg = session._send.call_args[0][0]
     ids = sorted(e["id"] for e in unseal(gek, PURPOSE_INDEX, "index_sync", GROUP, msg)["entries"])
     assert ids == [BLOCKED, KEPT]
diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py
index 44864ba..c0ab052 100644
--- a/packages/meshbay-node/tests/test_disk_io_off_loop.py
+++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py
@@ -34,6 +34,7 @@ from meshbay_node.indexer.group_index import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_node.roots import Root, RootSet
 from meshbay_node.transfers import LeaselessReads
+from meshbay_node.transport import wire
 from meshbay_node.transport.webrtc import files, media_tools, upload_handlers
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 from node_source import webrtc_files
@@ -412,3 +413,45 @@ async def test_a_slow_scratch_read_does_not_stop_the_loop(tmp_path, monkeypatch)
     assert blob == CONTENT
     assert ticker.ticks > FREE_TICKS / 2, (
         f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s read")
+
+
+# ── The directory listing of a full index ───────────────────────────────────
+#
+# Every `index_sync` lists the group's directories by walking all of its roots.
+# On the loop, a library of several large roots was seconds in which the node
+# answered nobody: found live, a tester adding a sixth directory lost "Add a
+# directory" because the node missed the desktop client's 3 s check.
+
+async def test_a_slow_directory_walk_does_not_stop_the_loop(tmp_path, monkeypatch):
+    session, _, _, _ = await _served(tmp_path)
+    monkeypatch.setattr(wire, "_walk_dirs", _slow(wire._walk_dirs))
+
+    with _Ticker() as ticker:
+        await session._do_index_sync()
+
+    assert ticker.ticks > FREE_TICKS / 2, (
+        f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s walk")
+    assert session.sent[-1]["type"] == MNP.INDEX_SYNC
+
+
+async def test_an_index_sync_that_fails_still_answers(tmp_path):
+    """It is spawned now, and a spawned task that fails only logs: without its
+    own answer the client would wait out its timeout for nothing."""
+    session, ctx, _, _ = await _served(tmp_path)
+    del ctx["index"]
+
+    await session._do_index_sync()
+
+    assert session.sent[-1] == {"type": "error", "detail": "Request failed"}
+
+
+def test_the_walk_lists_directories_and_skips_hidden_ones(tmp_path):
+    root = tmp_path / "films"
+    (root / "a" / "b").mkdir(parents=True)
+    (root / ".hidden" / "inside").mkdir(parents=True)
+    (root / "a" / "file.bin").write_bytes(b"x")
+    gone = tmp_path / "gone"
+
+    dirs = wire._walk_dirs([("films", root, True), ("gone", gone, False)])
+
+    assert dirs == ["films", "films/a", "films/a/b", "gone"]