diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/daemon.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 237 |
1 files changed, 217 insertions, 20 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index c68f999..af85340 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -45,7 +45,7 @@ from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config from meshbay_node.roots import RootSet, RootError from meshbay_node.hub_client import HubClient, HubConfig -from meshbay_node.indexer import DirectoryIndexer +from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore from meshbay_node.roster import Roster from meshbay_node.transport import ( @@ -119,6 +119,17 @@ class NodeDaemon: self._denylist = ( Denylist(path=config.data_dir / "denylist.json") if Denylist else None) self._chat_stores: dict[str, ChatStore] = {} + self._index_caches: dict[str, IndexCache] = {} + # Coalesces a burst of index changes (one per debounced watchdog + # event) into a single broadcast — see _on_index_change. 0.5s is + # short enough nobody notices the wait, long enough that dropping a + # few hundred files into a watched folder produces one push instead + # of one per file. + self._broadcast_coalesce_secs = 0.5 + self._pending_broadcasts: dict[str, asyncio.TimerHandle] = {} + # group_id -> (version, {id: entry}) as of the last thing actually + # broadcast — the comparison point for the next delta. + self._last_broadcast_snapshot: dict[str, tuple] = {} self._audit_store: AuditStore | None = None self._bundle_store: BundleStore | None = None self._roster: Roster | None = None @@ -234,12 +245,31 @@ class NodeDaemon: log.info("No GEK yet for group %s — will accept first setup", group_cfg.name) + index_cache = IndexCache( + db_path=data_dir / group_cfg.id[:16] / "index_cache.db") + await index_cache.open() + self._index_caches[group_cfg.id] = index_cache + + # Read once at load, like member_upload/enabled_apps below — + # kept current in place afterwards by set_scan_settings + # (ops.py), which updates both this indexer object directly + # and roster.db, so a restart picks up the same values. + scan_settings = ( + await self._roster.scan_settings(group_cfg.id) + if self._roster else { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }) + indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=keys.sk_ed25519, gek=gek, on_change=self._on_index_change, + cache=index_cache, + reconcile_secs=scan_settings["reconcile_interval_secs"], + debounce_secs=scan_settings["debounce_secs"], ) await indexer.start(defer_scan=True) self._indexers.append(indexer) @@ -253,6 +283,20 @@ class NodeDaemon: "gek": gek, "roots": roots, "index": indexer.index, + # Live reference, mutated in place by the indexer itself + # (see IndexProgress in indexer.py) — read, never copied, + # by the handshake ack and the periodic progress pusher. + "progress": indexer.progress, + # Bound method, called when a peer completes the + # handshake — resets reconcile's backoff (indexer.py + # _reconcile_loop) so the backstop is prompt again now + # that someone is actually looking. + "note_activity": indexer.note_activity, + # Shown to the operator in Settings, and kept current in + # place by set_scan_settings (ops.py) — same reasoning as + # member_upload below. + "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], + "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, # Admission policy comes from node.toml, never from the hub: # a hub that could declare a group open would be handed its key. @@ -456,6 +500,8 @@ class NodeDaemon: if gctx: self._tasks.append(asyncio.create_task( _bg_scan(idx, group_cfg.name, gctx))) + self._tasks.append(asyncio.create_task( + self._progress_pusher(idx))) # 12. Wait for shutdown stop_event = asyncio.Event() @@ -561,18 +607,40 @@ class NodeDaemon: if gek: log.info("GEK loaded for new group %s", group_cfg.id[:8]) + index_cache = IndexCache( + db_path=data_dir / group_cfg.id[:16] / "index_cache.db") + await index_cache.open() + self._index_caches[group_cfg.id] = index_cache + + scan_settings = ( + await self._roster.scan_settings(group_cfg.id) + if self._roster else { + "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS, + "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS, + }) + indexer = DirectoryIndexer( roots=roots, group_id=group_cfg.id, sk_node=sk_ed, gek=gek, on_change=self._on_index_change, + cache=index_cache, + reconcile_secs=scan_settings["reconcile_interval_secs"], + debounce_secs=scan_settings["debounce_secs"], ) - await indexer.start() + # Registered *before* start() runs its (blocking, possibly very + # long — see the StarWars benchmark) initial scan, specifically + # so /api/groups/{id}/index-status can see indexer.progress + # while a brand-new group is still scanning — this is the one + # group state that must stay visible during the very window the + # group is not yet authorized for member connections (below). self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index self._state["indexers"][group_cfg.id] = indexer + await indexer.start() + data_dir.mkdir(parents=True, exist_ok=True) chat_db = data_dir / group_cfg.id[:16] / "chat.db" store = ChatStore(db_path=chat_db) @@ -583,6 +651,10 @@ class NodeDaemon: "gek": gek, "roots": roots, "index": indexer.index, + "progress": indexer.progress, + "note_activity": indexer.note_activity, + "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], + "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, "join_policy": group_cfg.join_policy, "member_upload": ( @@ -600,6 +672,11 @@ class NodeDaemon: log.info("Hot-loaded group %s (%s, %d roots)", group_cfg.name, group_cfg.id[:8], len(roots)) added_names.append(group_cfg.name) + # The initial scan above already ran to completion (indexer.start() + # is not deferred here), so this only matters for whatever scans + # this group as time goes on — a root added later, reconcile + # picking one back up. + self._tasks.append(asyncio.create_task(self._progress_pusher(indexer))) # ── Tear down removed groups ───────────────────────────────────── removed_names = [] @@ -618,6 +695,16 @@ class NodeDaemon: await store.close() except Exception: pass + cache = self._index_caches.pop(gid, None) + if cache: + try: + await cache.close() + except Exception: + pass + pending = self._pending_broadcasts.pop(gid, None) + if pending: + pending.cancel() + self._last_broadcast_snapshot.pop(gid, None) self._state["indexes"].pop(gid, None) self._state["indexers"].pop(gid, None) old_name = gid[:8] @@ -708,46 +795,149 @@ class NodeDaemon: log.warning("No unwrappable GEK bundle found for group %s", group_id[:8]) return None + async def _progress_pusher(self, indexer: DirectoryIndexer, + interval: float = 2.0) -> None: + """ + Watches indexer.progress and pushes a light INDEX_PROGRESS message to + this group's connected peers — never the index itself, that stays + _on_index_change's job. Runs for the node's whole lifetime: a scan + can start from several places (initial scan, a root added later, + reconcile picking a root back up), and this only needs to notice the + flag, not why it changed. + + The final push at the False transition is what lets a presence dot + reliably turn back off on an already-connected client — the + handshake ack only covers the moment of connecting. `interval` is a + parameter (not a bare constant) only so a test can drive this loop + without waiting on the real 2s cadence. + """ + was_scanning = False + while True: + await asyncio.sleep(interval) + progress = indexer.progress + now_scanning = progress.scanning + if now_scanning or was_scanning: + self._push_index_progress(indexer.group_id, progress) + was_scanning = now_scanning + + def _push_index_progress(self, group_id: str, progress) -> None: + if not self._webrtc: + return + msg = { + "type": MNP.INDEX_PROGRESS, + "v": MNP_VERSION, + "group_id": group_id, + "scanning": progress.scanning, + "scanned_bytes": progress.scanned_bytes, + "total_bytes": progress.total_bytes, + } + pushed = 0 + for session in list(self._webrtc._sessions.values()): + if session._group_id == group_id: + try: + session._send(msg) + pushed += 1 + except Exception: + pass + if pushed: + log.debug("Index progress pushed to %d peer(s) for group %s", + pushed, group_id[:8]) + async def _on_index_change(self, indexer: DirectoryIndexer) -> None: - """Called when a DirectoryIndexer detects file changes.""" + """ + Called when a DirectoryIndexer detects file changes — once per + debounced watchdog event, so dropping N files into a watched folder + calls this N times in quick succession. Coalesces those into one + broadcast (_broadcast_index_change) rather than one push per file: + the timer is reset on every call and only fires once calls stop + arriving for _broadcast_coalesce_secs. + """ + group_id = indexer.group_id + loop = asyncio.get_event_loop() + pending = self._pending_broadcasts.pop(group_id, None) + if pending: + pending.cancel() + + def fire() -> None: + self._pending_broadcasts.pop(group_id, None) + asyncio.ensure_future(self._broadcast_index_change(indexer)) + + self._pending_broadcasts[group_id] = loop.call_later( + self._broadcast_coalesce_secs, fire) + + async def _broadcast_index_change(self, indexer: DirectoryIndexer) -> None: + """ + The actual push, run once per coalesced burst. Sends a full + INDEX_SYNC the first time a group is ever broadcast (no previous + snapshot to diff against — the client's own first fetchIndex() call + already covers that case) and an INDEX_DELTA every time after, + computed against the last thing this method actually sent. + """ group_id = indexer.group_id idx = indexer.index log.info("Index changed for group %s: %d files (v%d)", group_id[:8], idx.count, idx.version) - # 11.5 — Push updated index to connected WebRTC peers in this group + prev = self._last_broadcast_snapshot.get(group_id) + delta = None + if prev is not None: + prev_version, prev_entries = prev + previous = GroupIndex._snapshot( + idx.group_id, idx.sk_node, idx.gek, prev_version, prev_entries) + delta = idx.diff(previous) + self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id()) + + # 11.5 — Push to connected WebRTC peers in this group if self._webrtc: - entries = [ - { - "id": e.id, "name": e.name, "path": e.path, - "size": e.size, "type": e.type, "added_at": e.added_at, + if delta is not None: + msg = { + "type": MNP.INDEX_DELTA, + "v": MNP_VERSION, + "group_id": idx.group_id, + "base_version": delta.base_version, + "version": delta.version, + "additions": [ + {"id": e.id, "name": e.name, "path": e.path, + "size": e.size, "type": e.type, "added_at": e.added_at} + for e in delta.additions + ], + "deletions": delta.deletions, + } + else: + msg = { + "type": MNP.INDEX_SYNC, + "v": MNP_VERSION, + "group_id": idx.group_id, + "version": idx.version, + "entries": [ + {"id": e.id, "name": e.name, "path": e.path, + "size": e.size, "type": e.type, "added_at": e.added_at} + for e in idx.entries + ], } - for e in idx.entries - ] - sync_msg = { - "type": MNP.INDEX_SYNC, - "v": MNP_VERSION, - "group_id": idx.group_id, - "version": idx.version, - "entries": entries, - } pushed = 0 for session in list(self._webrtc._sessions.values()): if session._group_id == group_id: try: - session._send(sync_msg) + session._send(msg) pushed += 1 except Exception: pass if pushed: - log.info("Index pushed to %d WebRTC peers", pushed) + log.info("Index %s pushed to %d WebRTC peers", + "delta" if delta is not None else "sync", pushed) # 11.9 — Register file hashes with hub swarm table (public groups only, H7) group_cfg = next( (g for g in self._config.groups if g.id == group_id), None) if (self._hub and self._state.get("endpoint_hint") and group_cfg and group_cfg.visibility == "public"): - hashes = [e.id for e in idx.entries] + # Only the newly added hashes once there is a delta to know them + # from — registering the whole library again on every change is + # the same O(changes x library size) cost the delta above exists + # to avoid. + hashes = ([e.id for e in delta.additions] if delta is not None + else [e.id for e in idx.entries]) if hashes: endpoint = f"webrtc:{self._config.node.quic_port}" asyncio.ensure_future(self._register_swarm(hashes, endpoint)) @@ -772,6 +962,10 @@ class NodeDaemon: log.info("Shutting down...") self._state["status"] = "stopping" + for handle in self._pending_broadcasts.values(): + handle.cancel() + self._pending_broadcasts.clear() + for task in self._tasks: task.cancel() for task in self._tasks: @@ -795,6 +989,9 @@ class NodeDaemon: for store in self._chat_stores.values(): await store.close() + for cache in self._index_caches.values(): + await cache.close() + for indexer in self._indexers: await indexer.stop() |