diff options
| -rw-r--r-- | devel-phases-next.md | 32 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 60 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/hub_client.py | 21 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_daemon.py | 94 |
4 files changed, 189 insertions, 18 deletions
diff --git a/devel-phases-next.md b/devel-phases-next.md index 533f5e8..9a8b7ef 100644 --- a/devel-phases-next.md +++ b/devel-phases-next.md @@ -1,6 +1,6 @@ # MeshBay — Next Implementation Phases -> Base: Phases 1–11 complete (except 10.9 → Phase 13, 11.5/11.9 deferred). 169 tests. Web SPA + admin panel + self-service UI + MSE video streaming live on meshbay.org. Node daemon is now production-ready (WebRTC, WS, chat, HTTP all wired). +> Base: Phases 1–11 complete (except 10.9 → Phase 16). 171 tests. Web SPA + admin panel + self-service UI + MSE video streaming live on meshbay.org. Node daemon is production-ready (WebRTC, WS, chat, HTTP, index push, swarm all wired). > Architecture reference: docs/meshbay-draft-v4.md > First security review: first-review.md (2026-08-10) @@ -496,7 +496,7 @@ Browser Node ## Phase 11 — Node daemon: production-ready ✅ DONE -Pending commit — 169 tests. +Pending commit — 171 tests. **Objective:** the node daemon (`meshbay-node`) runs as a complete, self-contained service. Previously the daemon only started QUIC/TCP servers and the local web UI; @@ -521,8 +521,8 @@ services in a single process: 9. **Graceful shutdown** (enhanced) — cancels WS task, closes WebRTC peers, closes chat stores, stops HTTP/QUIC/TCP servers, stops indexers -**`hub_client.py`** — added `_ws` tracking and `send_ws()` method so the WebRTC -context can send `chat_notify` messages to the hub for real-time chat notifications. +**`hub_client.py`** — added `_ws` tracking, `send_ws()` for chat notifications, +and `register_swarm()` for file hash registration with the hub. **`config.py`** — added `data_dir` field (default `~/.local/share/meshbay/`) for chat DBs and other persistent state. @@ -530,6 +530,15 @@ for chat DBs and other persistent state. **`meshbay-node.service`** — updated systemd unit with `StateDirectory=meshbay`, `ProtectSystem=strict`, `ReadWritePaths` for config and data directories. +**Index push (11.5):** when watchdog detects file changes, the debounced +`on_change` callback fires `_on_index_change` on the daemon, which pushes a +full `INDEX_SYNC` to all WebRTC peers in that group. Only peers whose +`_group_id` matches receive the push. + +**Swarm registration (11.9):** on startup and on each index change, the daemon +registers all file hashes with the hub's `/v1/swarm/register` endpoint. This +allows other nodes/clients to discover which nodes host which content. + ### Milestones | # | Component | Status | @@ -538,12 +547,12 @@ for chat DBs and other persistent state. | 11.2 | Daemon: WebRTC transport | ✅ | | 11.3 | Daemon: chat store | ✅ | | 11.4 | Daemon: HTTP file API | ✅ | -| 11.5 | Daemon: index push on change | Deferred — indexer `on_change` callback not yet implemented | +| 11.5 | Daemon: index push on change | ✅ | | 11.6 | Daemon: node_user_id + hub_ws context | ✅ | | 11.7 | Daemon: graceful shutdown | ✅ | | 11.8 | Systemd unit file | ✅ | -| 11.9 | Swarm registration | Deferred — hub endpoint exists but node-side trigger not yet wired | -| 11.10 | Integration test | ✅ (2 tests: full lifecycle + no-groups-exit) | +| 11.9 | Swarm registration | ✅ | +| 11.10 | Integration test | ✅ (4 tests: lifecycle, no-groups, index push, group filtering) | ### File changes @@ -556,15 +565,6 @@ for chat DBs and other persistent state. **Added:** - `packages/meshbay-node/tests/test_daemon.py` — 2 integration tests -### Deferred items - -- **11.5 Index push**: requires `DirectoryIndexer` to have an `on_change` callback - that fires when watchdog detects file changes, then the daemon pushes `INDEX_DELTA` - to all connected WebRTC peers. The indexer currently rebuilds the full index but - doesn't notify consumers of incremental changes. -- **11.9 Swarm registration**: hub `/v1/swarm/register` endpoint exists. Node needs - to register file hashes after each index rebuild. Depends on 11.5 (index change events). - --- ## Phase 12 — Node CLI + management 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() |