From c66ee41d8476461939c5f4e7fdc71c5d7fb4a85c Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Tue, 11 Aug 2026 22:16:59 +0200 Subject: feat(node): index push to WebRTC peers + swarm registration (11.5, 11.9) When watchdog detects file changes, the daemon now: - Pushes INDEX_SYNC to all connected WebRTC peers in that group - Registers file hashes with hub /v1/swarm/register endpoint Also registers all file hashes on startup for initial discovery. hub_client: add register_swarm() method for bulk hash registration. Co-Authored-By: Claude Opus 4.6 --- packages/meshbay-node/src/meshbay_node/daemon.py | 60 +++++++++++++- .../meshbay-node/src/meshbay_node/hub_client.py | 21 +++++ packages/meshbay-node/tests/test_daemon.py | 94 +++++++++++++++++++++- 3 files changed, 173 insertions(+), 2 deletions(-) (limited to 'packages') diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 6a8bbc9..5b6e770 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -31,6 +31,8 @@ from pathlib import Path import uvicorn +from meshbay_common import MNP_VERSION +from meshbay_common.protocol import MNP from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, load_config, write_example_config from meshbay_node.hub_client import HubClient, HubConfig @@ -161,6 +163,7 @@ class NodeDaemon: group_id=group_cfg.id, sk_node=keys.sk_ed25519, gek=gek, + on_change=self._on_index_change, ) await indexer.start() self._indexers.append(indexer) @@ -330,7 +333,14 @@ class NodeDaemon: "yes" if self._webrtc else "no", "yes" if self._quic_server else "no") - # 11. Wait for shutdown + # 11. Initial swarm registration + endpoint = f"webrtc:{self._config.node.quic_port}" + for gctx in groups_ctx.values(): + hashes = [e.id for e in gctx["index"].entries] + if hashes: + asyncio.ensure_future(self._register_swarm(hashes, endpoint)) + + # 12. Wait for shutdown stop_event = asyncio.Event() loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): @@ -339,6 +349,54 @@ class NodeDaemon: await self._shutdown() + async def _on_index_change(self, indexer: DirectoryIndexer) -> None: + """Called when a DirectoryIndexer detects file changes.""" + 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 + if self._webrtc: + 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 + ] + 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) + pushed += 1 + except Exception: + pass + if pushed: + log.info("Index pushed to %d WebRTC peers", pushed) + + # 11.9 — Register file hashes with hub swarm table + if self._hub and self._state.get("endpoint_hint"): + hashes = [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)) + + async def _register_swarm(self, hashes: list[str], endpoint: str) -> None: + try: + n = await self._hub.register_swarm(hashes, endpoint) + log.info("Swarm: registered %d/%d hashes", n, len(hashes)) + except Exception as e: + log.warning("Swarm registration failed: %s", e) + async def _shutdown(self) -> None: log.info("Shutting down...") self._state["status"] = "stopping" diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py index ca0e776..ba9d3ff 100644 --- a/packages/meshbay-node/src/meshbay_node/hub_client.py +++ b/packages/meshbay-node/src/meshbay_node/hub_client.py @@ -329,6 +329,27 @@ class HubClient: finally: self._ws = None + # ── Swarm registration ───────────────────────────────────────────────── + + async def register_swarm(self, content_hashes: list[str], endpoint: str) -> int: + """Register file hashes in the hub swarm table. Returns count registered.""" + if self._session is None: + raise RuntimeError("Not logged in") + await self.ensure_fresh_token() + + registered = 0 + for h in content_hashes: + try: + r = await self._http.post("/v1/swarm/register", json={ + "content_hash": h, + "endpoint": endpoint, + }, headers=self._session.auth_headers) + if r.status_code in (201, 200): + registered += 1 + except Exception: + pass + return registered + # ── Convenience: full startup sequence ─────────────────────────────────── async def startup(self, endpoint_hint: str | None = None) -> HubSession: diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py index e74f7fc..faf12e3 100644 --- a/packages/meshbay-node/tests/test_daemon.py +++ b/packages/meshbay-node/tests/test_daemon.py @@ -2,7 +2,8 @@ Integration test: Node daemon wires all components correctly. Phase 11 — verifies that NodeDaemon creates chat stores, WebRTC transport, -and shuts down cleanly. Hub interaction is mocked. +index push on change, swarm registration, and shuts down cleanly. +Hub interaction is mocked. """ import asyncio @@ -16,6 +17,7 @@ from unittest.mock import AsyncMock, MagicMock, patch from meshbay_common.crypto import generate_gek from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer import DirectoryIndexer @pytest.fixture @@ -169,3 +171,93 @@ async def test_daemon_no_groups_exits(tmp_path): assert daemon._state["status"] == "starting" assert len(daemon._chat_stores) == 0 + + +@pytest.mark.asyncio +async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hub_pk_pem): + """Index change callback pushes updated index to WebRTC peers.""" + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"), + node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000), + groups=[GroupConfig( + id="a" * 32, + name="test-group", + shared_dir=str(shared_dir), + visibility="private", + port=29000, quic_port=29010, http_port=29001, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + daemon._hub = AsyncMock() + daemon._hub.register_swarm = AsyncMock(return_value=2) + daemon._state["endpoint_hint"] = "node123" + + sk_node = Ed25519PrivateKey.generate() + indexer = DirectoryIndexer( + root=shared_dir, group_id="a" * 32, + sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + mock_session = MagicMock() + mock_session._group_id = "a" * 32 + mock_session._send = MagicMock() + + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"peer1": mock_session} + daemon._webrtc = mock_webrtc + + await daemon._on_index_change(indexer) + + mock_session._send.assert_called_once() + msg = mock_session._send.call_args[0][0] + assert msg["type"] == "index_sync" + assert msg["group_id"] == "a" * 32 + assert len(msg["entries"]) == indexer.index.count + + await asyncio.sleep(0.1) + daemon._hub.register_swarm.assert_called_once() + call_args = daemon._hub.register_swarm.call_args + assert len(call_args[0][0]) == indexer.index.count + + +@pytest.mark.asyncio +async def test_daemon_index_change_skips_other_group_peers( + tmp_path, shared_dir, gek, hub_pk_pem +): + """Index change only pushes to peers in the same group.""" + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"), + node=NodeConfig(), + groups=[], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + daemon._hub = AsyncMock() + daemon._hub.register_swarm = AsyncMock(return_value=0) + daemon._state["endpoint_hint"] = "node123" + + sk_node = Ed25519PrivateKey.generate() + indexer = DirectoryIndexer( + root=shared_dir, group_id="a" * 32, + sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + same_group = MagicMock() + same_group._group_id = "a" * 32 + same_group._send = MagicMock() + + other_group = MagicMock() + other_group._group_id = "b" * 32 + other_group._send = MagicMock() + + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": same_group, "p2": other_group} + daemon._webrtc = mock_webrtc + + await daemon._on_index_change(indexer) + + same_group._send.assert_called_once() + other_group._send.assert_not_called() -- cgit v1.2.3