laminar_core/cluster/discovery/
mod.rs

1//! Peer discovery: `StaticDiscovery` (seed list) and `GossipDiscovery`
2//! (chitchat).
3#![allow(clippy::disallowed_types)] // cold path: discovery metadata (serde + rkyv)
4
5mod static_discovery;
6pub use static_discovery::{StaticDiscovery, StaticDiscoveryConfig};
7
8mod gossip_discovery;
9pub use gossip_discovery::{keys, GossipDiscovery, GossipDiscoveryConfig};
10
11use std::collections::HashMap;
12use std::fmt;
13
14use serde::{Deserialize, Serialize};
15use tokio::sync::watch;
16
17pub use crate::state::NodeId;
18
19/// Publish the membership watch only on a real change. Unconditional `send` each
20/// discovery tick starves consumers that debounce on `changed()` for a quiet period
21/// (the rebalance controller, so a dead node's vnodes are never shed).
22pub(crate) fn publish_if_changed(tx: &watch::Sender<Vec<NodeInfo>>, mut peer_list: Vec<NodeInfo>) {
23    peer_list.sort_by_key(|n| n.id.0); // peer map order is arbitrary; equality must not depend on it
24    tx.send_if_modified(|cur| {
25        // Ignore `last_heartbeat_ms` — it ticks every refresh and would notify forever.
26        fn same_member(a: &NodeInfo, b: &NodeInfo) -> bool {
27            // Destructure so a new NodeInfo field forces an update here; compare
28            // all fields except last_heartbeat_ms (it ticks every refresh).
29            let NodeInfo {
30                id,
31                name,
32                rpc_address,
33                raft_address,
34                state,
35                metadata,
36                last_heartbeat_ms: _,
37            } = a;
38            id == &b.id
39                && name == &b.name
40                && rpc_address == &b.rpc_address
41                && raft_address == &b.raft_address
42                && state == &b.state
43                && metadata == &b.metadata
44        }
45        let same = cur.len() == peer_list.len()
46            && cur.iter().zip(&peer_list).all(|(a, b)| same_member(a, b));
47        if same {
48            false
49        } else {
50            *cur = peer_list;
51            true
52        }
53    });
54}
55
56/// Current lifecycle state of a node.
57#[derive(
58    Debug,
59    Clone,
60    Copy,
61    PartialEq,
62    Eq,
63    Serialize,
64    Deserialize,
65    rkyv::Archive,
66    rkyv::Serialize,
67    rkyv::Deserialize,
68)]
69pub enum NodeState {
70    /// Node is joining the cluster but not yet fully active.
71    Joining,
72    /// Node is active and participating in the cluster.
73    Active,
74    /// Node is suspected of failure (missed heartbeats).
75    Suspected,
76    /// Node is gracefully draining before leaving.
77    Draining,
78    /// Node has left the cluster.
79    Left,
80}
81
82impl fmt::Display for NodeState {
83    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
84        match self {
85            Self::Joining => write!(f, "joining"),
86            Self::Active => write!(f, "active"),
87            Self::Suspected => write!(f, "suspected"),
88            Self::Draining => write!(f, "draining"),
89            Self::Left => write!(f, "left"),
90        }
91    }
92}
93
94/// Hardware and deployment metadata for a node.
95#[derive(
96    Debug,
97    Clone,
98    PartialEq,
99    Serialize,
100    Deserialize,
101    rkyv::Archive,
102    rkyv::Serialize,
103    rkyv::Deserialize,
104)]
105pub struct NodeMetadata {
106    /// Number of CPU cores available.
107    pub cores: u32,
108    /// Total memory in bytes.
109    pub memory_bytes: u64,
110    /// Failure domain (e.g., rack, zone, region).
111    pub failure_domain: Option<String>,
112    /// Arbitrary key-value tags.
113    pub tags: HashMap<String, String>,
114    /// `LaminarDB` version string.
115    pub version: String,
116}
117
118impl Default for NodeMetadata {
119    fn default() -> Self {
120        Self {
121            cores: 1,
122            memory_bytes: 0,
123            failure_domain: None,
124            tags: HashMap::new(),
125            version: String::new(),
126        }
127    }
128}
129
130/// Full information about a discovered node.
131#[derive(
132    Debug,
133    Clone,
134    PartialEq,
135    Serialize,
136    Deserialize,
137    rkyv::Archive,
138    rkyv::Serialize,
139    rkyv::Deserialize,
140)]
141pub struct NodeInfo {
142    /// The node's unique identifier.
143    pub id: NodeId,
144    /// Human-readable name.
145    pub name: String,
146    /// Address for gRPC communication.
147    pub rpc_address: String,
148    /// Legacy wire-schema field; current runtimes publish it empty and do not bind a Raft service.
149    pub raft_address: String,
150    /// Current lifecycle state.
151    pub state: NodeState,
152    /// Hardware/deployment metadata.
153    pub metadata: NodeMetadata,
154    /// Timestamp of the last received heartbeat (millis since epoch).
155    pub last_heartbeat_ms: i64,
156}
157
158/// Node ids eligible to own vnodes: only `Active` nodes. Joining,
159/// Suspected, Draining, and Left are excluded so a draining or failing
160/// node sheds its vnodes on the next rotation.
161#[must_use]
162pub fn assignable_node_ids(members: &[NodeInfo]) -> Vec<NodeId> {
163    let mut ids: Vec<NodeId> = members
164        .iter()
165        .filter(|m| matches!(m.state, NodeState::Active))
166        .map(|m| m.id)
167        .filter(|id| !id.is_unassigned())
168        .collect();
169    ids.sort_unstable();
170    ids.dedup();
171    ids
172}
173
174/// A membership change event.
175#[derive(Debug, Clone)]
176pub enum MembershipEvent {
177    /// A new node has been discovered.
178    NodeJoined(Box<NodeInfo>),
179    /// A node's state has changed.
180    NodeStateChanged {
181        /// The node whose state changed.
182        node_id: NodeId,
183        /// Previous state.
184        old_state: NodeState,
185        /// New state.
186        new_state: NodeState,
187    },
188    /// A node has left or been removed from the cluster.
189    NodeLeft(NodeId),
190}
191
192/// Errors that can occur during discovery operations.
193#[derive(Debug, thiserror::Error)]
194pub enum DiscoveryError {
195    /// Failed to bind to the specified address.
196    #[error("bind error: {0}")]
197    Bind(String),
198
199    /// Failed to connect to a seed/peer node.
200    #[error("connection error to {address}: {reason}")]
201    Connection {
202        /// The address that failed.
203        address: String,
204        /// Reason for failure.
205        reason: String,
206    },
207
208    /// Serialization/deserialization failure.
209    #[error("serialization error: {0}")]
210    Serialization(String),
211
212    /// The discovery service is not running.
213    #[error("discovery not started")]
214    NotStarted,
215
216    /// The discovery service has been shut down.
217    #[error("discovery shut down")]
218    ShutDown,
219
220    /// An I/O error occurred.
221    #[error("I/O error: {0}")]
222    Io(#[from] std::io::Error),
223}
224
225/// Trait for node discovery in a cluster.
226///
227/// Implementations provide the mechanism by which nodes find and
228/// track each other. The trait is async and designed for long-running
229/// background tasks.
230#[allow(async_fn_in_trait)]
231pub trait Discovery: Send + Sync + 'static {
232    /// Start the discovery service.
233    ///
234    /// This spawns background tasks for heartbeating and failure detection.
235    async fn start(&mut self) -> Result<(), DiscoveryError>;
236
237    /// Get the current set of known peers (excluding self).
238    async fn peers(&self) -> Result<Vec<NodeInfo>, DiscoveryError>;
239
240    /// Announce this node's updated information to the cluster.
241    async fn announce(&self, info: NodeInfo) -> Result<(), DiscoveryError>;
242
243    /// Subscribe to membership change events.
244    ///
245    /// Returns a watch receiver that is updated whenever the membership
246    /// changes. The value is the list of all known peers.
247    fn membership_watch(&self) -> watch::Receiver<Vec<NodeInfo>>;
248
249    /// Gracefully stop the discovery service.
250    async fn stop(&mut self) -> Result<(), DiscoveryError>;
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256
257    #[test]
258    fn test_node_id_display() {
259        assert_eq!(NodeId(42).to_string(), "node-42");
260    }
261
262    #[test]
263    fn test_node_id_unassigned() {
264        assert!(NodeId::UNASSIGNED.is_unassigned());
265        assert!(!NodeId(1).is_unassigned());
266    }
267
268    #[test]
269    fn test_node_state_display() {
270        assert_eq!(NodeState::Active.to_string(), "active");
271        assert_eq!(NodeState::Suspected.to_string(), "suspected");
272        assert_eq!(NodeState::Draining.to_string(), "draining");
273    }
274
275    fn info_with(id: u64, state: NodeState) -> NodeInfo {
276        NodeInfo {
277            id: NodeId(id),
278            name: format!("n{id}"),
279            rpc_address: String::new(),
280            raft_address: String::new(),
281            state,
282            metadata: NodeMetadata::default(),
283            last_heartbeat_ms: 0,
284        }
285    }
286
287    #[test]
288    fn assignable_includes_only_active_sorted_deduped() {
289        let members = vec![
290            info_with(5, NodeState::Active),
291            info_with(2, NodeState::Joining),
292            info_with(3, NodeState::Suspected),
293            info_with(4, NodeState::Draining),
294            info_with(6, NodeState::Left),
295            info_with(1, NodeState::Active),
296            info_with(1, NodeState::Active), // dup
297        ];
298        assert_eq!(assignable_node_ids(&members), vec![NodeId(1), NodeId(5)]);
299    }
300
301    #[test]
302    fn assignable_drops_unassigned() {
303        let mut unassigned = info_with(7, NodeState::Active);
304        unassigned.id = NodeId::UNASSIGNED;
305        let members = vec![unassigned, info_with(7, NodeState::Active)];
306        assert_eq!(assignable_node_ids(&members), vec![NodeId(7)]);
307    }
308
309    #[test]
310    fn test_node_metadata_default() {
311        let meta = NodeMetadata::default();
312        assert_eq!(meta.cores, 1);
313        assert_eq!(meta.memory_bytes, 0);
314        assert!(meta.failure_domain.is_none());
315        assert!(meta.tags.is_empty());
316    }
317
318    #[test]
319    fn test_node_id_serialization() {
320        let id = NodeId(123);
321        let json = serde_json::to_string(&id).unwrap();
322        let back: NodeId = serde_json::from_str(&json).unwrap();
323        assert_eq!(id, back);
324    }
325
326    #[test]
327    fn test_node_info_serialization() {
328        let info = NodeInfo {
329            id: NodeId(1),
330            name: "test-node".into(),
331            rpc_address: "127.0.0.1:9000".into(),
332            raft_address: "127.0.0.1:9001".into(),
333            state: NodeState::Active,
334            metadata: NodeMetadata::default(),
335            last_heartbeat_ms: 1000,
336        };
337        let json = serde_json::to_string(&info).unwrap();
338        let back: NodeInfo = serde_json::from_str(&json).unwrap();
339        assert_eq!(back.id, info.id);
340        assert_eq!(back.name, "test-node");
341    }
342}
laminar_core/cluster/discovery/mod.rs

laminar_core/cluster/discovery/
mod.rs
