fedimint_server/consensus/
peer_availability.rs

1//! This guardian's view of its peers' availability.
2//!
3//! A peer counts as available in a 1-minute slot if it sent us any AlephBFT
4//! message in it. Every running peer sends consensus traffic while the
5//! federation is active, including a slow one that lags behind and still runs
6//! an older session, while an offline or frozen peer, or one whose consensus
7//! is stuck, sends nothing. A slot in which we heard from fewer than
8//! `threshold - 1` peers means the federation was idle, or we were down or
9//! cut off ourselves, so it does not count.
10//!
11//! [`PeerActivity`] counts slots in memory; [`persist_peer_availability`]
12//! folds them every 10 minutes into exponential moving averages with the time
13//! constants in [`PEER_AVAILABILITY_WINDOWS_SECS`], persisted as
14//! [`PeerAvailabilityState`].
15
16use std::collections::{BTreeMap, BTreeSet};
17use std::io::{Error, Read, Write};
18use std::sync::{Arc, Mutex};
19use std::time::Duration;
20
21use fedimint_core::db::{Database, IDatabaseTransactionOpsCoreTyped};
22use fedimint_core::encoding::{Decodable, DecodeError, Encodable};
23use fedimint_core::envs::is_running_in_test_env;
24use fedimint_core::module::registry::ModuleDecoderRegistry;
25use fedimint_core::peer_availability::{
26    PEER_AVAILABILITY_WINDOWS_SECS, PeerAvailability, WindowAvailability,
27};
28use fedimint_core::task::sleep;
29use fedimint_core::time::duration_since_epoch;
30use fedimint_core::{NumPeers, PeerId};
31use fedimint_logging::LOG_CONSENSUS;
32use tracing::debug;
33
34use super::db::PeerAvailabilityKey;
35
36const SLOT_SECS: u64 = 60;
37
38/// Folds the [`PeerActivity`] into the persisted [`PeerAvailabilityState`]
39/// every 10 minutes, or every 10 seconds in tests. A restart loses at most one
40/// interval of activity.
41pub async fn persist_peer_availability(db: Database, peer_activity: Arc<Mutex<PeerActivity>>) {
42    let interval = if is_running_in_test_env() {
43        Duration::from_secs(10)
44    } else {
45        Duration::from_secs(600)
46    };
47
48    loop {
49        sleep(interval).await;
50
51        let now_secs = duration_since_epoch().as_secs();
52        let activity = peer_activity.lock().expect("poisoned").take(now_secs);
53
54        debug!(target: LOG_CONSENSUS, ?activity, "Peer activity since last update");
55
56        let mut dbtx = db.begin_transaction().await;
57        let mut state = dbtx
58            .get_value(&PeerAvailabilityKey)
59            .await
60            .unwrap_or_default();
61        state.update(now_secs, &activity);
62        dbtx.insert_entry(&PeerAvailabilityKey, &state).await;
63        dbtx.commit_tx().await;
64    }
65}
66
67/// Slots of consensus traffic per peer since the last [`PeerActivity::take`].
68#[derive(Debug)]
69pub struct PeerActivity {
70    identity: PeerId,
71    num_peers: NumPeers,
72    slot: u64,
73    heard_in_slot: BTreeSet<PeerId>,
74    observed_slots: u32,
75    active_slots: BTreeMap<PeerId, u32>,
76    last_heard_secs: BTreeMap<PeerId, u64>,
77}
78
79/// What [`PeerActivity`] observed since it was last taken.
80#[derive(Debug, Clone, Default, PartialEq)]
81pub struct ObservedActivity {
82    pub observed_secs: u64,
83    /// Each peer's share of the observed slots in `0.0..=1.0`
84    pub availability: BTreeMap<PeerId, f64>,
85    pub last_heard_secs: BTreeMap<PeerId, u64>,
86}
87
88impl PeerActivity {
89    pub fn new(identity: PeerId, num_peers: NumPeers) -> Self {
90        Self {
91            identity,
92            num_peers,
93            slot: 0,
94            heard_in_slot: BTreeSet::new(),
95            observed_slots: 0,
96            active_slots: BTreeMap::new(),
97            last_heard_secs: BTreeMap::new(),
98        }
99    }
100
101    /// Records a consensus message from `peer`.
102    pub fn heard(&mut self, peer: PeerId, now_secs: u64) {
103        self.advance(now_secs);
104        self.heard_in_slot.insert(peer);
105        self.last_heard_secs.insert(peer, now_secs);
106    }
107
108    /// Returns the activity of the slots closed since the last call.
109    pub fn take(&mut self, now_secs: u64) -> ObservedActivity {
110        self.advance(now_secs);
111
112        let observed_slots = std::mem::take(&mut self.observed_slots);
113        let mut active_slots = std::mem::take(&mut self.active_slots);
114
115        // We were up in every slot we observed
116        active_slots.insert(self.identity, observed_slots);
117        if 0 < observed_slots {
118            self.last_heard_secs.insert(self.identity, now_secs);
119        }
120
121        ObservedActivity {
122            observed_secs: u64::from(observed_slots) * SLOT_SECS,
123            availability: self
124                .num_peers
125                .peer_ids()
126                .filter(|_| 0 < observed_slots)
127                .map(|peer| {
128                    let active = active_slots.get(&peer).copied().unwrap_or(0);
129                    (peer, f64::from(active) / f64::from(observed_slots))
130                })
131                .collect(),
132            last_heard_secs: self.last_heard_secs.clone(),
133        }
134    }
135
136    /// Closes the current slot once `now_secs` is past it. Slots in which
137    /// nothing at all arrived are not observed.
138    fn advance(&mut self, now_secs: u64) {
139        let slot = now_secs / SLOT_SECS;
140        if slot == self.slot {
141            return;
142        }
143
144        // We never hear from ourselves
145        if self.num_peers.threshold() - 1 <= self.heard_in_slot.len() {
146            self.observed_slots += 1;
147            for peer in &self.heard_in_slot {
148                *self.active_slots.entry(*peer).or_default() += 1;
149            }
150        }
151
152        self.heard_in_slot.clear();
153        self.slot = slot;
154    }
155}
156
157/// Moving averages of each peer's availability, one per window.
158#[derive(Debug, Clone, Default, PartialEq, Encodable, Decodable)]
159pub struct PeerAvailabilityState {
160    updated_at_secs: u64,
161    /// Last time each peer sent us a consensus message
162    last_seen_secs: BTreeMap<PeerId, u64>,
163    /// Per window: moving average of 1 over the time we observed
164    coverage: Vec<Average>,
165    /// Per window: moving average of each peer's availability
166    availability: Vec<BTreeMap<PeerId, Average>>,
167}
168
169/// An `f64` stored by its bits, so the averages survive restarts exactly.
170#[derive(Debug, Clone, Copy, Default, PartialEq)]
171struct Average(f64);
172
173impl Encodable for Average {
174    fn consensus_encode<W: Write>(&self, writer: &mut W) -> Result<(), Error> {
175        self.0.to_bits().consensus_encode(writer)
176    }
177}
178
179impl Decodable for Average {
180    fn consensus_decode_partial<R: Read>(
181        reader: &mut R,
182        modules: &ModuleDecoderRegistry,
183    ) -> Result<Self, DecodeError> {
184        Ok(Self(f64::from_bits(u64::consensus_decode_partial(
185            reader, modules,
186        )?)))
187    }
188}
189
190impl PeerAvailabilityState {
191    /// Folds in the activity observed since the last update.
192    pub fn update(&mut self, now_secs: u64, activity: &ObservedActivity) {
193        for (peer, heard) in &activity.last_heard_secs {
194            let last_seen = self.last_seen_secs.entry(*peer).or_default();
195            *last_seen = (*last_seen).max(*heard);
196        }
197
198        // Everything decays over the real elapsed time, but only the observed
199        // part of it adds new weight
200        let elapsed = now_secs.saturating_sub(self.updated_at_secs) as f64;
201        let observed = activity.observed_secs as f64;
202
203        let windows = PEER_AVAILABILITY_WINDOWS_SECS.len();
204        self.coverage.resize(windows, Average::default());
205        self.availability.resize(windows, BTreeMap::new());
206
207        for (window, window_secs) in PEER_AVAILABILITY_WINDOWS_SECS.iter().enumerate() {
208            let decay = (-elapsed.max(observed) / *window_secs as f64).exp();
209            let gain = 1.0 - (-observed / *window_secs as f64).exp();
210
211            self.coverage[window].0 = self.coverage[window].0 * decay + gain;
212
213            for average in self.availability[window].values_mut() {
214                average.0 *= decay;
215            }
216
217            for (peer, value) in &activity.availability {
218                self.availability[window].entry(*peer).or_default().0 += gain * value;
219            }
220        }
221
222        self.updated_at_secs = now_secs;
223    }
224
225    /// Returns the averages as seen at `now_secs`.
226    pub fn report(&self, now_secs: u64) -> PeerAvailability {
227        let elapsed = now_secs.saturating_sub(self.updated_at_secs) as f64;
228
229        PeerAvailability {
230            last_seen_secs: self.last_seen_secs.clone(),
231            windows: PEER_AVAILABILITY_WINDOWS_SECS
232                .iter()
233                .zip(&self.coverage)
234                .zip(&self.availability)
235                .map(
236                    |((window_secs, coverage), availability)| WindowAvailability {
237                        window_secs: *window_secs,
238                        // Time since the last update is unobserved, so coverage
239                        // decays over it while the ratio below does not change
240                        coverage_bps: to_bps(coverage.0 * (-elapsed / *window_secs as f64).exp()),
241                        peers: availability
242                            .iter()
243                            .map(|(peer, average)| (*peer, to_bps(average.0 / coverage.0)))
244                            .collect(),
245                    },
246                )
247                .collect(),
248        }
249    }
250}
251
252fn to_bps(value: f64) -> u16 {
253    (value * 10_000.0).round().clamp(0.0, 10_000.0) as u16
254}
255
256#[cfg(test)]
257mod tests;
fedimint_server/consensus/peer_availability.rs

fedimint_server/consensus/
peer_availability.rs
