diff options
Diffstat (limited to 'packages/meshbay-node')
4 files changed, 78 insertions, 46 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index e11a654..5fc70fe 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -241,14 +241,13 @@ class NodeDaemon: gek=gek, on_change=self._on_index_change, ) - await indexer.start() + await indexer.start(defer_scan=True) self._indexers.append(indexer) self._state["indexes"][group_cfg.id] = indexer.index self._state["indexers"][group_cfg.id] = indexer - log.info("Indexing group %s: %s (%d files)", + log.info("Group %s configured: %s (scan deferred)", group_cfg.name, - ", ".join(f"{r.name}={r.path}" for r in roots), - indexer.index.count) + ", ".join(f"{r.name}={r.path}" for r in roots)) groups_ctx[group_cfg.id] = { "gek": gek, @@ -268,8 +267,8 @@ class NodeDaemon: } if not groups_ctx: - log.error("No valid groups configured — exiting") - return + log.warning("No groups configured yet — admin UI and hub " + "connection stay up; attach a group to go live") # 5. Chat stores (one SQLite DB per group) for gid in groups_ctx: @@ -290,14 +289,14 @@ class NodeDaemon: denylist = self._denylist # 6. WebRTC transport (browser clients) - first = next(iter(groups_ctx.values())) + first = next(iter(groups_ctx.values()), None) if WEBRTC_AVAILABLE: self._webrtc = WebRTCTransport( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, - gek=first["gek"], - roots=first["roots"], - index=first["index"], + gek=first["gek"] if first else None, + roots=first["roots"] if first else None, + index=first["index"] if first else None, groups=groups_ctx, denylist=denylist, max_concurrent_streams=self._config.node.max_concurrent_streams, @@ -307,6 +306,7 @@ class NodeDaemon: # _group_ctx(). Assigning the first group's store transport-wide # sent every group's chat to one database and served it back to # members of every other group (finding H1). + self._webrtc._ctx["groups"] = groups_ctx self._webrtc._ctx["hub_ws"] = _WsSender(hub) self._webrtc._ctx["node_user_id"] = session.user_id self._webrtc._ctx["audit_store"] = self._audit_store @@ -342,14 +342,15 @@ class NodeDaemon: self._quic_server = QuicChunkServer( sk_node=keys.sk_ed25519, hub_pk_pem=session.hub_pk_pem, - gek=first["gek"], - roots=first["roots"], - index=first["index"], + gek=first["gek"] if first else None, + roots=first["roots"] if first else None, + index=first["index"] if first else None, host="::", port=self._config.node.quic_port, groups=groups_ctx, denylist=denylist, ) + self._quic_server._ctx["groups"] = groups_ctx await self._quic_server.start() log.info("QUIC server on port %d (%d groups)", self._config.node.quic_port, len(groups_ctx)) @@ -434,18 +435,23 @@ class NodeDaemon: "yes" if self._webrtc else "no", "yes" if self._quic_server else "no") - # 11. Initial swarm registration — PUBLIC groups only. - # Finding H7: registering every group's hashes hands the hub a content - # fingerprint of every private file on the node, which is exactly the - # metadata the "hub stores no content metadata" claim rules out. It also - # lets anyone confirm whether a known file exists in the network. - endpoint = f"webrtc:{self._config.node.quic_port}" - for gctx in groups_ctx.values(): - if gctx.get("visibility") != "public": - continue - hashes = [e.id for e in gctx["index"].entries] - if hashes: - asyncio.ensure_future(self._register_swarm(hashes, endpoint)) + # 11. Background initial scan — files appear progressively. + async def _bg_scan(indexer, name, gctx): + await indexer.initial_scan() + log.info("Background scan complete for %s: %d files", + name, indexer.index.count) + # Swarm registration for public groups (after files are known). + if gctx.get("visibility") == "public": + endpoint = f"webrtc:{self._config.node.quic_port}" + hashes = [e.id for e in gctx["index"].entries] + if hashes: + await self._register_swarm(hashes, endpoint) + + for idx, group_cfg in zip(self._indexers, self._config.groups): + gctx = groups_ctx.get(group_cfg.id) + if gctx: + self._tasks.append(asyncio.create_task( + _bg_scan(idx, group_cfg.name, gctx))) # 12. Wait for shutdown stop_event = asyncio.Event() @@ -486,7 +492,7 @@ class NodeDaemon: log.error("Reload failed, keeping the running config: %s", e) return - groups_ctx = self._state.get("groups_ctx") or {} + groups_ctx = self._state.get("groups_ctx", {}) hosted = set(groups_ctx) incoming = {g.id for g in fresh.groups if g.id} @@ -647,7 +653,7 @@ class NodeDaemon: log.warning( "Node key not linked. Open the admin UI, copy this " "node's key, and paste it in Settings > Link Node on " - "%s. Retrying in 30s...", + "%s. Retrying in 5s...", self._config.hub.url, ) else: @@ -655,10 +661,10 @@ class NodeDaemon: log.warning( "Hub rejected the node credentials for user %r. " "Register that account on %s first, then link this " - "node's key. Retrying in 30s...", + "node's key. Retrying in 5s...", self._config.hub.username, self._config.hub.url, ) - await asyncio.sleep(30) + await asyncio.sleep(5) else: raise except Exception as e: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index d9b1d2c..482b556 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -240,10 +240,16 @@ class DirectoryIndexer: # ── Watchdog integration ────────────────────────────────────────────────── - async def start(self) -> None: - """Start initial scan + filesystem watcher + reconciler.""" + async def start(self, *, defer_scan: bool = False) -> None: + """Start initial scan + filesystem watcher + reconciler. + + With ``defer_scan=True`` the watcher and reconciler start + immediately but the initial scan is skipped — call + :meth:`initial_scan` yourself when ready. + """ self._loop = asyncio.get_event_loop() - await self.initial_scan() + if not defer_scan: + await self.initial_scan() self._start_observer() self._reconciler = asyncio.create_task(self._reconcile_loop()) diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py index 7974be5..ab1613d 100644 --- a/packages/meshbay-node/tests/test_daemon.py +++ b/packages/meshbay-node/tests/test_daemon.py @@ -160,11 +160,11 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem) assert store._db is None @pytest.mark.asyncio -async def test_daemon_no_groups_exits(tmp_path): - """Daemon with no valid groups exits cleanly.""" +async def test_daemon_no_groups_stays_up(tmp_path, hub_pk_pem): + """Daemon with no valid groups stays up (admin UI + hub connection alive).""" config = Config( hub=HubConfig(url="http://localhost:9999", username="testuser"), - node=NodeConfig(), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), groups=[GroupConfig(id="", name="empty", shared_dir="")], keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), data_dir=tmp_path / "data", @@ -177,28 +177,48 @@ async def test_daemon_no_groups_exits(tmp_path): mock_session = MagicMock() mock_session.node_id = "node123" mock_session.user_id = "user123" - mock_session.hub_pk_pem = b"pem" + mock_session.hub_pk_pem = hub_pk_pem - mock_server = AsyncMock() - mock_server.serve = AsyncMock() + shutdown_event = asyncio.Event() with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \ - patch("meshbay_node.daemon.HubClient") as MockHub, \ - patch("meshbay_node.daemon.uvicorn") as mock_uvicorn: - - mock_uvicorn.Config = MagicMock() - mock_uvicorn.Server = MagicMock(return_value=mock_server) + patch("meshbay_node.daemon.HubClient") as MockHub: 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 - await daemon.run() + async def mock_maintain_ws(**kwargs): + await shutdown_event.wait() + + hub_instance.maintain_ws = mock_maintain_ws - assert len(daemon._chat_stores) == 0 + async def run_daemon(): + with patch("signal.SIGINT", 2), \ + patch("signal.SIGTERM", 15): + try: + await asyncio.wait_for(daemon.run(), timeout=5) + except (asyncio.TimeoutError, Exception): + pass + + task = asyncio.create_task(run_daemon()) + await asyncio.sleep(1) + + assert daemon._state["status"] == "running" + assert len(daemon._chat_stores) == 0 + + shutdown_event.set() + await daemon._shutdown() + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass @pytest.mark.asyncio async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hub_pk_pem): diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index 78a631a..725b800 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -488,7 +488,7 @@ def test_swarm_registration_skips_private_groups(): / "daemon.py").read_text() assert 'visibility' in source and '_register_swarm' in source # Both registration sites must gate on public visibility. - for marker in ['gctx.get("visibility") != "public"', + for marker in ['gctx.get("visibility") == "public"', 'group_cfg.visibility == "public"']: assert marker in source, f"swarm registration not gated: {marker}" |