rings-node 0.20.0

Rings is a structured peer-to-peer network implementation using WebRTC, Chord algorithm, and full WebAssembly (WASM) support.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

use std::collections::HashMap;

use rings_core::dht::Did;

use crate::error::OnionQueueAdmissionReason;

/// Pure global/per-peer cardinality bound shared by queue and session reducers.
///
/// Invariant: `total = sum(per_peer.values())`, zero peer entries are absent,
/// `total <= max_total`, and each peer count is at most `max_per_peer`.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct PeerQuota {
    total: usize,
    per_peer: HashMap<Did, usize>,
    max_total: usize,
    max_per_peer: usize,
}

impl PeerQuota {
    pub(crate) fn new(max_total: usize, max_per_peer: usize) -> Self {
        Self {
            total: 0,
            per_peer: HashMap::new(),
            max_total,
            max_per_peer,
        }
    }

    pub(crate) const fn total(&self) -> usize {
        self.total
    }

    pub(crate) fn peer_total(&self, peer: Did) -> usize {
        self.per_peer.get(&peer).copied().unwrap_or_default()
    }

    /// Prove that one successor reservation is within both bounds without mutating state.
    pub(crate) fn can_reserve(&self, peer: Did) -> Result<(), OnionQueueAdmissionReason> {
        self.successor(peer).map(|_| ())
    }

    fn successor(&self, peer: Did) -> Result<(usize, usize), OnionQueueAdmissionReason> {
        let next_total = self
            .total
            .checked_add(1)
            .ok_or(OnionQueueAdmissionReason::CounterOverflow)?;
        let next_peer = self
            .peer_total(peer)
            .checked_add(1)
            .ok_or(OnionQueueAdmissionReason::CounterOverflow)?;
        if next_total > self.max_total {
            return Err(OnionQueueAdmissionReason::GlobalFull);
        }
        if next_peer > self.max_per_peer {
            return Err(OnionQueueAdmissionReason::PeerFull);
        }
        Ok((next_total, next_peer))
    }

    /// Reserve one unit atomically after the immutable successor proof succeeds.
    pub(crate) fn reserve(&mut self, peer: Did) -> Result<(), OnionQueueAdmissionReason> {
        let (next_total, next_peer) = self.successor(peer)?;
        self.total = next_total;
        self.per_peer.insert(peer, next_peer);
        Ok(())
    }

    /// Release one existing unit; an inconsistent predecessor leaves state unchanged.
    pub(crate) fn release(&mut self, peer: Did) -> bool {
        let Some(peer_total) = self.per_peer.get(&peer).copied() else {
            return false;
        };
        let Some(next_total) = self.total.checked_sub(1) else {
            return false;
        };
        let Some(next_peer) = peer_total.checked_sub(1) else {
            return false;
        };
        self.total = next_total;
        if next_peer == 0 {
            self.per_peer.remove(&peer);
        } else {
            self.per_peer.insert(peer, next_peer);
        }
        true
    }

    /// Release one peer's entire projection atomically.
    ///
    /// Post: on success, the peer is absent and `total` is reduced by exactly the returned count.
    pub(crate) fn release_peer(&mut self, peer: Did) -> Option<usize> {
        let peer_total = self.per_peer.get(&peer).copied()?;
        let next_total = self.total.checked_sub(peer_total)?;
        self.total = next_total;
        self.per_peer.remove(&peer);
        Some(peer_total)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_reserve_and_release_preserve_global_and_peer_projection() {
        let mut quota = PeerQuota::new(2, 1);
        let first = Did::from(1_u32);
        let second = Did::from(2_u32);

        assert_eq!(quota.reserve(first), Ok(()));
        assert_eq!(
            quota.reserve(first),
            Err(OnionQueueAdmissionReason::PeerFull)
        );
        assert_eq!(quota.total(), 1);
        assert_eq!(quota.peer_total(first), 1);
        assert_eq!(quota.reserve(second), Ok(()));
        assert_eq!(
            quota.reserve(Did::from(3_u32)),
            Err(OnionQueueAdmissionReason::GlobalFull)
        );
        assert!(quota.release(first));
        assert!(!quota.release(first));
        assert_eq!(quota.total(), 1);
        assert_eq!(quota.peer_total(second), 1);
        assert_eq!(quota.release_peer(second), Some(1));
        assert_eq!(quota.release_peer(second), None);
        assert_eq!(quota.total(), 0);
    }
}