rings_core/dht/chord/
peer_ring.rs

1use std::sync::Arc;
2use std::sync::Mutex;
3use std::sync::MutexGuard;
4
5use async_trait::async_trait;
6
7use super::PeerRingAction;
8use super::RemoteAction;
9use super::TopoInfo;
10use crate::dht::did::BiasId;
11use crate::dht::entry::Entry;
12use crate::dht::finger::DEFAULT_FINGER_TABLE_SIZE;
13use crate::dht::successor::SuccessorReader;
14use crate::dht::successor::SuccessorSeq;
15use crate::dht::topology;
16use crate::dht::topology::FindSuccessorStep;
17use crate::dht::topology::SuccessorRemoval;
18use crate::dht::topology::TopologyAction;
19use crate::dht::topology::TopologyEvent;
20use crate::dht::topology::TopologyState;
21use crate::dht::topology::TopologyStep;
22use crate::dht::types::Chord;
23use crate::dht::types::CorrectChord;
24use crate::dht::virtual_node::VirtualNodeConfig;
25use crate::dht::Did;
26use crate::dht::FingerTable;
27use crate::dht::LiveDid;
28use crate::error::Error;
29use crate::error::Result;
30use crate::storage::KvStorageInterface;
31use crate::storage::MemStorage;
32
33/// Storage accepted by [`PeerRing::new_with_storage`].
34#[cfg(all(feature = "wasm", target_family = "wasm"))]
35pub type EntryStorage = Box<dyn KvStorageInterface<Entry>>;
36
37/// Storage accepted by [`PeerRing::new_with_storage`].
38#[cfg(not(all(feature = "wasm", target_family = "wasm")))]
39pub type EntryStorage = Box<dyn KvStorageInterface<Entry> + Send + Sync>;
40
41/// Chord routing and replicated-storage state for one network peer.
42pub struct PeerRing {
43    /// The DID of the current node.
44    pub did: Did,
45    finger: Arc<Mutex<FingerTable>>,
46    successor_seq: SuccessorSeq,
47    predecessor: Arc<Mutex<Option<Did>>>,
48    /// Persistent replicated-entry storage.
49    pub storage: EntryStorage,
50    /// Local fetched-entry cache.
51    pub cache: EntryStorage,
52    storage_virtual_node_config: VirtualNodeConfig,
53    topology_transition: Mutex<()>,
54}
55
56impl PeerRing {
57    /// Construct a peer ring with caller-provided entry storage.
58    pub fn new_with_storage(did: Did, succ_max: u8, storage: EntryStorage) -> Self {
59        Self::new_with_storage_and_finger_table_size(
60            did,
61            succ_max,
62            storage,
63            DEFAULT_FINGER_TABLE_SIZE,
64        )
65    }
66
67    /// Construct a peer ring with caller-provided storage and finger-table size.
68    ///
69    /// `Did` is 160-bit. Sizes above [`DEFAULT_FINGER_TABLE_SIZE`] are clamped
70    /// by [`FingerTable::new`]; zero disables finger maintenance.
71    pub fn new_with_storage_and_finger_table_size(
72        did: Did,
73        succ_max: u8,
74        storage: EntryStorage,
75        finger_table_size: usize,
76    ) -> Self {
77        Self::new_with_storage_finger_table_size_and_virtual_nodes(
78            did,
79            succ_max,
80            storage,
81            finger_table_size,
82            VirtualNodeConfig::disabled(),
83        )
84    }
85
86    /// Construct a peer ring with storage and virtual ownership configuration.
87    pub fn new_with_storage_finger_table_size_and_virtual_nodes(
88        did: Did,
89        succ_max: u8,
90        storage: EntryStorage,
91        finger_table_size: usize,
92        virtual_nodes: VirtualNodeConfig,
93    ) -> Self {
94        Self {
95            successor_seq: SuccessorSeq::new(did, succ_max),
96            predecessor: Arc::new(Mutex::new(None)),
97            finger: Arc::new(Mutex::new(FingerTable::new(did, finger_table_size))),
98            storage,
99            cache: Box::new(MemStorage::new()),
100            storage_virtual_node_config: virtual_nodes,
101            topology_transition: Mutex::new(()),
102            did,
103        }
104    }
105
106    /// Return the successor sequence.
107    #[deprecated(note = "use PeerRing::successors")]
108    pub fn lock_successor(&self) -> Result<SuccessorSeq> {
109        Ok(self.successor_seq.clone())
110    }
111
112    /// Return the successor sequence.
113    pub fn successors(&self) -> SuccessorSeq {
114        self.successor_seq.clone()
115    }
116
117    fn lock_finger_state(&self) -> Result<MutexGuard<'_, FingerTable>> {
118        self.finger.lock().map_err(|_| Error::DHTSyncLockError)
119    }
120
121    fn lock_predecessor_state(&self) -> Result<MutexGuard<'_, Option<Did>>> {
122        self.predecessor.lock().map_err(|_| Error::DHTSyncLockError)
123    }
124
125    #[cfg(all(test, not(all(feature = "wasm", target_family = "wasm"))))]
126    pub(crate) fn lock_finger(&self) -> Result<MutexGuard<'_, FingerTable>> {
127        self.lock_finger_state()
128    }
129
130    #[cfg(test)]
131    pub(crate) fn replace_fingers_for_test(&self, fingers: &[(usize, Did)]) -> Result<()> {
132        let _transition = self
133            .topology_transition
134            .lock()
135            .map_err(|_| Error::DHTSyncLockError)?;
136        let mut observed = self.lock_finger_state()?;
137        for (index, did) in fingers {
138            if *index >= observed.slot_count() {
139                return Err(Error::InvalidMessage(format!(
140                    "test finger index {index} exceeds slot count {}",
141                    observed.slot_count()
142                )));
143            }
144            if *did == self.did {
145                return Err(Error::InvalidMessage(
146                    "test finger fixture cannot contain the local DID".to_owned(),
147                ));
148            }
149        }
150        observed.reset_finger();
151        for (index, did) in fingers {
152            observed.set(*index, *did);
153        }
154        Ok(())
155    }
156
157    #[cfg(test)]
158    pub(crate) fn lock_predecessor(&self) -> Result<MutexGuard<'_, Option<Did>>> {
159        self.lock_predecessor_state()
160    }
161
162    /// Remove a node from finger, predecessor, and successor state.
163    pub fn remove(&self, did: Did) -> Result<()> {
164        self.remove_with_successor_evidence(did, SuccessorRemoval::Preserve)
165    }
166
167    /// Remove an unavailable node using transport-validated successor evidence.
168    pub(crate) fn remove_unavailable(&self, did: Did, replacements: Vec<Did>) -> Result<()> {
169        self.remove_with_successor_evidence(did, SuccessorRemoval::ReplaceWith(replacements))
170    }
171
172    fn remove_with_successor_evidence(&self, did: Did, successor: SuccessorRemoval) -> Result<()> {
173        self.transition_topology(TopologyEvent::Remove {
174            peer: did,
175            successor,
176        })
177        .map(|_| ())
178    }
179
180    /// Calculate the DID's clockwise bias from this node.
181    pub fn bias(&self, did: Did) -> BiasId {
182        BiasId::new(self.did, did)
183    }
184
185    pub(crate) fn topology_state(&self) -> Result<TopologyState> {
186        self.with_topology_state(Clone::clone)
187    }
188
189    pub(crate) fn with_topology_state<T>(
190        &self,
191        observe: impl FnOnce(&TopologyState) -> T,
192    ) -> Result<T> {
193        let _transition = self
194            .topology_transition
195            .lock()
196            .map_err(|_| Error::DHTSyncLockError)?;
197        let state = self.topology_state_unlocked()?;
198        Ok(observe(&state))
199    }
200
201    fn topology_state_unlocked(&self) -> Result<TopologyState> {
202        let successors = self.successor_seq.list()?;
203        let predecessor = *self.lock_predecessor_state()?;
204        let finger = self.lock_finger_state()?;
205        Ok(TopologyState::new(
206            self.did,
207            successors,
208            predecessor,
209            finger.list().clone(),
210            finger.fix_finger_index(),
211        ))
212    }
213
214    pub(in crate::dht) const fn storage_virtual_node_config(&self) -> VirtualNodeConfig {
215        self.storage_virtual_node_config
216    }
217
218    fn transition_topology(&self, event: TopologyEvent) -> Result<TopologyStep> {
219        self.transition_topology_with_observer(event, |_| {})
220    }
221
222    pub(super) fn transition_topology_with_observer(
223        &self,
224        event: TopologyEvent,
225        observe_snapshot: impl FnOnce(&TopologyState),
226    ) -> Result<TopologyStep> {
227        let _transition = self
228            .topology_transition
229            .lock()
230            .map_err(|_| Error::DHTSyncLockError)?;
231        let current = self.topology_state_unlocked()?;
232        observe_snapshot(&current);
233        let next = topology::step(&current, event, self.successor_seq.capacity());
234        self.interpret_topology_state_unlocked(&next.state)?;
235        Ok(next)
236    }
237
238    fn interpret_topology_state_unlocked(&self, next: &TopologyState) -> Result<()> {
239        let mut predecessor = self.lock_predecessor_state()?;
240        let mut finger = self.lock_finger_state()?;
241        self.successor_seq.replace_state(&next.successors)?;
242        *predecessor = next.predecessor;
243        finger.replace_state(&next.fingers, next.fix_finger_index);
244        Ok(())
245    }
246
247    fn topology_action(&self, action: TopologyAction) -> PeerRingAction {
248        match action {
249            TopologyAction::FindSuccessorForConnect { next, did } => {
250                PeerRingAction::RemoteAction(next, RemoteAction::FindSuccessorForConnect(did))
251            }
252            TopologyAction::FindSuccessorForFix { next, did, index } => {
253                PeerRingAction::RemoteAction(next, RemoteAction::FindSuccessorForFix { did, index })
254            }
255            TopologyAction::QuerySuccessorList(did) => {
256                PeerRingAction::RemoteAction(did, RemoteAction::QueryForSuccessorList)
257            }
258            TopologyAction::Notify(did) => {
259                PeerRingAction::RemoteAction(did, RemoteAction::Notify(self.did))
260            }
261        }
262    }
263
264    fn topology_leaf_actions(&self, actions: Vec<TopologyAction>) -> PeerRingAction {
265        let mut actions = actions
266            .into_iter()
267            .map(|action| self.topology_action(action))
268            .collect::<Vec<_>>();
269        match actions.len() {
270            0 => PeerRingAction::None,
271            1 => actions.pop().unwrap_or(PeerRingAction::None),
272            _ => PeerRingAction::MultiActions(actions),
273        }
274    }
275
276    fn topology_multi_actions(&self, actions: Vec<TopologyAction>) -> PeerRingAction {
277        PeerRingAction::MultiActions(
278            actions
279                .into_iter()
280                .map(|action| self.topology_action(action))
281                .collect(),
282        )
283    }
284
285    pub(crate) fn apply_fixed_finger(&self, index: usize, successor: Did) -> Result<()> {
286        self.transition_topology(TopologyEvent::ApplyFinger { index, successor })
287            .map(|_| ())
288    }
289
290    pub(crate) fn admit_connected(
291        &self,
292        peer: Did,
293        fixed_fingers: Vec<topology::ConditionalFingerUpdate>,
294    ) -> Result<PeerRingAction> {
295        let next = self.transition_topology(TopologyEvent::Admit {
296            peer,
297            fixed_fingers,
298        })?;
299        Ok(self.topology_multi_actions(next.actions))
300    }
301}
302
303impl Chord<PeerRingAction> for PeerRing {
304    fn join(&self, did: Did) -> Result<PeerRingAction> {
305        let next = self.transition_topology(TopologyEvent::Join { peer: did })?;
306        Ok(self.topology_leaf_actions(next.actions))
307    }
308
309    fn find_successor(&self, did: Did) -> Result<PeerRingAction> {
310        let state = self.topology_state()?;
311        let result = match topology::find_successor(&state, did) {
312            FindSuccessorStep::Local(successor) => Ok(PeerRingAction::Some(successor)),
313            FindSuccessorStep::Remote { next, did } => Ok(PeerRingAction::RemoteAction(
314                next,
315                RemoteAction::FindSuccessor(did),
316            )),
317        };
318
319        tracing::debug!(
320            "find_successor: self: {}, did: {}, successor: {:?}, result: {:?}",
321            self.did,
322            did,
323            state.successors,
324            result
325        );
326        result
327    }
328
329    fn notify(&self, did: Did) -> Result<Did> {
330        let next = self.transition_topology(TopologyEvent::Notify { predecessor: did })?;
331        next.state.predecessor.ok_or(Error::PeerRingInvalidAction)
332    }
333
334    fn fix_fingers(&self) -> Result<PeerRingAction> {
335        let next = self.transition_topology(TopologyEvent::FixFinger)?;
336        Ok(self.topology_leaf_actions(next.actions))
337    }
338}
339
340#[cfg_attr(all(feature = "wasm", target_family = "wasm"), async_trait(?Send))]
341#[cfg_attr(not(all(feature = "wasm", target_family = "wasm")), async_trait)]
342impl CorrectChord<PeerRingAction> for PeerRing {
343    async fn update_successor(&self, did: impl LiveDid) -> Result<PeerRingAction> {
344        if !did.live().await {
345            return Ok(PeerRingAction::RemoteAction(
346                did.into(),
347                RemoteAction::TryConnect,
348            ));
349        }
350        let next = self.transition_topology(TopologyEvent::UpdateSuccessor {
351            successor: did.into(),
352        })?;
353        Ok(self.topology_leaf_actions(next.actions))
354    }
355
356    async fn extend_successor(&self, dids: &[impl LiveDid]) -> Result<PeerRingAction> {
357        let mut actions = vec![];
358        for did in dids {
359            if let PeerRingAction::RemoteAction(recipient, action) =
360                self.update_successor(did.clone()).await?
361            {
362                actions.push(PeerRingAction::RemoteAction(recipient, action));
363            }
364        }
365        Ok(PeerRingAction::MultiActions(actions))
366    }
367
368    async fn join_then_sync(&self, did: impl LiveDid) -> Result<PeerRingAction> {
369        if !did.live().await {
370            return Ok(PeerRingAction::None);
371        }
372        self.admit_connected(did.into(), Vec::new())
373    }
374
375    fn rectify(&self, pred: Did) -> Result<()> {
376        self.transition_topology(TopologyEvent::Notify { predecessor: pred })
377            .map(|_| ())
378    }
379
380    fn pre_stabilize(&self) -> Result<PeerRingAction> {
381        let successor = self.successors();
382        if successor.is_empty()? {
383            return Ok(PeerRingAction::None);
384        }
385        let head = successor.min()?;
386        Ok(PeerRingAction::RemoteAction(
387            head,
388            RemoteAction::QueryForSuccessorListAndPred,
389        ))
390    }
391
392    fn stabilize(&self, info: TopoInfo) -> Result<PeerRingAction> {
393        let next = self.transition_topology(TopologyEvent::Stabilize {
394            successors: info.successors,
395            predecessor: info.predecessor,
396        })?;
397        Ok(self.topology_multi_actions(next.actions))
398    }
399
400    fn topo_info(&self) -> Result<TopoInfo> {
401        self.try_into()
402    }
403}
rings_core/dht/chord/peer_ring.rs

rings_core/dht/chord/
peer_ring.rs
