feat(swim): implement gossip dissemination with piggyback propagation

Introduce the dissemination module (DisseminationQueue, PendingUpdate,
apply_and_disseminate) that carries membership deltas as piggyback
payloads on every outgoing probe datagram.

Each outgoing Ping, PingReq, and Ack now attaches up to max_piggyback
rumours selected from the queue. A rumour is retired after it has been
forwarded ceil(lambda * log2(n+1)) times, matching the bound from the
SWIM:GREED paper (Das et al. §4.3) that guarantees with high probability
every live member receives each delta.

Inbound piggyback is ingested before dispatching the message so that a
self-refutation incarnation bump is reflected in the outgoing Ack of the
same round-trip. SwimConfig gains max_piggyback and fanout_lambda fields
with validation; ProbeRound and DetectorRunner are wired to pass them
through. Integration tests cover cross-node delta propagation and
self-refutation via piggyback.

Author: farhan-syah

nodedb-cluster/src/swim/config.rs

@@ -39,6 +39,17 @@ pub struct SwimConfig {
     /// Seed incarnation for a freshly-booted local node. Always `0` in
     /// production; exposed for deterministic unit tests.
     pub initial_incarnation: Incarnation,
+
+    /// Maximum number of membership deltas to piggyback on a single
+    /// outgoing SWIM datagram. Caps per-message bandwidth and bounds
+    /// the encoded payload size below a UDP MTU.
+    pub max_piggyback: usize,
+
+    /// Gossip fanout multiplier (`lambda` in Das §4.3). The
+    /// dissemination queue drops a rumour after it has been carried
+    /// on `ceil(fanout_lambda * log2(n+1))` outgoing messages, which
+    /// with high probability reaches every member.
+    pub fanout_lambda: u32,
 }
 
 impl SwimConfig {
@@ -51,6 +62,8 @@ impl SwimConfig {
             suspicion_mult: 4,
             min_suspicion: Duration::from_secs(2),
             initial_incarnation: Incarnation::ZERO,
+            max_piggyback: 6,
+            fanout_lambda: 3,
         }
     }
 
@@ -88,6 +101,18 @@ impl SwimConfig {
                 reason: "must be non-zero",
             });
         }
+        if self.max_piggyback == 0 {
+            return Err(SwimError::InvalidConfig {
+                field: "max_piggyback",
+                reason: "must be at least 1",
+            });
+        }
+        if self.fanout_lambda == 0 {
+            return Err(SwimError::InvalidConfig {
+                field: "fanout_lambda",
+                reason: "must be at least 1",
+            });
+        }
         Ok(())
     }
 }
@@ -171,4 +196,30 @@ mod tests {
             })
         ));
     }
+
+    #[test]
+    fn zero_max_piggyback_rejected() {
+        let mut cfg = SwimConfig::production();
+        cfg.max_piggyback = 0;
+        assert!(matches!(
+            cfg.validate(),
+            Err(SwimError::InvalidConfig {
+                field: "max_piggyback",
+                ..
+            })
+        ));
+    }
+
+    #[test]
+    fn zero_fanout_lambda_rejected() {
+        let mut cfg = SwimConfig::production();
+        cfg.fanout_lambda = 0;
+        assert!(matches!(
+            cfg.validate(),
+            Err(SwimError::InvalidConfig {
+                field: "fanout_lambda",
+                ..
+            })
+        ));
+    }
 }

nodedb-cluster/src/swim/detector/probe_round.rs

@@ -28,6 +28,7 @@ use crate::swim::wire::{Ping, PingReq, ProbeId, SwimMessage};
 
 use super::scheduler::ProbeScheduler;
 use super::transport::Transport;
+use crate::swim::dissemination::DisseminationQueue;
 use crate::swim::membership::MembershipList;
 
 /// Upper bound on concurrent inflight probes. The detector only issues a
@@ -98,8 +99,11 @@ pub struct ProbeRound<'a, F: Fn() -> ProbeId> {
     pub membership: &'a MembershipList,
     pub transport: &'a Arc<dyn Transport>,
     pub inflight: &'a Arc<InflightProbes>,
+    pub dissemination: &'a Arc<DisseminationQueue>,
     pub probe_timeout: Duration,
     pub k_indirect: usize,
+    pub max_piggyback: usize,
+    pub fanout_lambda: u32,
     pub next_probe_id: F,
     pub local_incarnation: Incarnation,
 }
@@ -113,12 +117,17 @@ impl<'a, F: Fn() -> ProbeId> ProbeRound<'a, F> {
             membership,
             transport,
             inflight,
+            dissemination,
             probe_timeout,
             k_indirect,
+            max_piggyback,
+            fanout_lambda,
             next_probe_id,
             local_incarnation,
         } = self;
 
+        let fanout = DisseminationQueue::fanout_threshold(membership.len(), fanout_lambda);
+
         let Some((target_id, target_addr)) = scheduler.next_target(membership) else {
             return Ok(ProbeOutcome::Idle);
         };
@@ -134,7 +143,7 @@ impl<'a, F: Fn() -> ProbeId> ProbeRound<'a, F> {
                     probe_id: direct_id,
                     from: local.clone(),
                     incarnation: local_incarnation,
-                    piggyback: vec![],
+                    piggyback: dissemination.take_for_message(max_piggyback, fanout),
                 }),
             )
             .await?;
@@ -170,7 +179,7 @@ impl<'a, F: Fn() -> ProbeId> ProbeRound<'a, F> {
                         from: local.clone(),
                         target: target_id.clone(),
                         target_addr: target_addr.to_string(),
-                        piggyback: vec![],
+                        piggyback: dissemination.take_for_message(max_piggyback, fanout),
                     }),
                 )
                 .await?;
@@ -251,6 +260,8 @@ mod tests {
             suspicion_mult: 4,
             min_suspicion: Duration::from_millis(500),
             initial_incarnation: Incarnation::ZERO,
+            max_piggyback: 6,
+            fanout_lambda: 3,
         }
     }
 
@@ -287,13 +298,17 @@ mod tests {
         let list = membership_with_peers("local", 7000, &[]).await;
         let mut sched = ProbeScheduler::with_seed(1);
         let inflight = Arc::new(InflightProbes::new());
+        let dissemination = Arc::new(DisseminationQueue::new());
         let outcome = ProbeRound {
             scheduler: &mut sched,
             membership: &list,
             transport: &local,
             inflight: &inflight,
+            dissemination: &dissemination,
             probe_timeout: cfg().probe_timeout,
             k_indirect: 2,
+            max_piggyback: 6,
+            fanout_lambda: 3,
             next_probe_id: pid_gen(1),
             local_incarnation: Incarnation::ZERO,
         }
@@ -314,13 +329,17 @@ mod tests {
         let list = membership_with_peers("local", 7000, &[("n1", 7001, MemberState::Alive)]).await;
         let mut sched = ProbeScheduler::with_seed(1);
         let inflight = Arc::new(InflightProbes::new());
+        let dissemination = Arc::new(DisseminationQueue::new());
         let outcome = ProbeRound {
             scheduler: &mut sched,
             membership: &list,
             transport: &local,
             inflight: &inflight,
+            dissemination: &dissemination,
             probe_timeout: cfg().probe_timeout,
             k_indirect: 2,
+            max_piggyback: 6,
+            fanout_lambda: 3,
             next_probe_id: pid_gen(1),
             local_incarnation: Incarnation::ZERO,
         }
@@ -367,13 +386,17 @@ mod tests {
             }
         });
 
+        let dissemination = Arc::new(DisseminationQueue::new());
         let outcome = ProbeRound {
             scheduler: &mut sched,
             membership: &list,
             transport: &local,
             inflight: &inflight,
+            dissemination: &dissemination,
             probe_timeout: cfg().probe_timeout,
             k_indirect: 2,
+            max_piggyback: 6,
+            fanout_lambda: 3,
             next_probe_id: pid_gen(1),
             local_incarnation: Incarnation::ZERO,
         }
@@ -436,13 +459,17 @@ mod tests {
             }
         });
 
+        let dissemination = Arc::new(DisseminationQueue::new());
         let outcome = ProbeRound {
             scheduler: &mut sched,
             membership: &list,
             transport: &local,
             inflight: &inflight,
+            dissemination: &dissemination,
             probe_timeout: cfg().probe_timeout,
             k_indirect: 2,
+            max_piggyback: 6,
+            fanout_lambda: 3,
             next_probe_id: pid_gen(1),
             local_incarnation: Incarnation::ZERO,
         }
@@ -468,5 +495,4 @@ mod tests {
             .expect_err("full");
         assert!(matches!(err, SwimError::ProbeInflightOverflow));
     }
-
 }