clawft_kernel/
mesh_liveness.rs

1//! Mesh liveness and round-trip time: a ping/pong between verified peers.
2//!
3//! Each node pings every verified peer on the [`PING_TOPIC`] control topic
4//! every [`LivenessConfig::interval`]. A pong only counts when it arrives on
5//! the verified connection of the peer that was pinged and echoes a nonce
6//! this node actually sent and has not yet seen answered. A counted pong
7//! feeds the heartbeat tracker ([`MeshRuntime::record_heartbeat`], which
8//! drives `Alive`/`Recovered` and cluster `last_heartbeat`) and a smoothed
9//! RTT. A pong may carry the responder's [`LoadSample`] (fleet P3); old peers
10//! send none. Unverified peers are never pinged, and their pings and pongs
11//! are ignored. Everything here is in memory; nothing is persisted.
12
13use std::collections::HashMap;
14use std::sync::{Arc, Mutex, Weak};
15use std::time::{Duration, Instant, SystemTime};
16
17use serde_json::{Value, json};
18
19use crate::ipc::{KernelMessage, MessagePayload, MessageTarget};
20use crate::mesh_delivery::PeerCtx;
21use crate::mesh_load::LoadSample;
22use crate::mesh_runtime::{MeshRuntime, PING_TOPIC, PeerControlSink};
23
24/// Tunables.
25#[derive(Debug, Clone, Copy)]
26pub struct LivenessConfig {
27    /// Time between ping rounds.
28    pub interval: Duration,
29    /// An unanswered ping older than this is forgotten (and counted as a miss).
30    pub timeout: Duration,
31}
32
33impl Default for LivenessConfig {
34    fn default() -> Self {
35        Self { interval: Duration::from_secs(10), timeout: Duration::from_secs(30) }
36    }
37}
38
39/// What this node knows about one peer's liveness.
40#[derive(Debug, Clone, PartialEq)]
41pub struct PeerLiveness {
42    /// Wall-clock time of the last counted pong.
43    pub last_seen: SystemTime,
44    /// Smoothed round-trip time (EWMA, alpha 0.2), milliseconds.
45    pub rtt_ms: f64,
46    /// Pings sent that timed out since the last counted pong.
47    pub missed: u32,
48    /// The load the peer attached to its last counted pong (it says so;
49    /// not signed). `None` from a peer that sends none.
50    pub load: Option<LoadSample>,
51}
52
53/// At most this many pings in flight per peer; a slow peer cannot grow the table.
54const MAX_OUTSTANDING_PER_PEER: usize = 4;
55
56#[derive(Default)]
57struct State {
58    seq: u64,
59    /// (peer, nonce) -> when the ping was sent.
60    outstanding: HashMap<(String, u64), Instant>,
61    peers: HashMap<String, PeerLiveness>,
62}
63
64/// The liveness service for one [`MeshRuntime`].
65pub struct Liveness {
66    rt: Weak<MeshRuntime>,
67    cfg: LivenessConfig,
68    state: Mutex<State>,
69}
70
71impl Liveness {
72    /// Build without starting the ping loop (tests drive [`Liveness::tick`]).
73    pub fn new(rt: &Arc<MeshRuntime>, cfg: LivenessConfig) -> Arc<Self> {
74        Arc::new(Self { rt: Arc::downgrade(rt), cfg, state: Mutex::new(State::default()) })
75    }
76
77    /// Install as the [`PING_TOPIC`] sink and spawn the ping loop. Call once.
78    pub fn start(self: &Arc<Self>) {
79        let Some(rt) = self.rt.upgrade() else { return };
80        rt.set_control_sink(PING_TOPIC, self.clone());
81        let me = Arc::downgrade(self);
82        let interval = self.cfg.interval;
83        tokio::spawn(async move {
84            let mut t = tokio::time::interval(interval);
85            t.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
86            loop {
87                t.tick().await;
88                let Some(me) = me.upgrade() else { break };
89                me.tick().await;
90            }
91        });
92    }
93
94    /// One round: expire stale pings, then ping every verified peer.
95    pub async fn tick(&self) {
96        let Some(rt) = self.rt.upgrade() else { return };
97        let targets: Vec<String> =
98            rt.peer_details().into_iter().filter(|p| p.verified).map(|p| p.node_id).collect();
99        let pings: Vec<(String, u64)> = {
100            let mut st = self.state.lock().unwrap();
101            let timeout = self.cfg.timeout;
102            let expired: Vec<(String, u64)> =
103                st.outstanding.iter().filter(|(_, at)| at.elapsed() > timeout).map(|(k, _)| k.clone()).collect();
104            for k in expired {
105                st.outstanding.remove(&k);
106                if let Some(p) = st.peers.get_mut(&k.0) {
107                    p.missed = p.missed.saturating_add(1);
108                }
109            }
110            // Forget peers that are no longer connected.
111            st.peers.retain(|id, _| targets.contains(id));
112            let mut out = Vec::new();
113            for peer in &targets {
114                let inflight = st.outstanding.keys().filter(|(p, _)| p == peer).count();
115                if inflight >= MAX_OUTSTANDING_PER_PEER {
116                    continue;
117                }
118                st.seq += 1;
119                let n = st.seq;
120                st.outstanding.insert((peer.clone(), n), Instant::now());
121                out.push((peer.clone(), n));
122            }
123            out
124        };
125        for (peer, n) in pings {
126            let _ = rt.route_to_remote(&peer, control(json!({"t": "ping", "n": n}))).await;
127        }
128    }
129
130    /// The liveness view of `peer`, if a pong from it has been counted.
131    pub fn peer(&self, peer: &str) -> Option<PeerLiveness> {
132        self.state.lock().unwrap().peers.get(peer).cloned()
133    }
134
135    /// Count a pong from a verified `peer` for nonce `n`. Returns true when it counted.
136    fn on_pong(&self, peer: &str, n: u64, load: Option<LoadSample>) -> bool {
137        let rtt_ms = {
138            let mut st = self.state.lock().unwrap();
139            let Some(sent) = st.outstanding.remove(&(peer.to_string(), n)) else { return false };
140            let sample = sent.elapsed().as_secs_f64() * 1000.0;
141            let e = st.peers.entry(peer.to_string()).or_insert(PeerLiveness {
142                last_seen: SystemTime::now(),
143                rtt_ms: sample,
144                missed: 0,
145                load: None,
146            });
147            e.rtt_ms = e.rtt_ms * 0.8 + sample * 0.2;
148            e.last_seen = SystemTime::now();
149            e.missed = 0;
150            e.load = load;
151            e.rtt_ms
152        };
153        if let Some(rt) = self.rt.upgrade() {
154            rt.record_heartbeat(peer);
155        }
156        tracing::trace!(peer, rtt_ms, "mesh pong");
157        true
158    }
159}
160
161impl PeerControlSink for Liveness {
162    fn on_peer_control(&self, ctx: &PeerCtx, _conn: u64, payload: &Value) -> Vec<Value> {
163        // Only a verified peer takes part: unverified ids are claims.
164        if !ctx.node_verified {
165            return Vec::new();
166        }
167        let Some(n) = payload.get("n").and_then(Value::as_u64) else { return Vec::new() };
168        match payload.get("t").and_then(Value::as_str) {
169            Some("ping") => {
170                let mut pong = json!({"t": "pong", "n": n});
171                if let Some(l) = LoadSample::local() {
172                    pong["load"] = l.to_json();
173                }
174                vec![pong]
175            }
176            Some("pong") => {
177                let load = payload.get("load").and_then(LoadSample::from_json);
178                self.on_pong(&ctx.peer_id, n, load);
179                Vec::new()
180            }
181            _ => Vec::new(),
182        }
183    }
184}
185
186fn control(value: Value) -> KernelMessage {
187    KernelMessage::new(0, MessageTarget::Topic(PING_TOPIC.to_string()), MessagePayload::Json(value))
188}
189
190#[cfg(test)]
191#[path = "mesh_liveness_tests.rs"]
192mod tests;
clawft_kernel/mesh_liveness.rs

clawft_kernel/
mesh_liveness.rs
