rings_core/dht/chord/
storage.rs

1use async_trait::async_trait;
2
3use super::PeerRing;
4use super::PeerRingAction;
5use super::RemoteAction;
6use crate::dht::entry::Entry;
7use crate::dht::entry::EntryLookupEvidence;
8use crate::dht::entry::EntryLookupKey;
9use crate::dht::entry::EntryOperation;
10use crate::dht::entry::PlacedEntryOperation;
11use crate::dht::entry::PlacementMiss;
12use crate::dht::types::ChordStorage;
13use crate::dht::types::ChordStorageCache;
14use crate::dht::Did;
15use crate::error::Error;
16use crate::error::Result;
17
18impl PeerRing {
19    /// Join an incoming replicated entry delta into local storage.
20    ///
21    /// Post: the stored value is the least upper bound of the previous local
22    /// value and `incoming` when a previous value exists; otherwise it is
23    /// `incoming` normalized for storage.
24    pub(crate) async fn join_storage_entry(&self, key: Did, incoming: Entry) -> Result<Entry> {
25        let incoming = incoming.try_into_storage_entry()?;
26        let stored = if let Some(local) = self.storage.get(&key.to_string()).await? {
27            local.join(incoming)?
28        } else {
29            incoming
30        }
31        .try_into_storage_entry()?;
32        self.storage.put(&key.to_string(), &stored).await?;
33        Ok(stored)
34    }
35
36    fn storage_fetch_fallback_successor(&self) -> Result<Option<Did>> {
37        Ok(self
38            .topology_state()?
39            .successors
40            .into_iter()
41            .find(|successor| *successor != self.did))
42    }
43
44    async fn entry_lookup_inner<const REDUNDANT: u16>(
45        &self,
46        entry_key: Did,
47        fallback_on_local_virtual_miss: bool,
48    ) -> Result<PeerRingAction> {
49        let mut ret = vec![];
50        let mut misses = vec![];
51        for placement_key in entry_key.rotate_affine(REDUNDANT)? {
52            let query = EntryLookupKey::new(entry_key, placement_key);
53            let act = match self.find_storage_owner(placement_key) {
54                Ok(PeerRingAction::Some(succ)) => {
55                    match self.storage.get(&placement_key.to_string()).await {
56                        Ok(Some(value)) => {
57                            let observed_misses = std::mem::take(&mut misses);
58                            Ok(PeerRingAction::SomeEntry(EntryLookupEvidence::new(
59                                value,
60                                observed_misses,
61                            )))
62                        }
63                        Ok(None) => {
64                            tracing::debug!(
65                                "Cannot find entry in local storage, try to query from successor"
66                            );
67                            if succ == self.did {
68                                if fallback_on_local_virtual_miss
69                                    && self.storage_virtual_nodes_enabled()?
70                                {
71                                    if let Some(next) = self.storage_fetch_fallback_successor()? {
72                                        Ok(PeerRingAction::RemoteAction(
73                                            next,
74                                            RemoteAction::FindEntry(query),
75                                        ))
76                                    } else {
77                                        misses.push(PlacementMiss::new(placement_key, succ));
78                                        Ok(PeerRingAction::None)
79                                    }
80                                } else {
81                                    misses.push(PlacementMiss::new(placement_key, succ));
82                                    Ok(PeerRingAction::None)
83                                }
84                            } else {
85                                Ok(PeerRingAction::RemoteAction(
86                                    succ,
87                                    RemoteAction::FindEntry(query),
88                                ))
89                            }
90                        }
91                        Err(error) => Err(error),
92                    }
93                }
94                Ok(PeerRingAction::RemoteAction(next, RemoteAction::FindSuccessor(id))) => {
95                    Ok(PeerRingAction::RemoteAction(
96                        next,
97                        RemoteAction::FindEntry(EntryLookupKey::new(entry_key, id)),
98                    ))
99                }
100                Ok(action) => Err(Error::unexpected_peer_ring_action(action)),
101                Err(error) => Err(error),
102            }?;
103            if act.is_remote() {
104                ret.push(act);
105            } else if act.is_some_entry() {
106                return Ok(act);
107            }
108        }
109        if !misses.is_empty() {
110            ret.push(PeerRingAction::EntryMisses(misses));
111        }
112        Ok(ret.into())
113    }
114
115    /// Look up an [`Entry`] for a local storage fetch.
116    ///
117    /// A fresh node with storage virtual nodes enabled can observe itself as the
118    /// owner for an existing placement before sync has copied historical data
119    /// locally. Local fetches may ask a known successor for that placement so
120    /// read repair can converge instead of treating the fresh local miss as
121    /// authoritative. Remote `SearchEntry` handling uses [`ChordStorage`] and
122    /// intentionally does not enable this fallback.
123    pub(crate) async fn entry_lookup_for_fetch<const REDUNDANT: u16>(
124        &self,
125        entry_key: Did,
126    ) -> Result<PeerRingAction> {
127        self.entry_lookup_inner::<REDUNDANT>(entry_key, true).await
128    }
129}
130
131#[cfg_attr(all(feature = "wasm", target_family = "wasm"), async_trait(?Send))]
132#[cfg_attr(not(all(feature = "wasm", target_family = "wasm")), async_trait)]
133impl<const REDUNDANT: u16> ChordStorage<PeerRingAction, REDUNDANT> for PeerRing {
134    async fn entry_lookup(&self, entry_key: Did) -> Result<PeerRingAction> {
135        self.entry_lookup_inner::<REDUNDANT>(entry_key, false).await
136    }
137
138    async fn entry_operate(&self, op: EntryOperation) -> Result<PeerRingAction> {
139        let op = op.stamped(self.did)?;
140        let entry_key = op.did()?;
141        let mut ret = vec![];
142        for entry_key in entry_key.rotate_affine(REDUNDANT)? {
143            let act = match self.find_storage_owner(entry_key) {
144                Ok(PeerRingAction::Some(_)) => {
145                    let this = match self.storage.get(&entry_key.to_string()).await? {
146                        Some(this) => this,
147                        None => op.clone().gen_default_entry()?,
148                    };
149                    let entry = this.operate(op.clone(), self.did)?;
150                    self.join_storage_entry(entry_key, entry).await?;
151                    Ok(PeerRingAction::None)
152                }
153                Ok(PeerRingAction::RemoteAction(next, RemoteAction::FindSuccessor(_))) => {
154                    Ok(PeerRingAction::RemoteAction(
155                        next,
156                        RemoteAction::FindEntryForOperate(PlacedEntryOperation {
157                            placement: entry_key,
158                            op: op.clone(),
159                        }),
160                    ))
161                }
162                Ok(action) => Err(Error::unexpected_peer_ring_action(action)),
163                Err(error) => Err(error),
164            }?;
165            if act.is_remote() {
166                ret.push(act);
167            }
168        }
169        Ok(ret.into())
170    }
171}
172
173#[cfg_attr(all(feature = "wasm", target_family = "wasm"), async_trait(?Send))]
174#[cfg_attr(not(all(feature = "wasm", target_family = "wasm")), async_trait)]
175impl ChordStorageCache<PeerRingAction> for PeerRing {
176    async fn local_cache_put(&self, entry: Entry) -> Result<()> {
177        self.cache.put(&entry.did.to_string(), &entry).await
178    }
179
180    async fn local_cache_get(&self, entry_key: Did) -> Result<Option<Entry>> {
181        self.cache.get(&entry_key.to_string()).await
182    }
183}
rings_core/dht/chord/storage.rs

rings_core/dht/chord/
storage.rs
