From 92e6b9823119b5461efc304a81e79e186a928e6d Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 7 Oct 2026 21:31:10 +0200 Subject: 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 --- packages/meshbay-node/src/meshbay_node/daemon.py | 25 ++++++++++++++++-------- 1 file changed, 17 insertions(+), 8 deletions(-) (limited to 'packages/meshbay-node/src/meshbay_node/daemon.py') diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index d6293ee..e1aac41 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -67,7 +67,11 @@ from meshbay_node.transport import ( WEBRTC_AVAILABLE, Denylist, ) -from meshbay_node.transport.wire import index_delta_message, index_sync_message +from meshbay_node.transport.wire import ( + index_delta_message, + index_sync_message, + list_dirs_off_loop, +) if QUIC_AVAILABLE: from meshbay_node.transport import QuicChunkServer @@ -1452,9 +1456,11 @@ class NodeDaemon(EnrichmentMixin): # "nobody is listening", not a case to send in clear for. if peers and idx.gek: hidden = self._blocklist if self._is_public(group_id) else () + dirs = (None if delta is not None + else await list_dirs_off_loop(indexer.roots)) msg = (index_delta_message(idx, delta, indexer.roots, hidden) if delta is not None - else index_sync_message(idx, indexer.roots, hidden)) + else index_sync_message(idx, indexer.roots, hidden, dirs)) pushed = 0 for session in peers: try: @@ -1488,16 +1494,18 @@ class NodeDaemon(EnrichmentMixin): return if self._blocklist.replace(hashes): log.info("Content blocklist: %d hashes", len(self._blocklist)) - self._push_public_indexes() + await self._push_public_indexes() - def _on_blocklist_update(self, add: list, remove: list) -> None: + def _on_blocklist_update(self, add: list, remove: list): + """Returns the push it started, if any, for whoever wants to wait on it.""" if not self._hosts_public_group(): - return + return None if self._blocklist.apply(add, remove): log.info("Content blocklist updated: +%d -%d", len(add), len(remove)) - self._push_public_indexes() + return spawn(self._push_public_indexes()) + return None - def _push_public_indexes(self) -> None: + async def _push_public_indexes(self) -> None: """Resend each public group's whole index, so what the list now hides leaves every connected member's view, and what it released comes back.""" if not self._webrtc: @@ -1506,7 +1514,8 @@ class NodeDaemon(EnrichmentMixin): idx = indexer.index if not self._is_public(idx.group_id) or not idx.gek: continue - msg = index_sync_message(idx, indexer.roots, self._blocklist) + dirs = await list_dirs_off_loop(indexer.roots) + msg = index_sync_message(idx, indexer.roots, self._blocklist, dirs) for session in list(self._webrtc._sessions.values()): if session._group_id == idx.group_id: try: -- cgit v1.2.3