summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src/meshbay_node/transport
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-08-23 21:55:20 +0200
committerChristophe Besson <cbesson@gmail.com>2026-08-23 21:55:20 +0200
commitb3709ac4d362987a9d025616c95065ceed0d216b (patch)
tree32e0cc5cc2775eddf516d114fa9799347a214bda /packages/meshbay-node/src/meshbay_node/transport
parent012ba5b0cb8c556ce773423ca38d5184b74659ac (diff)
downloadmeshbay-b3709ac4d362987a9d025616c95065ceed0d216b.tar.gz
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
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py108
1 files changed, 108 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index b6f572a..fa6c3e9 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -63,6 +63,7 @@ from meshbay_common.adminop import (
OP_MEMBER_UNPIN,
OP_MEMBER_UPLOAD,
OP_APPS_ENABLED,
+ OP_SET_SCAN_SETTINGS,
OP_ROOT_ADD,
OP_ROOT_REMOVE,
OP_GROUP_ATTACH,
@@ -85,6 +86,7 @@ from meshbay_common.join import (
from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
from meshbay_common.protocol import MNP
from meshbay_node.indexer import GroupIndex
+from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node import ops
from meshbay_node.roots import (
RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name,
@@ -409,6 +411,8 @@ class WebRTCPeerSession:
self._do_member_upload(msg)
elif mtype == MNP.APPS_ENABLED:
self._do_apps_enabled(msg)
+ elif mtype == MNP.SET_SCAN_SETTINGS:
+ self._do_set_scan_settings(msg)
elif mtype == MNP.MEMBER_UNPIN:
self._do_member_unpin(msg)
elif mtype == MNP.GEK_ROTATE:
@@ -659,6 +663,20 @@ class WebRTCPeerSession:
# setting (or one whose context has not loaded it yet) hides
# nothing.
"enabled_apps": list(self._group_ctx().get("enabled_apps") or []),
+ # So a client that connects mid-scan shows the indexing state
+ # immediately, instead of waiting for the next periodic
+ # INDEX_PROGRESS push. Never a path or filename — see
+ # IndexProgress in indexer.py.
+ "indexing": self._indexing_status(),
+ # Current values only — not enforced from here, just shown to
+ # the operator in Settings so the number on screen matches what
+ # the indexer is actually doing (set_scan_settings, ops.py).
+ "scan_settings": {
+ "reconcile_interval_secs": self._group_ctx().get(
+ "reconcile_interval_secs", DirectoryIndexer.DEFAULT_RECONCILE_SECS),
+ "debounce_secs": self._group_ctx().get(
+ "debounce_secs", DirectoryIndexer.DEFAULT_DEBOUNCE_SECS),
+ },
}
if node_user_id:
ack["node_user_id"] = node_user_id
@@ -668,6 +686,13 @@ class WebRTCPeerSession:
self._send(ack)
self._audit("handshake")
+ # Someone is here now — reconcile's backstop should be prompt again
+ # rather than however far its backoff had stretched while nobody
+ # was connected (indexer.py DirectoryIndexer.note_activity).
+ note_activity = self._group_ctx().get("note_activity")
+ if note_activity:
+ note_activity()
+
async def _do_gek_bundle_fetch(self) -> None:
"""Serve the caller's wrapped GEK bundle during the handshake window."""
bundle_store = self._ctx.get("bundle_store")
@@ -1648,6 +1673,69 @@ class WebRTCPeerSession:
except Exception:
pass
+ # Reconcile's backstop and the watchdog debounce (indexer.py
+ # DirectoryIndexer) — how hard the node works on the operator's own
+ # disk, not a member-facing permission. Signed for the same reason as
+ # apps_enabled: consistency of the authorization model, not because a
+ # wrong value here is itself dangerous.
+ MIN_RECONCILE_SECS = 10.0
+ MAX_RECONCILE_SECS = 24 * 3600.0
+ MIN_DEBOUNCE_SECS = 0.0
+ MAX_DEBOUNCE_SECS = 300.0
+
+ def _do_set_scan_settings(self, msg: dict) -> None:
+ try:
+ reconcile = float(msg.get("reconcile_interval_secs"))
+ debounce = float(msg.get("debounce_secs"))
+ except (TypeError, ValueError):
+ self._send({"type": "error", "detail": "Invalid scan settings"})
+ return
+ if not (self.MIN_RECONCILE_SECS <= reconcile <= self.MAX_RECONCILE_SECS):
+ self._send({"type": "error",
+ "detail": f"reconcile_interval_secs must be between "
+ f"{self.MIN_RECONCILE_SECS:.0f} and "
+ f"{self.MAX_RECONCILE_SECS:.0f}"})
+ return
+ if not (self.MIN_DEBOUNCE_SECS <= debounce <= self.MAX_DEBOUNCE_SECS):
+ self._send({"type": "error",
+ "detail": f"debounce_secs must be between "
+ f"{self.MIN_DEBOUNCE_SECS:.0f} and "
+ f"{self.MAX_DEBOUNCE_SECS:.0f}"})
+ return
+ if not self._has_admin_authority():
+ self._send({"type": "error", "detail": "No authorized key for this"})
+ return
+ self._issue_admin_challenge(
+ OP_SET_SCAN_SETTINGS, f"{reconcile:g},{debounce:g}")
+
+ async def _admin_exec_set_scan_settings(
+ self, pending: dict, transcript: bytes, sig: bytes,
+ ) -> None:
+ try:
+ reconcile_s, debounce_s = pending["subject"].split(",")
+ reconcile, debounce = float(reconcile_s), float(debounce_s)
+ except (ValueError, KeyError):
+ self._send({"type": "error", "detail": "Invalid scan settings"})
+ return
+ if not await self._verify_admin_sig(transcript, sig):
+ self._send({"type": "error", "detail": "Signature verification failed"})
+ self._audit("admin_auth_failed", f"set_scan_settings:{pending['subject']}")
+ return
+ try:
+ result = await self._run_op(
+ ops.set_scan_settings, self._group_id or "", reconcile, debounce)
+ except ops.OpError as e:
+ self._send({"type": "error", "detail": e.message})
+ return
+ self._audit("set_scan_settings", pending["subject"])
+
+ notice = {"type": MNP.SET_SCAN_SETTINGS_ACK, "v": MNP_VERSION, **result}
+ for uid, session in list(self._peer_registry().items()):
+ try:
+ session._send(notice)
+ except Exception:
+ pass
+
# ── Node management (D5) ─────────────────────────────────────────────────
async def _do_node_status(self, msg: dict) -> None:
@@ -2012,6 +2100,23 @@ class WebRTCPeerSession:
return self._ctx["groups"][self._group_id]
return self._ctx
+ def _indexing_status(self) -> dict:
+ """
+ {"scanning": bool, "scanned_bytes": int, "total_bytes": int} for the
+ handshake ack and INDEX_PROGRESS pushes — never a path or filename,
+ that stays local to the operator's own admin UI. Absent "progress"
+ (context not loaded, or a group with no indexer at all) reads as
+ idle rather than erroring.
+ """
+ progress = self._group_ctx().get("progress")
+ if progress is None:
+ return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0}
+ return {
+ "scanning": progress.scanning,
+ "scanned_bytes": progress.scanned_bytes,
+ "total_bytes": progress.total_bytes,
+ }
+
def _peer_registry(self) -> dict:
"""
Connected peers for THIS group only.
@@ -2626,6 +2731,9 @@ class WebRTCPeerSession:
elif pending["op"] == OP_APPS_ENABLED:
self._spawn(
self._admin_exec_apps_enabled(pending, transcript, sig_bytes))
+ elif pending["op"] == OP_SET_SCAN_SETTINGS:
+ self._spawn(
+ self._admin_exec_set_scan_settings(pending, transcript, sig_bytes))
elif pending["op"] == OP_ROOT_ADD:
self._spawn(
self._admin_exec_root_add(pending, transcript, sig_bytes))