clawft_kernel/
mesh_swarm_picker.rs

1//! Piece and peer selection for swarm fetches (ADR-099 section 6, card
2//! mesh-placement-25). Pure logic, no I/O.
3//!
4//! - [`PiecePicker`] hands each peer the **rarest** piece it holds that is
5//!   still needed and not already requested from someone else. Rarity is
6//!   the number of live peers that hold the piece; ties go to the lowest
7//!   index so a fetch is reproducible. When a peer is lost its in-flight
8//!   pieces return to the pool and its holdings stop counting.
9//! - [`order_peers`] ranks fetch candidates: higher trust tier first (a LAN
10//!   id is self-asserted), then same LAN as the local node, then by measured link speed (peers never measured rank as an
11//!   assumed default, so they get tried), then by id.
12//! - [`LinkStats`] keeps a smoothed bytes-per-second figure per peer from
13//!   the transfers this node actually performed. It is locally measured:
14//!   a peer's own claims about its link are never used.
15
16use std::collections::{BTreeMap, HashMap};
17
18use dashmap::DashMap;
19
20use clawft_types::placement::TrustTier;
21
22use crate::mesh_artifact_types::Bitfield;
23
24/// What the picker tells one peer's worker to do next.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum Pick {
27    /// Request this piece from the peer.
28    Piece(u32),
29    /// Nothing free for this peer now, but pieces it holds are in flight
30    /// elsewhere and may come back: wait for a change.
31    Wait,
32    /// The peer holds nothing still needed: done with it.
33    Nothing,
34}
35
36/// Which pieces are held, in flight and available where.
37#[derive(Debug, Clone)]
38pub struct PiecePicker {
39    have: Bitfield,
40    in_flight: BTreeMap<u32, String>,
41    peers: HashMap<String, Bitfield>,
42}
43
44impl PiecePicker {
45    /// Picker over `have` (pieces already held locally).
46    pub fn new(have: Bitfield) -> Self {
47        Self {
48            have,
49            in_flight: BTreeMap::new(),
50            peers: HashMap::new(),
51        }
52    }
53
54    /// A peer joined with this holding.
55    pub fn add_peer(&mut self, peer: &str, has: Bitfield) {
56        self.peers.insert(peer.to_string(), has);
57    }
58
59    /// A peer is gone: its holdings stop counting and its in-flight pieces
60    /// return to the pool.
61    pub fn remove_peer(&mut self, peer: &str) {
62        self.peers.remove(peer);
63        self.in_flight.retain(|_, p| p != peer);
64    }
65
66    /// The peer says it cannot supply piece `i` after all.
67    pub fn peer_lacks(&mut self, peer: &str, i: u32) {
68        if let Some(b) = self.peers.get_mut(peer) {
69            b.set(i, false);
70        }
71        if self.in_flight.get(&i).is_some_and(|p| p == peer) {
72            self.in_flight.remove(&i);
73        }
74    }
75
76    /// Live peers holding piece `i`.
77    pub fn availability(&self, i: u32) -> usize {
78        self.peers.values().filter(|b| b.get(i)).count()
79    }
80
81    /// Next piece for `peer`, claiming it (see [`Pick`]).
82    pub fn pick(&mut self, peer: &str) -> Pick {
83        let Some(has) = self.peers.get(peer) else {
84            return Pick::Nothing;
85        };
86        let mut best: Option<(usize, u32)> = None;
87        let mut blocked = false;
88        for i in self.have.missing() {
89            if !has.get(i) {
90                continue;
91            }
92            if self.in_flight.contains_key(&i) {
93                blocked = true;
94                continue;
95            }
96            let key = (self.availability(i), i);
97            if best.is_none_or(|b| key < b) {
98                best = Some(key);
99            }
100        }
101        match best {
102            Some((_, i)) => {
103                self.in_flight.insert(i, peer.to_string());
104                Pick::Piece(i)
105            }
106            None if blocked => Pick::Wait,
107            None => Pick::Nothing,
108        }
109    }
110
111    /// Piece `i` arrived and verified.
112    pub fn complete(&mut self, i: u32) {
113        self.have.set(i, true);
114        self.in_flight.remove(&i);
115    }
116
117    /// Piece `i` was not delivered: back to the pool.
118    pub fn release(&mut self, i: u32) {
119        self.in_flight.remove(&i);
120    }
121
122    /// True when every piece is held.
123    pub fn is_complete(&self) -> bool {
124        self.have.is_complete()
125    }
126
127    /// Pieces still missing.
128    pub fn missing(&self) -> u32 {
129        self.have.len() - self.have.count()
130    }
131
132    /// Missing pieces no live peer holds.
133    pub fn unavailable(&self) -> usize {
134        self.have.missing().filter(|&i| self.availability(i) == 0).count()
135    }
136}
137
138/// A peer that may hold an artifact.
139#[derive(Debug, Clone, PartialEq)]
140pub struct PeerCandidate {
141    /// Peer node id.
142    pub peer_id: String,
143    /// LAN the peer advertises (`net.lan` fact), for locality. Self-asserted,
144    /// so it only orders peers of the same tier.
145    pub lan_id: Option<String>,
146    /// Trust tier this node assigned the peer (`Discovered` when unknown).
147    pub tier: TrustTier,
148}
149
150impl PeerCandidate {
151    /// Candidate with no locality information.
152    pub fn new(peer_id: impl Into<String>) -> Self {
153        Self {
154            peer_id: peer_id.into(),
155            lan_id: None,
156            tier: TrustTier::Discovered,
157        }
158    }
159
160    /// Candidate with trust tier `tier`.
161    pub fn with_tier(mut self, tier: TrustTier) -> Self {
162        self.tier = tier;
163        self
164    }
165
166    /// Candidate on `lan`.
167    pub fn on_lan(mut self, lan: impl Into<String>) -> Self {
168        self.lan_id = Some(lan.into());
169        self
170    }
171}
172
173/// Link speed assumed for a peer this node never transferred from, so it
174/// is tried rather than starved (bytes per second).
175pub const ASSUMED_LINK_BPS: f64 = 10.0 * 1024.0 * 1024.0;
176
177/// Rank `candidates`: higher trust tier first, then same LAN as `local_lan`,
178/// then faster measured link, then peer id. Banned peers are dropped.
179pub fn order_peers(
180    candidates: Vec<PeerCandidate>,
181    local_lan: Option<&str>,
182    links: &LinkStats,
183    is_banned: &dyn Fn(&str) -> bool,
184) -> Vec<PeerCandidate> {
185    let mut v: Vec<(bool, f64, PeerCandidate)> = candidates
186        .into_iter()
187        .filter(|c| !is_banned(&c.peer_id))
188        .map(|c| {
189            let same_lan = matches!((local_lan, c.lan_id.as_deref()), (Some(a), Some(b)) if a == b);
190            let bps = links.bytes_per_sec(&c.peer_id).unwrap_or(ASSUMED_LINK_BPS);
191            (same_lan, bps, c)
192        })
193        .collect();
194    v.sort_by(|a, b| {
195        b.2.tier
196            .cmp(&a.2.tier)
197            .then(b.0.cmp(&a.0))
198            .then(b.1.total_cmp(&a.1))
199            .then(a.2.peer_id.cmp(&b.2.peer_id))
200    });
201    v.into_iter().map(|(_, _, c)| c).collect()
202}
203
204/// Smoothed per-peer throughput measured by this node.
205#[derive(Debug, Default)]
206pub struct LinkStats {
207    peers: DashMap<String, f64>,
208}
209
210/// Weight of the newest sample in the smoothed figure.
211const EWMA_ALPHA: f64 = 0.3;
212
213impl LinkStats {
214    /// Record `bytes` received from `peer` in `secs` seconds.
215    pub fn record(&self, peer: &str, bytes: u64, secs: f64) {
216        if bytes == 0 || secs <= 0.0 {
217            return;
218        }
219        let sample = bytes as f64 / secs;
220        self.peers
221            .entry(peer.to_string())
222            .and_modify(|v| *v = EWMA_ALPHA * sample + (1.0 - EWMA_ALPHA) * *v)
223            .or_insert(sample);
224    }
225
226    /// Smoothed bytes per second from `peer`, if ever measured.
227    pub fn bytes_per_sec(&self, peer: &str) -> Option<f64> {
228        self.peers.get(peer).map(|v| *v)
229    }
230}
231
232#[cfg(test)]
233#[path = "mesh_swarm_picker_tests.rs"]
234mod tests;
clawft_kernel/mesh_swarm_picker.rs

clawft_kernel/
mesh_swarm_picker.rs
