clawft_kernel/
mesh_discovery.rs

1//! Peer discovery for mesh networking (K6.2).
2//!
3//! Discovery is the mechanism by which nodes find each other.
4//! Three strategies are supported:
5//! - Static seed peers (always available, no extra deps)
6//! - mDNS for LAN discovery (behind mesh-discovery feature)
7//! - Kademlia DHT for WAN discovery (behind mesh-discovery feature)
8//!
9//! # Peer event stream (WEFT-120)
10//!
11//! Live mesh membership changes are published as [`MeshPeerEvent`] on a
12//! [`MeshPeerEventBus`]. [`crate::mesh_runtime::MeshRuntime`] emits events
13//! on connect / disconnect / health transitions; [`crate::cluster::ClusterMembership`]
14//! (and `ClusterService` when the `cluster` feature is on) subscribe so
15//! cluster membership reflects live mesh state without manual pull-sync.
16
17use async_trait::async_trait;
18use serde::{Deserialize, Serialize};
19use tokio::sync::broadcast;
20
21/// A discovered peer.
22#[derive(Debug, Clone, Serialize, Deserialize)]
23pub struct DiscoveredPeer {
24    /// Node identifier.
25    pub node_id: String,
26    /// Network address.
27    pub address: String,
28    /// Platform type.
29    pub platform: String,
30    /// How this peer was discovered.
31    pub source: DiscoverySource,
32}
33
34/// How a peer was discovered.
35#[non_exhaustive]
36#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
37pub enum DiscoverySource {
38    /// Static seed peer from configuration.
39    SeedPeer,
40    /// Discovered via mDNS on the local network.
41    Mdns,
42    /// Discovered via Kademlia DHT.
43    Kademlia,
44    /// Learned from another peer's peer list exchange.
45    PeerExchange,
46    /// Manually added by operator.
47    Manual,
48}
49
50/// Live mesh peer membership / health change (WEFT-120).
51///
52/// Published by the mesh runtime whenever a peer joins, leaves, or
53/// transitions health state. Consumers (notably `ClusterService` /
54/// `ClusterMembership`) apply these atomically so cluster membership
55/// mirrors mesh state without a pull-based `sync_to_membership` loop.
56#[non_exhaustive]
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
58pub enum MeshPeerEvent {
59    /// Peer established a mesh connection (or reconnected after drop).
60    Joined {
61        /// Remote node id.
62        node_id: String,
63        /// Network address if known (seed / discovery map).
64        address: Option<String>,
65        /// Platform label when known.
66        platform: Option<String>,
67        /// The peer counts as a cluster member: admission verified `node_id`
68        /// for this connection, or admission is not `enforce` (under the
69        /// default `observe`, and under `off`, every peer counts). False only
70        /// under `enforce` for a connection whose id is just a claim (a
71        /// dialled seed, a legacy leaf); membership then holds it as
72        /// [`NodeState::Unverified`](crate::cluster::NodeState::Unverified),
73        /// not `Active`. Absent in older events (read as false).
74        #[serde(default)]
75        verified: bool,
76    },
77    /// Peer disconnected (graceful leave or connection drop).
78    Left {
79        /// Remote node id.
80        node_id: String,
81    },
82    /// Heartbeat / successful probe — peer is alive.
83    Alive {
84        /// Remote node id.
85        node_id: String,
86    },
87    /// SWIM / heartbeat marked the peer suspect.
88    Suspect {
89        /// Remote node id.
90        node_id: String,
91    },
92    /// Peer confirmed unreachable (partition or dead).
93    Unreachable {
94        /// Remote node id.
95        node_id: String,
96    },
97    /// Peer recovered after Unreachable / Suspect (partition recovery).
98    Recovered {
99        /// Remote node id.
100        node_id: String,
101        /// Address if known on recovery path.
102        address: Option<String>,
103        /// As for [`MeshPeerEvent::Joined::verified`].
104        #[serde(default)]
105        verified: bool,
106    },
107}
108
109impl MeshPeerEvent {
110    /// Node id referenced by this event.
111    pub fn node_id(&self) -> &str {
112        match self {
113            MeshPeerEvent::Joined { node_id, .. }
114            | MeshPeerEvent::Left { node_id }
115            | MeshPeerEvent::Alive { node_id }
116            | MeshPeerEvent::Suspect { node_id }
117            | MeshPeerEvent::Unreachable { node_id }
118            | MeshPeerEvent::Recovered { node_id, .. } => node_id.as_str(),
119        }
120    }
121}
122
123/// Default capacity for the peer-event broadcast bus.
124const DEFAULT_PEER_EVENT_CAPACITY: usize = 256;
125
126/// Fan-out bus for [`MeshPeerEvent`] (tokio broadcast).
127///
128/// Multiple subscribers (cluster membership, observability, tests) can
129/// each hold a receiver. Slow consumers lag and skip old events rather
130/// than blocking mesh I/O.
131#[derive(Clone, Debug)]
132pub struct MeshPeerEventBus {
133    tx: broadcast::Sender<MeshPeerEvent>,
134}
135
136impl MeshPeerEventBus {
137    /// Create a bus with the default capacity.
138    pub fn new() -> Self {
139        Self::with_capacity(DEFAULT_PEER_EVENT_CAPACITY)
140    }
141
142    /// Create a bus with an explicit ring capacity.
143    pub fn with_capacity(capacity: usize) -> Self {
144        let (tx, _) = broadcast::channel(capacity.max(1));
145        Self { tx }
146    }
147
148    /// Publish a peer event (best-effort; no-op when no receivers).
149    pub fn emit(&self, event: MeshPeerEvent) {
150        let _ = self.tx.send(event);
151    }
152
153    /// Subscribe to the peer-event stream.
154    pub fn subscribe(&self) -> broadcast::Receiver<MeshPeerEvent> {
155        self.tx.subscribe()
156    }
157
158    /// Number of active receivers.
159    pub fn receiver_count(&self) -> usize {
160        self.tx.receiver_count()
161    }
162
163    /// Shared sender for advanced wiring (e.g. external discovery).
164    pub fn sender(&self) -> broadcast::Sender<MeshPeerEvent> {
165        self.tx.clone()
166    }
167}
168
169impl Default for MeshPeerEventBus {
170    fn default() -> Self {
171        Self::new()
172    }
173}
174
175/// Discovery backend trait.
176#[async_trait]
177pub trait DiscoveryBackend: Send + Sync + 'static {
178    /// Name of this discovery mechanism.
179    fn name(&self) -> &str;
180    /// Start the discovery mechanism.
181    async fn start(&mut self) -> Result<(), DiscoveryError>;
182    /// Poll for newly discovered peers (non-blocking).
183    async fn poll(&mut self) -> Vec<DiscoveredPeer>;
184    /// Stop the discovery mechanism.
185    async fn stop(&mut self) -> Result<(), DiscoveryError>;
186}
187
188/// Discovery errors.
189#[non_exhaustive]
190#[derive(Debug, thiserror::Error)]
191pub enum DiscoveryError {
192    #[error("discovery backend error: {0}")]
193    Backend(String),
194    #[error("invalid seed address: {0}")]
195    InvalidAddress(String),
196    #[error("discovery not started")]
197    NotStarted,
198}
199
200/// Discovery coordinator that manages multiple discovery backends.
201pub struct DiscoveryCoordinator {
202    backends: Vec<Box<dyn DiscoveryBackend>>,
203    known_peers: std::collections::HashSet<String>,
204}
205
206impl DiscoveryCoordinator {
207    pub fn new() -> Self {
208        Self {
209            backends: Vec::new(),
210            known_peers: std::collections::HashSet::new(),
211        }
212    }
213
214    /// Add a discovery backend.
215    pub fn add_backend(&mut self, backend: Box<dyn DiscoveryBackend>) {
216        self.backends.push(backend);
217    }
218
219    /// Start all discovery backends.
220    pub async fn start_all(&mut self) -> Result<(), DiscoveryError> {
221        for backend in &mut self.backends {
222            backend.start().await?;
223        }
224        Ok(())
225    }
226
227    /// Poll all backends for new peers. Returns only newly discovered peers.
228    pub async fn poll_all(&mut self) -> Vec<DiscoveredPeer> {
229        let mut new_peers = Vec::new();
230        for backend in &mut self.backends {
231            let peers = backend.poll().await;
232            for peer in peers {
233                if self.known_peers.insert(peer.node_id.clone()) {
234                    new_peers.push(peer);
235                }
236            }
237        }
238        new_peers
239    }
240
241    /// Stop all discovery backends.
242    pub async fn stop_all(&mut self) -> Result<(), DiscoveryError> {
243        for backend in &mut self.backends {
244            backend.stop().await?;
245        }
246        Ok(())
247    }
248
249    /// Number of known peers.
250    pub fn known_peer_count(&self) -> usize {
251        self.known_peers.len()
252    }
253
254    /// Check if a peer is already known.
255    pub fn is_known(&self, node_id: &str) -> bool {
256        self.known_peers.contains(node_id)
257    }
258}
259
260impl Default for DiscoveryCoordinator {
261    fn default() -> Self {
262        Self::new()
263    }
264}
265
266#[cfg(test)]
267mod tests {
268    use super::*;
269
270    /// A mock discovery backend for testing.
271    struct MockBackend {
272        name: &'static str,
273        peers: Vec<DiscoveredPeer>,
274        started: bool,
275    }
276
277    impl MockBackend {
278        fn new(name: &'static str, peers: Vec<DiscoveredPeer>) -> Self {
279            Self {
280                name,
281                peers,
282                started: false,
283            }
284        }
285    }
286
287    #[async_trait]
288    impl DiscoveryBackend for MockBackend {
289        fn name(&self) -> &str {
290            self.name
291        }
292        async fn start(&mut self) -> Result<(), DiscoveryError> {
293            self.started = true;
294            Ok(())
295        }
296        async fn poll(&mut self) -> Vec<DiscoveredPeer> {
297            std::mem::take(&mut self.peers)
298        }
299        async fn stop(&mut self) -> Result<(), DiscoveryError> {
300            self.started = false;
301            Ok(())
302        }
303    }
304
305    fn make_peer(id: &str, source: DiscoverySource) -> DiscoveredPeer {
306        DiscoveredPeer {
307            node_id: id.to_string(),
308            address: format!("127.0.0.1:900{}", id.len()),
309            platform: "linux".into(),
310            source,
311        }
312    }
313
314    #[tokio::test]
315    async fn coordinator_add_and_start() {
316        let mut coord = DiscoveryCoordinator::new();
317        assert_eq!(coord.known_peer_count(), 0);
318
319        let backend = MockBackend::new("mock", vec![make_peer("a", DiscoverySource::Manual)]);
320        coord.add_backend(Box::new(backend));
321        coord.start_all().await.unwrap();
322    }
323
324    #[tokio::test]
325    async fn coordinator_poll_deduplicates() {
326        let mut coord = DiscoveryCoordinator::new();
327
328        let b1 = MockBackend::new(
329            "b1",
330            vec![
331                make_peer("node-1", DiscoverySource::SeedPeer),
332                make_peer("node-2", DiscoverySource::SeedPeer),
333            ],
334        );
335        let b2 = MockBackend::new(
336            "b2",
337            vec![
338                make_peer("node-2", DiscoverySource::Mdns), // duplicate id
339                make_peer("node-3", DiscoverySource::Mdns),
340            ],
341        );
342        coord.add_backend(Box::new(b1));
343        coord.add_backend(Box::new(b2));
344        coord.start_all().await.unwrap();
345
346        let peers = coord.poll_all().await;
347        // node-2 appears in both backends; should only be returned once.
348        assert_eq!(peers.len(), 3);
349        assert_eq!(coord.known_peer_count(), 3);
350        assert!(coord.is_known("node-1"));
351        assert!(coord.is_known("node-2"));
352        assert!(coord.is_known("node-3"));
353    }
354
355    #[tokio::test]
356    async fn coordinator_second_poll_empty() {
357        let mut coord = DiscoveryCoordinator::new();
358        let b = MockBackend::new("b", vec![make_peer("x", DiscoverySource::Manual)]);
359        coord.add_backend(Box::new(b));
360        coord.start_all().await.unwrap();
361
362        let first = coord.poll_all().await;
363        assert_eq!(first.len(), 1);
364
365        let second = coord.poll_all().await;
366        assert!(second.is_empty());
367    }
368
369    #[tokio::test]
370    async fn coordinator_stop_all() {
371        let mut coord = DiscoveryCoordinator::new();
372        let b = MockBackend::new("b", vec![]);
373        coord.add_backend(Box::new(b));
374        coord.start_all().await.unwrap();
375        coord.stop_all().await.unwrap();
376    }
377
378    #[test]
379    fn discovered_peer_serde_roundtrip() {
380        let peer = make_peer("serde-test", DiscoverySource::Kademlia);
381        let json = serde_json::to_string(&peer).unwrap();
382        let back: DiscoveredPeer = serde_json::from_str(&json).unwrap();
383        assert_eq!(back.node_id, "serde-test");
384        assert_eq!(back.source, DiscoverySource::Kademlia);
385    }
386
387    #[test]
388    fn mesh_peer_event_node_id() {
389        let e = MeshPeerEvent::Joined {
390            node_id: "n1".into(),
391            address: Some("127.0.0.1:9".into()),
392            platform: None,
393            verified: false,
394        };
395        assert_eq!(e.node_id(), "n1");
396        assert_eq!(
397            MeshPeerEvent::Left {
398                node_id: "n2".into()
399            }
400            .node_id(),
401            "n2"
402        );
403    }
404
405    #[tokio::test]
406    async fn peer_event_bus_delivers_to_subscriber() {
407        let bus = MeshPeerEventBus::new();
408        let mut rx = bus.subscribe();
409        bus.emit(MeshPeerEvent::Alive {
410            node_id: "p".into(),
411        });
412        let got = rx.recv().await.unwrap();
413        assert_eq!(
414            got,
415            MeshPeerEvent::Alive {
416                node_id: "p".into()
417            }
418        );
419    }
420
421    #[test]
422    fn mesh_peer_event_serde_roundtrip() {
423        let e = MeshPeerEvent::Recovered {
424            node_id: "r1".into(),
425            address: Some("10.0.0.2:9489".into()),
426            verified: true,
427        };
428        let json = serde_json::to_string(&e).unwrap();
429        let back: MeshPeerEvent = serde_json::from_str(&json).unwrap();
430        assert_eq!(back, e);
431    }
432
433    #[test]
434    fn discovery_source_serde_roundtrip() {
435        let sources = [
436            DiscoverySource::SeedPeer,
437            DiscoverySource::Mdns,
438            DiscoverySource::Kademlia,
439            DiscoverySource::PeerExchange,
440            DiscoverySource::Manual,
441        ];
442        for src in &sources {
443            let json = serde_json::to_string(src).unwrap();
444            let back: DiscoverySource = serde_json::from_str(&json).unwrap();
445            assert_eq!(&back, src);
446        }
447    }
448
449    #[test]
450    fn discovery_error_display() {
451        let e = DiscoveryError::Backend("timeout".into());
452        assert!(e.to_string().contains("timeout"));
453        let e = DiscoveryError::InvalidAddress("bad:addr".into());
454        assert!(e.to_string().contains("bad:addr"));
455        let e = DiscoveryError::NotStarted;
456        assert!(e.to_string().contains("not started"));
457    }
458}
clawft_kernel/mesh_discovery.rs

clawft_kernel/
mesh_discovery.rs
