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                state,
34                metadata,
35                last_heartbeat_ms: _,
36            } = a;
37            id == &b.id
38                && name == &b.name
39                && rpc_address == &b.rpc_address
40                && state == &b.state
41                && metadata == &b.metadata
42        }
43        let same = cur.len() == peer_list.len()
44            && cur.iter().zip(&peer_list).all(|(a, b)| same_member(a, b));
45        if same {
46            false
47        } else {
48            *cur = peer_list;
49            true
50        }
51    });
52}
53
54/// Current lifecycle state of a node.
55#[derive(
56    Debug,
57    Clone,
58    Copy,
59    PartialEq,
60    Eq,
61    Serialize,
62    Deserialize,
63    rkyv::Archive,
64    rkyv::Serialize,
65    rkyv::Deserialize,
66)]
67pub enum NodeState {
68    /// Node is joining the cluster but not yet fully active.
69    Joining,
70    /// Node is active and participating in the cluster.
71    Active,
72    /// Node is suspected of failure (missed heartbeats).
73    Suspected,
74    /// Node is gracefully draining before leaving.
75    Draining,
76    /// Node has left the cluster.
77    Left,
78}
79
80impl fmt::Display for NodeState {
81    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
82        match self {
83            Self::Joining => write!(f, "joining"),
84            Self::Active => write!(f, "active"),
85            Self::Suspected => write!(f, "suspected"),
86            Self::Draining => write!(f, "draining"),
87            Self::Left => write!(f, "left"),
88        }
89    }
90}
91
92/// Hardware and deployment metadata for a node.
93#[derive(
94    Debug,
95    Clone,
96    PartialEq,
97    Serialize,
98    Deserialize,
99    rkyv::Archive,
100    rkyv::Serialize,
101    rkyv::Deserialize,
102)]
103pub struct NodeMetadata {
104    /// Number of CPU cores available.
105    pub cores: u32,
106    /// Total memory in bytes.
107    pub memory_bytes: u64,
108    /// Failure domain (e.g., rack, zone, region).
109    pub failure_domain: Option<String>,
110    /// Arbitrary key-value tags.
111    pub tags: HashMap<String, String>,
112    /// `LaminarDB` version string.
113    pub version: String,
114}
115
116impl Default for NodeMetadata {
117    fn default() -> Self {
118        Self {
119            cores: 1,
120            memory_bytes: 0,
121            failure_domain: None,
122            tags: HashMap::new(),
123            version: String::new(),
124        }
125    }
126}
127
128/// Full information about a discovered node.
129#[derive(
130    Debug,
131    Clone,
132    PartialEq,
133    Serialize,
134    Deserialize,
135    rkyv::Archive,
136    rkyv::Serialize,
137    rkyv::Deserialize,
138)]
139pub struct NodeInfo {
140    /// The node's unique identifier.
141    pub id: NodeId,
142    /// Human-readable name.
143    pub name: String,
144    /// Address for gRPC communication.
145    pub rpc_address: String,
146    /// Current lifecycle state.
147    pub state: NodeState,
148    /// Hardware/deployment metadata.
149    pub metadata: NodeMetadata,
150    /// Timestamp of the last received heartbeat (millis since epoch).
151    pub last_heartbeat_ms: i64,
152}
153
154/// Node ids eligible to own vnodes: only `Active` nodes. Joining,
155/// Suspected, Draining, and Left are excluded so a draining or failing
156/// node sheds its vnodes on the next rotation.
157#[must_use]
158pub fn assignable_node_ids(members: &[NodeInfo]) -> Vec<NodeId> {
159    let mut ids: Vec<NodeId> = members
160        .iter()
161        .filter(|m| matches!(m.state, NodeState::Active))
162        .map(|m| m.id)
163        .filter(|id| !id.is_unassigned())
164        .collect();
165    ids.sort_unstable();
166    ids.dedup();
167    ids
168}
169
170/// A membership change event.
171#[derive(Debug, Clone)]
172pub enum MembershipEvent {
173    /// A new node has been discovered.
174    NodeJoined(Box<NodeInfo>),
175    /// A node's state has changed.
176    NodeStateChanged {
177        /// The node whose state changed.
178        node_id: NodeId,
179        /// Previous state.
180        old_state: NodeState,
181        /// New state.
182        new_state: NodeState,
183    },
184    /// A node has left or been removed from the cluster.
185    NodeLeft(NodeId),
186}
187
188/// Errors that can occur during discovery operations.
189#[derive(Debug, thiserror::Error)]
190pub enum DiscoveryError {
191    /// Failed to bind to the specified address.
192    #[error("bind error: {0}")]
193    Bind(String),
194
195    /// Failed to connect to a seed/peer node.
196    #[error("connection error to {address}: {reason}")]
197    Connection {
198        /// The address that failed.
199        address: String,
200        /// Reason for failure.
201        reason: String,
202    },
203
204    /// Serialization/deserialization failure.
205    #[error("serialization error: {0}")]
206    Serialization(String),
207
208    /// The discovery service is not running.
209    #[error("discovery not started")]
210    NotStarted,
211
212    /// The discovery service has been shut down.
213    #[error("discovery shut down")]
214    ShutDown,
215
216    /// An I/O error occurred.
217    #[error("I/O error: {0}")]
218    Io(#[from] std::io::Error),
219}
220
221/// Trait for node discovery in a cluster.
222///
223/// Implementations provide the mechanism by which nodes find and
224/// track each other. The trait is async and designed for long-running
225/// background tasks.
226#[allow(async_fn_in_trait)]
227pub trait Discovery: Send + Sync + 'static {
228    /// Start the discovery service.
229    ///
230    /// This spawns background tasks for heartbeating and failure detection.
231    async fn start(&mut self) -> Result<(), DiscoveryError>;
232
233    /// Get the current set of known peers (excluding self).
234    async fn peers(&self) -> Result<Vec<NodeInfo>, DiscoveryError>;
235
236    /// Announce this node's updated information to the cluster.
237    async fn announce(&self, info: NodeInfo) -> Result<(), DiscoveryError>;
238
239    /// Subscribe to membership change events.
240    ///
241    /// Returns a watch receiver that is updated whenever the membership
242    /// changes. The value is the list of all known peers.
243    fn membership_watch(&self) -> watch::Receiver<Vec<NodeInfo>>;
244
245    /// Gracefully stop the discovery service.
246    async fn stop(&mut self) -> Result<(), DiscoveryError>;
247}
248
249#[cfg(test)]
250mod tests;
laminar_core/cluster/discovery/mod.rs

laminar_core/cluster/discovery/
mod.rs
