rings_core/
measure.rs

1//! This module provide the `Measure` struct and its implementations.
2//! It is used to assess the reliability of remote peers.
3#![warn(missing_docs)]
4use std::sync::Arc;
5
6use async_trait::async_trait;
7
8use crate::dht::Did;
9
10/// Type of Measure, see [Measure].
11#[cfg(not(feature = "wasm"))]
12pub type MeasureImpl = Arc<dyn BehaviourJudgement + Send + Sync>;
13
14/// Type of Measure, see [crate::measure::Measure].
15#[cfg(feature = "wasm")]
16pub type MeasureImpl = Arc<dyn BehaviourJudgement>;
17
18/// The tag of counters in measure.
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
20pub enum MeasureCounter {
21    /// The number of sent messages.
22    Sent,
23    /// The number of failed to sent messages.
24    FailedToSend,
25    /// The number of received messages.
26    Received,
27    /// The number of failed to receive messages.
28    FailedToReceive,
29    /// The number of connected.
30    Connect,
31    /// The number of disconnect.
32    Disconnected,
33}
34
35/// Local peer-quality class derived from observation counters.
36///
37/// This value is advisory. It orders DHT connection attempts, but it is not a
38/// Chord membership, ownership, or storage-placement proof.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum PeerQuality {
41    /// The peer has positive successful observations and remains below failure limits.
42    Healthy,
43    /// The local node has no useful recent evidence for this peer.
44    Unknown,
45    /// The peer reached one or more local failure limits.
46    Degraded,
47}
48
49impl PeerQuality {
50    /// Return the stable connection-priority rank: smaller is tried first.
51    pub const fn connection_rank(self) -> u8 {
52        match self {
53            Self::Healthy => 0,
54            Self::Unknown => 1,
55            Self::Degraded => 2,
56        }
57    }
58}
59
60/// Failure limits used to classify local peer-quality evidence.
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub struct PeerQualityThresholds {
63    disconnected: u64,
64    failed_to_send: u64,
65    failed_to_receive: u64,
66}
67
68impl PeerQualityThresholds {
69    /// Create classification thresholds.
70    pub const fn new(disconnected: u64, failed_to_send: u64, failed_to_receive: u64) -> Self {
71        Self {
72            disconnected,
73            failed_to_send,
74            failed_to_receive,
75        }
76    }
77}
78
79/// Recent local evidence used to classify a peer.
80///
81/// The counters are local observations only. They do not claim global
82/// reputation and are not signed or replicated.
83#[derive(Debug, Clone, Copy, PartialEq, Eq)]
84pub struct PeerQualityEvidence {
85    connected: u64,
86    disconnected: u64,
87    sent: u64,
88    failed_to_send: u64,
89    received: u64,
90    failed_to_receive: u64,
91}
92
93impl PeerQualityEvidence {
94    /// Build evidence from explicit counter values.
95    pub const fn new(
96        connected: u64,
97        disconnected: u64,
98        sent: u64,
99        failed_to_send: u64,
100        received: u64,
101        failed_to_receive: u64,
102    ) -> Self {
103        Self {
104            connected,
105            disconnected,
106            sent,
107            failed_to_send,
108            received,
109            failed_to_receive,
110        }
111    }
112
113    /// Read all counters for `did` from a measurement implementation.
114    pub async fn from_measure<M>(measure: &M, did: Did) -> Self
115    where M: Measure + ?Sized {
116        Self {
117            connected: measure.get_count(did, MeasureCounter::Connect).await,
118            disconnected: measure.get_count(did, MeasureCounter::Disconnected).await,
119            sent: measure.get_count(did, MeasureCounter::Sent).await,
120            failed_to_send: measure.get_count(did, MeasureCounter::FailedToSend).await,
121            received: measure.get_count(did, MeasureCounter::Received).await,
122            failed_to_receive: measure
123                .get_count(did, MeasureCounter::FailedToReceive)
124                .await,
125        }
126    }
127
128    /// Classify this evidence under the supplied thresholds.
129    pub const fn classify(self, thresholds: PeerQualityThresholds) -> PeerQuality {
130        if self.reaches_failure_limit(thresholds) {
131            PeerQuality::Degraded
132        } else if self.has_positive_observation() {
133            PeerQuality::Healthy
134        } else {
135            PeerQuality::Unknown
136        }
137    }
138
139    /// Return whether any successful local observation exists.
140    pub const fn has_positive_observation(self) -> bool {
141        self.connected > 0 || self.sent > 0 || self.received > 0
142    }
143
144    /// Return whether any failure counter has reached its configured limit.
145    pub const fn reaches_failure_limit(self, thresholds: PeerQualityThresholds) -> bool {
146        self.disconnected >= thresholds.disconnected
147            || self.failed_to_send >= thresholds.failed_to_send
148            || self.failed_to_receive >= thresholds.failed_to_receive
149    }
150}
151
152/// Order DHT connection candidates by advisory peer quality.
153///
154/// Invariant: the returned list is a stable permutation of the input candidate
155/// sequence. The transformation never inserts or removes a `Did`; it only moves
156/// `Healthy` before `Unknown` before `Degraded`.
157/// Preservation: because the set of candidates is unchanged, Chord ownership,
158/// successor responsibility, and storage placement remain determined only by the
159/// DHT transition that produced those candidates.
160pub fn order_peers_by_quality(
161    candidates: impl IntoIterator<Item = (Did, PeerQuality)>,
162) -> Vec<Did> {
163    let mut ranked = candidates
164        .into_iter()
165        .enumerate()
166        .map(|(index, (did, quality))| (quality.connection_rank(), index, did))
167        .collect::<Vec<_>>();
168    ranked.sort_by_key(|(rank, index, _)| (*rank, *index));
169    ranked.into_iter().map(|(_, _, did)| did).collect()
170}
171
172/// `Measure` is used to assess the reliability of peers by counting their behaviour.
173/// It currently count the number of sent and received messages in a given period (1 hour).
174/// The method [Measure::incr] should be called in the proper places.
175#[cfg_attr(feature = "wasm", async_trait(?Send))]
176#[cfg_attr(not(feature = "wasm"), async_trait)]
177pub trait Measure {
178    /// `incr` increments the counter of the given peer.
179    async fn incr(&self, did: Did, counter: MeasureCounter);
180    /// `get_count` returns the counter of the given peer.
181    async fn get_count(&self, did: Did, counter: MeasureCounter) -> u64;
182}
183
184/// `BehaviourJudgement` classifies local evidence about a peer.
185#[cfg_attr(feature = "wasm", async_trait(?Send))]
186#[cfg_attr(not(feature = "wasm"), async_trait)]
187pub trait BehaviourJudgement: Measure {
188    /// Classify local peer quality for DHT connection scheduling.
189    ///
190    /// This value is advisory. It orders connection attempts and does not gate
191    /// Chord membership, routing, ownership, or storage placement.
192    async fn quality(&self, did: Did) -> PeerQuality;
193
194    /// Return the legacy boolean judgement for callers that need a yes/no decision.
195    ///
196    /// This method is intentionally independent from [Self::quality]. Mapping
197    /// the three-valued quality order to a boolean would turn advisory DHT
198    /// scheduling evidence into a hidden gating rule.
199    async fn good(&self, did: Did) -> bool;
200}
201
202/// `ConnectBehaviour` trait offers a default implementation for the `good` method, providing a judgement
203/// based on a node's behavior in establishing connections.
204/// The "goodness" of a node is measured by comparing disconnection counts against a given threshold.
205#[cfg_attr(feature = "wasm", async_trait(?Send))]
206#[cfg_attr(not(feature = "wasm"), async_trait)]
207pub trait ConnectBehaviour<const THRESHOLD: u64>: Measure {
208    /// This asynchronous method returns a boolean indicating whether the node identified by `did` has a satisfactory connection behavior.
209    async fn good(&self, did: Did) -> bool {
210        let conn = self.get_count(did, MeasureCounter::Connect).await;
211        let disconn = self.get_count(did, MeasureCounter::Disconnected).await;
212        tracing::debug!(
213            "[ConnectBehaviour] in threshold: {:}, connect: {:}, disconn: {:}",
214            THRESHOLD,
215            conn,
216            disconn
217        );
218        disconn < THRESHOLD
219    }
220}
221
222/// `MessageSendBehaviour` trait provides a default implementation for the `good` method, judging a node's
223/// behavior based on its message sending capabilities.
224/// The "goodness" of a node is measured by comparing the sent and failed-to-send counts against a given threshold.
225#[cfg_attr(feature = "wasm", async_trait(?Send))]
226#[cfg_attr(not(feature = "wasm"), async_trait)]
227pub trait MessageSendBehaviour<const THRESHOLD: u64>: Measure {
228    /// This asynchronous method returns a boolean indicating whether the node identified by `did` has a satisfactory message sending behavior.
229    async fn good(&self, did: Did) -> bool {
230        let failed = self.get_count(did, MeasureCounter::FailedToSend).await;
231        failed < THRESHOLD
232    }
233}
234
235/// `MessageRecvBehaviour` trait provides a default implementation for the `good` method, assessing a node's
236/// behavior based on its message receiving capabilities.
237/// The "goodness" of a node is measured by comparing the received and failed-to-receive counts against a given threshold.
238#[cfg_attr(feature = "wasm", async_trait(?Send))]
239#[cfg_attr(not(feature = "wasm"), async_trait)]
240pub trait MessageRecvBehaviour<const THRESHOLD: u64>: Measure {
241    /// This asynchronous method returns a boolean indicating whether the node identified by `did` has a satisfactory message receiving behavior.
242    async fn good(&self, did: Did) -> bool {
243        let failed = self.get_count(did, MeasureCounter::FailedToReceive).await;
244        failed < THRESHOLD
245    }
246}
247
248#[cfg(test)]
249mod tests {
250    use super::*;
251    use crate::ecc::SecretKey;
252
253    fn did() -> Did {
254        SecretKey::random().address().into()
255    }
256
257    #[test]
258    fn peer_quality_evidence_classifies_unknown_healthy_and_degraded() {
259        let thresholds = PeerQualityThresholds::new(3, 10, 10);
260        assert_eq!(
261            PeerQualityEvidence::new(0, 0, 0, 0, 0, 0).classify(thresholds),
262            PeerQuality::Unknown
263        );
264        assert_eq!(
265            PeerQualityEvidence::new(1, 0, 0, 0, 0, 0).classify(thresholds),
266            PeerQuality::Healthy
267        );
268        assert_eq!(
269            PeerQualityEvidence::new(1, 3, 0, 0, 0, 0).classify(thresholds),
270            PeerQuality::Degraded
271        );
272        assert_eq!(
273            PeerQualityEvidence::new(1, 0, 0, 10, 0, 0).classify(thresholds),
274            PeerQuality::Degraded
275        );
276        assert_eq!(
277            PeerQualityEvidence::new(1, 0, 0, 0, 0, 10).classify(thresholds),
278            PeerQuality::Degraded
279        );
280    }
281
282    #[test]
283    fn order_peers_by_quality_is_stable_permutation() {
284        let degraded = did();
285        let unknown_a = did();
286        let healthy = did();
287        let unknown_b = did();
288
289        let ordered = order_peers_by_quality([
290            (degraded, PeerQuality::Degraded),
291            (unknown_a, PeerQuality::Unknown),
292            (healthy, PeerQuality::Healthy),
293            (unknown_b, PeerQuality::Unknown),
294        ]);
295
296        assert_eq!(ordered, vec![healthy, unknown_a, unknown_b, degraded]);
297    }
298}
rings_core/measure.rs

rings_core/
measure.rs
