feat(cluster): add SWIM failure detection and membership protocol

Implement a SWIM (Scalable Weakly-consistent Infection-style Membership)
failure detector for the cluster layer. The protocol provides probabilistic
membership convergence with O(log n) dissemination, replacing the need for
a central membership oracle.

Includes:
- Incarnation counter for detecting stale membership state after restarts
- Member records and state machine (Alive/Suspect/Dead/Left)
- MembershipList with merge semantics for gossip convergence
- Wire message format and QUIC-framed codec for probe/ack/indirect-probe
- Configuration surface (probe interval, suspicion timeout, fanout)
- NodeId MessagePack derive to support membership wire encoding

Author: farhan-syah

nodedb-cluster/src/lib.rs

@@ -31,6 +31,7 @@ pub mod rebalance_scheduler;
 pub mod routing;
 pub mod rpc_codec;
 pub mod shard_split;
+pub mod swim;
 pub mod topology;
 pub mod transport;
 pub mod vshard_handler;
@@ -77,3 +78,4 @@ pub use lifecycle::{
 pub use rdma_transport::{RdmaConfig, RdmaTransport};
 pub use rebalance_scheduler::{NodeMetrics, RebalanceScheduler, RebalanceTrigger, SchedulerConfig};
 pub use shard_split::{SplitPlan, SplitStrategy, plan_graph_split, plan_vector_split};
+pub use swim::{Incarnation, Member, MemberState, MembershipList, SwimConfig, SwimError};

nodedb-cluster/src/swim/config.rs

@@ -0,0 +1,174 @@
+//! SWIM protocol configuration.
+//!
+//! Tunable parameters that govern failure-detection latency, bandwidth, and
+//! false-positive rate. Defaults follow the Lifeguard recommendations for
+//! a ≤ 256-node cluster and are safe for production without tuning.
+
+use std::time::Duration;
+
+use super::error::SwimError;
+use super::incarnation::Incarnation;
+
+/// Configuration for the SWIM failure detector.
+///
+/// All fields are validated at construction time via [`SwimConfig::validate`];
+/// an invalid config is a programmer error and returns a typed
+/// [`SwimError::InvalidConfig`] rather than panicking.
+#[derive(Debug, Clone)]
+pub struct SwimConfig {
+    /// Time between probe rounds (T' in the SWIM paper). One randomly-chosen
+    /// alive peer is pinged per interval.
+    pub probe_interval: Duration,
+
+    /// Round-trip deadline for a direct ping before falling back to k
+    /// indirect pings. Must be strictly less than `probe_interval`.
+    pub probe_timeout: Duration,
+
+    /// Number of indirect probe helpers (`k` in the paper).
+    pub indirect_probes: u8,
+
+    /// Multiplier on `probe_interval` used to compute the suspicion timeout
+    /// before a `Suspect` member is declared `Dead`. Lifeguard §3.1.
+    pub suspicion_mult: u8,
+
+    /// Minimum value for the suspicion timeout; protects small clusters from
+    /// sub-second suspicion windows. The effective timeout is
+    /// `max(min_suspicion, suspicion_mult * log2(n) * probe_interval)`.
+    pub min_suspicion: Duration,
+
+    /// Seed incarnation for a freshly-booted local node. Always `0` in
+    /// production; exposed for deterministic unit tests.
+    pub initial_incarnation: Incarnation,
+}
+
+impl SwimConfig {
+    /// Production defaults from Lifeguard, tuned for a ≤ 256-node cluster.
+    pub fn production() -> Self {
+        Self {
+            probe_interval: Duration::from_millis(1000),
+            probe_timeout: Duration::from_millis(500),
+            indirect_probes: 3,
+            suspicion_mult: 4,
+            min_suspicion: Duration::from_secs(2),
+            initial_incarnation: Incarnation::ZERO,
+        }
+    }
+
+    /// Validate the configuration. Returns `InvalidConfig` if any invariant
+    /// fails. Callers should treat validation failure as a fatal startup
+    /// error — SWIM cannot run with incoherent timing parameters.
+    pub fn validate(&self) -> Result<(), SwimError> {
+        if self.probe_interval.is_zero() {
+            return Err(SwimError::InvalidConfig {
+                field: "probe_interval",
+                reason: "must be non-zero",
+            });
+        }
+        if self.probe_timeout >= self.probe_interval {
+            return Err(SwimError::InvalidConfig {
+                field: "probe_timeout",
+                reason: "must be strictly less than probe_interval",
+            });
+        }
+        if self.indirect_probes == 0 {
+            return Err(SwimError::InvalidConfig {
+                field: "indirect_probes",
+                reason: "must be at least 1",
+            });
+        }
+        if self.suspicion_mult == 0 {
+            return Err(SwimError::InvalidConfig {
+                field: "suspicion_mult",
+                reason: "must be at least 1",
+            });
+        }
+        if self.min_suspicion.is_zero() {
+            return Err(SwimError::InvalidConfig {
+                field: "min_suspicion",
+                reason: "must be non-zero",
+            });
+        }
+        Ok(())
+    }
+}
+
+impl Default for SwimConfig {
+    fn default() -> Self {
+        Self::production()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn production_defaults_are_valid() {
+        SwimConfig::production().validate().expect("valid");
+    }
+
+    #[test]
+    fn zero_probe_interval_rejected() {
+        let mut cfg = SwimConfig::production();
+        cfg.probe_interval = Duration::ZERO;
+        assert!(matches!(
+            cfg.validate(),
+            Err(SwimError::InvalidConfig {
+                field: "probe_interval",
+                ..
+            })
+        ));
+    }
+
+    #[test]
+    fn probe_timeout_must_be_less_than_interval() {
+        let mut cfg = SwimConfig::production();
+        cfg.probe_timeout = cfg.probe_interval;
+        assert!(matches!(
+            cfg.validate(),
+            Err(SwimError::InvalidConfig {
+                field: "probe_timeout",
+                ..
+            })
+        ));
+    }
+
+    #[test]
+    fn zero_indirect_probes_rejected() {
+        let mut cfg = SwimConfig::production();
+        cfg.indirect_probes = 0;
+        assert!(matches!(
+            cfg.validate(),
+            Err(SwimError::InvalidConfig {
+                field: "indirect_probes",
+                ..
+            })
+        ));
+    }
+
+    #[test]
+    fn zero_suspicion_mult_rejected() {
+        let mut cfg = SwimConfig::production();
+        cfg.suspicion_mult = 0;
+        assert!(matches!(
+            cfg.validate(),
+            Err(SwimError::InvalidConfig {
+                field: "suspicion_mult",
+                ..
+            })
+        ));
+    }
+
+    #[test]
+    fn zero_min_suspicion_rejected() {
+        let mut cfg = SwimConfig::production();
+        cfg.min_suspicion = Duration::ZERO;
+        assert!(matches!(
+            cfg.validate(),
+            Err(SwimError::InvalidConfig {
+                field: "min_suspicion",
+                ..
+            })
+        ));
+    }
+}

nodedb-cluster/src/swim/error.rs

@@ -0,0 +1,105 @@
+//! Typed error variants for the SWIM subsystem.
+//!
+//! `SwimError` is the single error type returned by every public function
+//! in `nodedb_cluster::swim`. It is wired into the cluster-wide
+//! [`ClusterError`] enum via a `From` impl in `crate::error`, which in turn
+//! bridges to `nodedb_types::NodeDbError` at the public API boundary.
+
+use thiserror::Error;
+
+use nodedb_types::NodeId;
+
+use super::incarnation::Incarnation;
+use super::member::MemberState;
+
+/// Errors produced by the SWIM failure detector and membership layer.
+#[derive(Debug, Error)]
+pub enum SwimError {
+    /// A message or update referenced a node id not present in the
+    /// membership list. This is non-fatal — the detector will request a
+    /// full sync from the sender.
+    #[error("swim: unknown member {node_id}")]
+    UnknownMember { node_id: NodeId },
+
+    /// Received update carries an incarnation strictly older than the
+    /// locally recorded value, so the update is refuted.
+    #[error("swim: stale incarnation for {node_id}: received {received:?} <= local {local:?}")]
+    StaleIncarnation {
+        node_id: NodeId,
+        received: Incarnation,
+        local: Incarnation,
+    },
+
+    /// Received a `Suspect` update targeting the local node. The failure
+    /// detector must bump its own incarnation and broadcast an `Alive`
+    /// refutation. Callers treat this as a signal, not a fatal error.
+    #[error("swim: local node suspected at incarnation {incarnation:?}")]
+    SelfSuspected { incarnation: Incarnation },
+
+    /// A state transition violated the SWIM state machine (e.g. attempting
+    /// to move a `Left` member back to `Alive`). Always a bug.
+    #[error("swim: invalid state transition {from:?} -> {to:?}")]
+    InvalidTransition { from: MemberState, to: MemberState },
+
+    /// Configuration validation failed. Returned by [`super::SwimConfig::validate`].
+    #[error("swim: invalid config field {field}: {reason}")]
+    InvalidConfig {
+        field: &'static str,
+        reason: &'static str,
+    },
+
+    /// zerompk failed to serialize a `SwimMessage`. In practice this is
+    /// infallible for the current message schema — the variant exists so
+    /// future additions to the wire format cannot silently panic.
+    #[error("swim: encode failure: {detail}")]
+    Encode { detail: String },
+
+    /// zerompk failed to parse incoming bytes as a `SwimMessage`. Common
+    /// causes: truncated datagram, version skew, random UDP noise.
+    #[error("swim: decode failure: {detail}")]
+    Decode { detail: String },
+}
+
+impl From<SwimError> for crate::error::ClusterError {
+    fn from(err: SwimError) -> Self {
+        crate::error::ClusterError::Transport {
+            detail: err.to_string(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn display_contains_context() {
+        let err = SwimError::StaleIncarnation {
+            node_id: NodeId::new("n1"),
+            received: Incarnation::new(3),
+            local: Incarnation::new(5),
+        };
+        let msg = err.to_string();
+        assert!(msg.contains("n1"));
+        assert!(msg.contains('3'));
+        assert!(msg.contains('5'));
+    }
+
+    #[test]
+    fn invalid_config_display() {
+        let err = SwimError::InvalidConfig {
+            field: "probe_timeout",
+            reason: "must be strictly less than probe_interval",
+        };
+        assert!(err.to_string().contains("probe_timeout"));
+    }
+
+    #[test]
+    fn bridges_to_cluster_error() {
+        let err: crate::error::ClusterError = SwimError::UnknownMember {
+            node_id: NodeId::new("n42"),
+        }
+        .into();
+        assert!(matches!(err, crate::error::ClusterError::Transport { .. }));
+    }
+}