rings_core/dht/
entry.rs

1#![deny(missing_docs)]
2use std::collections::BTreeMap;
3use std::collections::BTreeSet;
4use std::str::FromStr;
5
6use serde::Deserialize;
7use serde::Serialize;
8
9use crate::algebra::JoinSemilattice;
10use crate::consts::ENTRY_DATA_MAX_LEN;
11use crate::dht::Did;
12use crate::ecc::HashStr;
13use crate::error::Error;
14use crate::error::Result;
15use crate::message::Encoded;
16use crate::message::Encoder;
17use crate::message::MessagePayload;
18use crate::message::MessageVerificationExt;
19
20mod crdt;
21
22pub use crdt::DataTopicBuffer;
23pub use crdt::EntryCrdt;
24pub use crdt::EntryDot;
25pub use crdt::EntryVersion;
26pub use crdt::RelayMessageSet;
27
28/// DHT storage entry categories.
29#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
30pub enum EntryKind {
31    /// Encoded data stored in DHT
32    Data,
33    /// A relayed but unreached message, which should be stored on
34    /// the successor of the destination Did.
35    RelayMessage,
36}
37
38#[derive(Clone, Debug, PartialEq, Eq)]
39enum EntryStampKind {
40    Overwrite,
41    Delta,
42}
43
44// Canonical stamp input for EntryVersion.operation.
45//
46// This digest is an unreleased CRDT tie-break witness between nodes running the
47// same code, not a stable storage key or cross-version protocol identifier.
48#[derive(Serialize)]
49struct OperationDigest<'a> {
50    kind: EntryKind,
51    did: Did,
52    data: &'a [Encoded],
53}
54
55/// Operations supported by a DHT storage entry.
56#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
57pub enum EntryOperation {
58    /// Create or update an [`Entry`].
59    Overwrite(Entry),
60    /// Extend data to a Data kind [`Entry`].
61    /// This operation will create an [`Entry`] if it does not exist.
62    Extend(Entry),
63    /// Extend data to a Data kind [`Entry`] uniquely.
64    /// If any element is already existed, move it to the end of the data vector.
65    /// This operation will create an [`Entry`] if it does not exist.
66    Touch(Entry),
67    /// Tombstone observed data or relay-message payloads in a two-phase set.
68    ///
69    /// The payload identifies the entry carrier and the values to
70    /// remove. If CRDT dots are present, those dots are the remove witnesses;
71    /// otherwise the receiver tombstones currently observed dots with matching
72    /// payload bytes.
73    Tombstone(Entry),
74    /// Compact a Data kind entry after removing listed payload bytes.
75    ///
76    /// The receiver computes the compacted live set from its current local
77    /// entry, not from a sender snapshot. This preserves concurrent live writes
78    /// already observed by the storage owner. The operation carries one
79    /// source-stamped register floor shared by every replica, so divergent
80    /// storage owners stay join-compatible after compaction.
81    CompactData(Entry),
82}
83
84/// A storage operation targeted at one concrete affine placement key.
85///
86/// Invariant: `placement` must be one of the affine replica keys derived from
87/// the operation's entry DID under the receiver's configured storage
88/// redundancy. The sender may choose a replica from that set, but cannot choose
89/// where the replica set itself lives.
90#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
91pub struct PlacedEntryOperation {
92    /// Placement key that must receive the operation.
93    pub placement: Did,
94    /// Operation to apply at `placement`.
95    pub op: EntryOperation,
96}
97
98impl PlacedEntryOperation {
99    /// Return the entry identity carried by this operation.
100    pub fn entry_key(&self) -> Result<Did> {
101        self.op.did()
102    }
103
104    /// Return whether `placement` is in this entry's affine replica set.
105    pub fn placement_belongs_to_entry(&self, redundancy: u16) -> Result<bool> {
106        let entry_key = self.entry_key()?;
107        placement_belongs_to_entry_key(entry_key, self.placement, redundancy)
108    }
109
110    /// Enforce that `placement` belongs to the operation's entry.
111    pub fn validate_placement(&self, redundancy: u16) -> Result<()> {
112        if self.placement_belongs_to_entry(redundancy)? {
113            return Ok(());
114        }
115
116        Err(Error::InvalidMessage(
117            "placed entry operation targets a placement outside the entry's affine replica set"
118                .to_string(),
119        ))
120    }
121}
122
123fn placement_belongs_to_entry_key(entry_key: Did, placement: Did, redundancy: u16) -> Result<bool> {
124    Ok(entry_key.rotate_affine(redundancy)?.contains(&placement))
125}
126
127/// A DHT storage entry with an [`EntryKind`] and a ring key represented as [`Did`].
128///
129/// An [`Entry`] is data stored by [`ChordStorage`](super::ChordStorage). It is not a
130/// Chord node and does not participate in successor, predecessor, or finger-table
131/// membership.
132///
133/// The [`Did`] of an [`Entry`] is in the following format:
134/// * If kind value is [EntryKind::Data], it's sha1 of data topic.
135/// * If kind value is [EntryKind::RelayMessage], it's the destination Did of
136///   message plus 1 (to ensure that the message is sent to the successor of destination),
137///   thus while destination node going online, it will sync message from its successor.
138#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
139pub struct Entry {
140    /// The ring key of this entry. It has the same representation as a node DID, but a
141    /// different domain meaning.
142    pub did: Did,
143    /// The data entity of `Entry`, encoded by [Encoder].
144    pub data: Vec<Encoded>,
145    /// The type indicates how the data is encoded and how the Did is generated.
146    pub kind: EntryKind,
147    /// CRDT metadata that makes replicated merge a join-semilattice operation.
148    #[serde(default)]
149    pub crdt: EntryCrdt,
150}
151
152/// An [`Entry`] paired with its Chord placement key.
153///
154/// `key` is the DHT storage location. `entry.did` is the resource identity. These two
155/// values may differ for redundant replicas.
156#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
157pub struct PlacedEntry {
158    /// The key used to place this value in DHT storage.
159    pub key: Did,
160    /// The stored entry value.
161    pub entry: Entry,
162}
163
164impl PlacedEntry {
165    /// Pair an entry value with the key where it is stored.
166    pub fn new(key: Did, entry: Entry) -> Self {
167        Self { key, entry }
168    }
169
170    /// Return whether `key` is in `entry.did`'s affine replica set.
171    pub fn placement_belongs_to_entry(&self, redundancy: u16) -> Result<bool> {
172        placement_belongs_to_entry_key(self.entry.did, self.key, redundancy)
173    }
174
175    /// Enforce that `key` belongs to `entry.did`'s affine replica set.
176    pub fn validate_placement(&self, redundancy: u16) -> Result<()> {
177        if self.placement_belongs_to_entry(redundancy)? {
178            return Ok(());
179        }
180
181        Err(Error::InvalidMessage(
182            "synced placed entry targets a placement outside the entry's affine replica set"
183                .to_string(),
184        ))
185    }
186}
187
188/// Durable-storage acknowledgement for an entry hand-off delta.
189///
190/// `key` is the placement key updated by the receiver. `entry` is the copied
191/// delta that the receiver joined into its local least upper bound. The sender
192/// compares the storage-normalized ack value with its current local value
193/// before deleting; if the sender has observed any newer durable delta
194/// meanwhile, deletion is skipped.
195#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
196pub struct SyncedEntryAck {
197    /// The placement key durably persisted by the sync receiver.
198    pub key: Did,
199    /// The exact value durably persisted by the sync receiver.
200    pub entry: Entry,
201}
202
203impl SyncedEntryAck {
204    /// Witness that `entry` was durably joined at `key`.
205    pub fn new(key: Did, entry: Entry) -> Self {
206        Self { key, entry }
207    }
208
209    /// Returns whether this ack proves that `local` equals the copied value.
210    ///
211    /// Post: comparison is performed on storage canonical forms, so legacy
212    /// entries without dots compare equal to the normalized value durably
213    /// persisted by the receiver.
214    pub fn confirms_local_value(&self, local: &Entry) -> Result<bool> {
215        Ok(self.entry.clone().try_into_storage_entry()?
216            == local.clone().try_into_storage_entry()?)
217    }
218}
219
220/// A lookup request for a concrete placement of an entry identity.
221///
222/// `resource` is `id(e)`. `placement` is one element of
223/// `place(resource, REDUNDANT)`.
224#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
225pub struct EntryLookupKey {
226    /// Entry identity being searched.
227    pub resource: Did,
228    /// Placement key being interrogated.
229    pub placement: Did,
230}
231
232impl EntryLookupKey {
233    /// Pair an entry identity with one of its placement keys.
234    pub fn new(resource: Did, placement: Did) -> Self {
235        Self {
236            resource,
237            placement,
238        }
239    }
240}
241
242/// A placement key observed missing during lookup.
243#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
244pub struct PlacementMiss {
245    /// Placement key whose responsible owner returned `None`.
246    pub key: Did,
247    /// Owner that was responsible for `key` when the miss was observed.
248    pub owner: Did,
249}
250
251impl PlacementMiss {
252    /// Witness that `owner` was queried for `key` and did not have the entry.
253    pub fn new(key: Did, owner: Did) -> Self {
254        Self { key, owner }
255    }
256}
257
258/// A successful lookup result plus the missing placements observed before it.
259#[derive(Clone, Debug, PartialEq, Eq)]
260pub struct EntryLookupEvidence {
261    /// Entry found by the lookup.
262    pub entry: Entry,
263    /// Placement misses observed as part of the same lookup.
264    pub misses: Vec<PlacementMiss>,
265}
266
267impl EntryLookupEvidence {
268    /// Construct lookup evidence.
269    pub fn new(entry: Entry, misses: Vec<PlacementMiss>) -> Self {
270        Self { entry, misses }
271    }
272}
273
274impl Entry {
275    /// Construct an entry with empty CRDT metadata.
276    pub fn new(did: Did, data: Vec<Encoded>, kind: EntryKind) -> Self {
277        Self {
278            did,
279            data,
280            kind,
281            crdt: EntryCrdt::default(),
282        }
283    }
284
285    /// Generate did from topic.
286    pub fn gen_did(topic: &str) -> Result<Did> {
287        let hash: HashStr = topic.into();
288        let did = Did::from_str(&hash.inner());
289        tracing::debug!("gen_did: topic: {}, did: {:?}", topic, did);
290        did
291    }
292}
293
294impl EntryOperation {
295    /// Return this operation with CRDT versions assigned at the operation boundary.
296    ///
297    /// Existing CRDT witnesses are preserved so forwarded operations keep the
298    /// origin's dot/version instead of being reissued by every routing hop.
299    pub fn stamped(self, actor: Did) -> Result<Self> {
300        Ok(match self {
301            EntryOperation::Overwrite(entry) => EntryOperation::Overwrite(
302                entry.ensure_stamp_after(actor, None, EntryStampKind::Overwrite)?,
303            ),
304            EntryOperation::Extend(entry) => EntryOperation::Extend(entry.ensure_stamp_after(
305                actor,
306                None,
307                EntryStampKind::Delta,
308            )?),
309            EntryOperation::Touch(entry) => EntryOperation::Touch(entry.ensure_stamp_after(
310                actor,
311                None,
312                EntryStampKind::Delta,
313            )?),
314            EntryOperation::Tombstone(entry) => EntryOperation::Tombstone(entry),
315            EntryOperation::CompactData(entry) => {
316                EntryOperation::CompactData(entry.ensure_overwrite_stamp_after(actor, None)?)
317            }
318        })
319    }
320
321    /// Extract the did of target Entry.
322    pub fn did(&self) -> Result<Did> {
323        Ok(match self {
324            EntryOperation::Overwrite(entry) => entry.did,
325            EntryOperation::Extend(entry) => entry.did,
326            EntryOperation::Touch(entry) => entry.did,
327            EntryOperation::Tombstone(entry) => entry.did,
328            EntryOperation::CompactData(entry) => entry.did,
329        })
330    }
331
332    /// Extract the kind of target Entry.
333    pub fn kind(&self) -> EntryKind {
334        match self {
335            EntryOperation::Overwrite(entry) => entry.kind,
336            EntryOperation::Extend(entry) => entry.kind,
337            EntryOperation::Touch(entry) => entry.kind,
338            EntryOperation::Tombstone(entry) => entry.kind,
339            EntryOperation::CompactData(entry) => entry.kind,
340        }
341    }
342
343    /// Generate a target Entry when it is not existed.
344    pub fn gen_default_entry(self) -> Result<Entry> {
345        Ok(Entry::new(self.did()?, vec![], self.kind()))
346    }
347}
348
349impl TryFrom<MessagePayload> for Entry {
350    type Error = Error;
351    fn try_from(msg: MessagePayload) -> Result<Self> {
352        // Relay entries target the signer's successor on R = Z / 2^160, so the
353        // `+ 1` intentionally wraps in the fixed-width DID ring.
354        let did = msg.signer() + Did::from(1u32);
355        let data = msg.encode()?;
356        Ok(Self {
357            did,
358            data: vec![data],
359            kind: EntryKind::RelayMessage,
360            crdt: EntryCrdt::default(),
361        })
362    }
363}
364
365impl TryFrom<(String, Encoded)> for Entry {
366    type Error = Error;
367    fn try_from((topic, e): (String, Encoded)) -> Result<Self> {
368        Ok(Self {
369            did: Self::gen_did(&topic)?,
370            data: vec![e],
371            kind: EntryKind::Data,
372            crdt: EntryCrdt::default(),
373        })
374    }
375}
376
377impl TryFrom<(String, String)> for Entry {
378    type Error = Error;
379    fn try_from((topic, s): (String, String)) -> Result<Self> {
380        let encoded_message = s.encode()?;
381        (topic, encoded_message).try_into()
382    }
383}
384
385impl TryFrom<String> for Entry {
386    type Error = Error;
387    fn try_from(topic: String) -> Result<Self> {
388        (topic.clone(), topic).try_into()
389    }
390}
391
392impl Entry {
393    fn with_element_dots(mut self, version: EntryVersion) -> Result<Self> {
394        self.crdt.dots = self
395            .data
396            .iter()
397            .enumerate()
398            .map(|(index, _)| EntryDot::for_index(version, index))
399            .collect::<Result<Vec<_>>>()?;
400        Ok(self)
401    }
402
403    fn stamp_overwrite(mut self, version: EntryVersion) -> Result<Self> {
404        self.crdt.register = Some(version);
405        self.with_element_dots(version)
406    }
407
408    fn stamp_delta(self, version: EntryVersion) -> Result<Self> {
409        self.with_element_dots(version)
410    }
411
412    fn stamp(self, version: EntryVersion, kind: EntryStampKind) -> Result<Self> {
413        match kind {
414            EntryStampKind::Overwrite => self.stamp_overwrite(version),
415            EntryStampKind::Delta => self.stamp_delta(version),
416        }
417    }
418
419    fn operation_digest(&self) -> Result<Did> {
420        let digest = OperationDigest {
421            kind: self.kind,
422            did: self.did,
423            data: &self.data,
424        };
425        let bytes = rings_codec::serialize(&digest).map_err(Error::CodecSerialize)?;
426        Did::try_from(HashStr::from_bytes(&bytes))
427    }
428
429    fn issue_version_after(&self, actor: Did, floor: Option<EntryVersion>) -> Result<EntryVersion> {
430        Ok(EntryVersion::issued_by(actor, self.operation_digest()?).after(floor))
431    }
432
433    fn ensure_stamp_after(
434        self,
435        actor: Did,
436        floor: Option<EntryVersion>,
437        kind: EntryStampKind,
438    ) -> Result<Self> {
439        match self.crdt.has_write_witness() {
440            true => Ok(self),
441            false => {
442                let version = self.issue_version_after(actor, floor)?;
443                self.stamp(version, kind)
444            }
445        }
446    }
447
448    fn ensure_overwrite_stamp_after(self, actor: Did, floor: Option<EntryVersion>) -> Result<Self> {
449        match self.crdt.register.is_some() {
450            true => Ok(self),
451            false => {
452                let version = self.issue_version_after(actor, floor)?;
453                self.stamp_overwrite(version)
454            }
455        }
456    }
457
458    fn max_observed_version(&self) -> Option<EntryVersion> {
459        self.crdt
460            .dots
461            .iter()
462            .map(|dot| dot.version)
463            .chain(self.crdt.tombstones.iter().map(|dot| dot.version))
464            .chain(self.crdt.register)
465            .max()
466    }
467
468    fn validate_same_carrier(&self, other: &Self) -> Result<()> {
469        if !self.same_kind_as(other) {
470            return Err(Error::EntryKindNotEqual);
471        }
472        if !self.same_key_as(other) {
473            return Err(Error::EntryDidNotEqual);
474        }
475        Ok(())
476    }
477
478    fn dot_for_element(&self, index: usize) -> Result<EntryDot> {
479        if let Some(dot) = self.crdt.dots.get(index).copied() {
480            return Ok(dot);
481        }
482        EntryDot::for_index(self.crdt.legacy_floor(), index)
483    }
484
485    fn topic_buffer(&self) -> Result<DataTopicBuffer> {
486        let mut values = BTreeMap::new();
487        for (index, value) in self.data.iter().cloned().enumerate() {
488            let dot = self.dot_for_element(index)?;
489            values
490                .entry(value)
491                .and_modify(|current: &mut EntryDot| {
492                    *current = (*current).max(dot);
493                })
494                .or_insert(dot);
495        }
496        Ok(DataTopicBuffer::new(
497            self.crdt.register,
498            values,
499            self.crdt.tombstones.iter().copied().collect(),
500        ))
501    }
502
503    fn relay_set(&self) -> Result<RelayMessageSet> {
504        Ok(RelayMessageSet::new(
505            self.topic_buffer()?,
506            self.crdt.tombstones.iter().copied().collect(),
507        ))
508    }
509
510    fn materialize_elements(
511        did: Did,
512        kind: EntryKind,
513        register: Option<EntryVersion>,
514        elements: impl IntoIterator<Item = (Encoded, EntryDot)>,
515        tombstones: BTreeSet<EntryDot>,
516    ) -> Self {
517        let mut visible = elements
518            .into_iter()
519            .filter(|(_, dot)| {
520                let visible_after_reset = register.is_none_or(|floor| dot.version >= floor);
521                visible_after_reset && !tombstones.contains(dot)
522            })
523            .collect::<Vec<_>>();
524        visible.sort_by(|(left_value, left_dot), (right_value, right_dot)| {
525            left_dot
526                .cmp(right_dot)
527                .then_with(|| left_value.cmp(right_value))
528        });
529        let skip_count = visible.len().saturating_sub(ENTRY_DATA_MAX_LEN);
530        let visible = visible.into_iter().skip(skip_count).collect::<Vec<_>>();
531        let (data, dots): (Vec<_>, Vec<_>) = visible.into_iter().unzip();
532
533        Self {
534            did,
535            data,
536            kind,
537            crdt: EntryCrdt {
538                register,
539                dots,
540                tombstones: tombstones.into_iter().collect(),
541            },
542        }
543    }
544
545    fn materialize_topic_buffer(&self, buffer: DataTopicBuffer) -> Self {
546        Self::materialize_elements(
547            self.did,
548            self.kind,
549            buffer.register,
550            buffer.values,
551            buffer.removes,
552        )
553    }
554
555    fn materialize_relay_set(&self, set: RelayMessageSet) -> Self {
556        Self::materialize_elements(
557            self.did,
558            self.kind,
559            set.adds.register,
560            set.adds.values,
561            set.removes,
562        )
563    }
564
565    fn compacted_data_dot(floor: EntryVersion, value: &Encoded) -> Result<EntryDot> {
566        let operation = Did::try_from(HashStr::from_bytes(value.value().as_bytes()))?;
567        let version =
568            EntryVersion::new(floor.logical_time_ms, floor.actor, operation).after(Some(floor));
569        EntryDot::for_index(version, 0)
570    }
571
572    fn compact_data_element(
573        floor: EntryVersion,
574        removal_values: &BTreeSet<Encoded>,
575        value: Encoded,
576        dot: EntryDot,
577    ) -> Result<Option<(Encoded, EntryDot)>> {
578        match dot.version < floor {
579            true if removal_values.contains(&value) => Ok(None),
580            true => Self::compacted_data_dot(floor, &value).map(|dot| Some((value, dot))),
581            false => Ok(Some((value, dot))),
582        }
583    }
584
585    fn data_compaction_candidates(
586        payload_order: &[Encoded],
587        live_values: BTreeMap<Encoded, EntryDot>,
588    ) -> Vec<(Encoded, EntryDot)> {
589        let (ordered_values, remaining_values) = payload_order.iter().fold(
590            (Vec::new(), live_values),
591            |(mut ordered, mut remaining), value| {
592                if let Some(dot) = remaining.remove(value) {
593                    ordered.push((value.clone(), dot));
594                }
595                (ordered, remaining)
596            },
597        );
598        ordered_values.into_iter().chain(remaining_values).collect()
599    }
600
601    fn compact_data_elements(
602        floor: EntryVersion,
603        removal_values: &BTreeSet<Encoded>,
604        values: impl IntoIterator<Item = (Encoded, EntryDot)>,
605    ) -> Result<Vec<(Encoded, EntryDot)>> {
606        values.into_iter().try_fold(
607            Vec::new(),
608            |mut elements, (value, dot)| -> Result<Vec<(Encoded, EntryDot)>> {
609                match Self::compact_data_element(floor, removal_values, value, dot)? {
610                    Some(element) => {
611                        elements.push(element);
612                        Ok(elements)
613                    }
614                    None => Ok(elements),
615                }
616            },
617        )
618    }
619
620    fn compact_data_output_floor(
621        current_floor: Option<EntryVersion>,
622        operation_floor: EntryVersion,
623    ) -> EntryVersion {
624        current_floor.map_or(operation_floor, |current| current.max(operation_floor))
625    }
626
627    fn compact_data_tombstones(
628        floor: EntryVersion,
629        tombstones: BTreeSet<EntryDot>,
630    ) -> BTreeSet<EntryDot> {
631        tombstones
632            .into_iter()
633            .filter(|dot| dot.version >= floor)
634            .collect()
635    }
636
637    /// Merge two entries from the same replicated carrier.
638    ///
639    /// Law: for a fixed `(did, kind)` carrier, this is the state-based CRDT
640    /// join. Data entries are bounded LWW element sets with an LWW overwrite
641    /// register; relay entries are two-phase sets whose remove side is carried
642    /// by tombstones.
643    pub fn join(&self, other: Self) -> Result<Self> {
644        self.validate_same_carrier(&other)?;
645        match self.kind {
646            EntryKind::Data => {
647                Ok(self.materialize_topic_buffer(self.topic_buffer()?.join(other.topic_buffer()?)))
648            }
649            EntryKind::RelayMessage => {
650                Ok(self.materialize_relay_set(self.relay_set()?.join(other.relay_set()?)))
651            }
652        }
653    }
654
655    /// Affine Transport entry to a list of affined did
656    pub fn affine(&self, scalar: u16) -> Result<Vec<Entry>> {
657        Ok(self
658            .did
659            .rotate_affine(scalar)?
660            .into_iter()
661            .map(|did| self.clone_with_did(did))
662            .collect())
663    }
664
665    /// Clone and setup with new DID
666    pub fn clone_with_did(&self, did: Did) -> Self {
667        let mut entry = self.clone();
668        entry.did = did;
669        entry
670    }
671
672    fn is_data_entry(&self) -> bool {
673        self.kind == EntryKind::Data
674    }
675
676    fn same_kind_as(&self, other: &Self) -> bool {
677        self.kind == other.kind
678    }
679
680    fn same_key_as(&self, other: &Self) -> bool {
681        self.did == other.did
682    }
683
684    /// Normalize an entry immediately before it is persisted.
685    ///
686    /// Post: normalization uses the same carrier materialization as
687    /// [`Self::join`]; there is no second cap strategy outside the CRDT.
688    /// Post: `result.data.len() <= ENTRY_DATA_MAX_LEN`.
689    /// Post: `result.data.len() == result.crdt.dots.len()` for Data and
690    /// RelayMessage entries.
691    pub fn try_into_storage_entry(self) -> Result<Self> {
692        match self.kind {
693            EntryKind::Data => {
694                let buffer = self.topic_buffer()?;
695                Ok(self.materialize_topic_buffer(buffer))
696            }
697            EntryKind::RelayMessage => {
698                let set = self.relay_set()?;
699                Ok(self.materialize_relay_set(set))
700            }
701        }
702    }
703
704    /// The entry point of [EntryOperation].
705    /// Will dispatch to different operation handlers according to the variant.
706    pub fn operate(&self, op: EntryOperation, actor: Did) -> Result<Self> {
707        match op {
708            EntryOperation::Overwrite(entry) => self.overwrite(entry, actor),
709            EntryOperation::Extend(entry) => self.extend(entry, actor),
710            EntryOperation::Touch(entry) => self.touch(entry, actor),
711            EntryOperation::Tombstone(entry) => self.tombstone(entry),
712            EntryOperation::CompactData(entry) => self.compact_data(entry, actor),
713        }
714    }
715
716    /// Overwrite current data with new data.
717    ///
718    /// Preservation: the replacement is represented as a CRDT join. A newly
719    /// stamped overwrite carries a reset floor, and materialization keeps only
720    /// dots at or after that floor, so older payload dots are removed without a
721    /// non-monotone assignment.
722    ///
723    /// The handler of [EntryOperation::Overwrite].
724    pub fn overwrite(&self, other: Self, actor: Did) -> Result<Self> {
725        if !self.is_data_entry() {
726            return Err(Error::EntryNotOverwritable);
727        }
728        self.join(other.ensure_stamp_after(
729            actor,
730            self.max_observed_version(),
731            EntryStampKind::Overwrite,
732        )?)
733    }
734
735    /// This method is used to extend data to a Data kind [`Entry`].
736    /// The handler of [EntryOperation::Extend].
737    pub fn extend(&self, other: Self, actor: Did) -> Result<Self> {
738        if !self.is_data_entry() {
739            return Err(Error::EntryNotAppendable);
740        }
741        self.join(other.ensure_stamp_after(
742            actor,
743            self.max_observed_version(),
744            EntryStampKind::Delta,
745        )?)
746    }
747
748    /// This method is used to extend data to a Data kind [`Entry`] uniquely.
749    /// If any element is already existed, move it to the end of the data vector.
750    /// The handler of [EntryOperation::Touch].
751    pub fn touch(&self, other: Self, actor: Did) -> Result<Self> {
752        if !self.is_data_entry() {
753            return Err(Error::EntryNotAppendable);
754        }
755        self.join(other.ensure_stamp_after(
756            actor,
757            self.max_observed_version(),
758            EntryStampKind::Delta,
759        )?)
760    }
761
762    /// Tombstone observed data or relay-message payloads.
763    ///
764    /// Pre: `self` and `other` are the same data or relay-message carrier.
765    /// Post: every removed payload is represented by an add-dot tombstone, so
766    /// future joins with stale add replicas cannot resurrect it.
767    pub fn tombstone(&self, other: Self) -> Result<Self> {
768        self.validate_same_carrier(&other)?;
769
770        let target_values = other.data.into_iter().collect::<BTreeSet<_>>();
771        let target_dots = other.crdt.dots.into_iter().collect::<BTreeSet<_>>();
772        let has_dot_witness = !target_dots.is_empty();
773
774        match self.kind {
775            EntryKind::Data => {
776                let mut buffer = self.topic_buffer()?;
777                for (value, dot) in &buffer.values {
778                    if target_dots.contains(dot)
779                        || (!has_dot_witness && target_values.contains(value))
780                    {
781                        buffer.removes.insert(*dot);
782                    }
783                }
784                Ok(self.materialize_topic_buffer(buffer))
785            }
786            EntryKind::RelayMessage => {
787                let mut set = self.relay_set()?;
788                for (value, dot) in &set.adds.values {
789                    if target_dots.contains(dot)
790                        || (!has_dot_witness && target_values.contains(value))
791                    {
792                        set.removes.insert(*dot);
793                    }
794                }
795                Ok(self.materialize_relay_set(set))
796            }
797        }
798    }
799
800    /// Compact a Data kind entry using the receiver's current visible payloads.
801    ///
802    /// Pre: `removals` names the same Data carrier as `self`.
803    /// Post: every current visible payload not listed in `removals` is preserved
804    /// under the greatest observed register floor, and older tombstone metadata
805    /// is pruned by that floor.
806    pub fn compact_data(&self, removals: Self, actor: Did) -> Result<Self> {
807        match self.is_data_entry() {
808            true => self.compact_data_entry(removals, actor),
809            false => Err(Error::EntryNotOverwritable),
810        }
811    }
812
813    fn compact_data_entry(&self, removals: Self, actor: Did) -> Result<Self> {
814        let removals = removals.ensure_overwrite_stamp_after(actor, self.max_observed_version())?;
815        self.validate_same_carrier(&removals)?;
816        let floor = removals.crdt.register.ok_or_else(|| {
817            Error::InvalidMessage("compact data operation has no register floor".to_string())
818        })?;
819        let removal_values = removals.data.into_iter().collect::<BTreeSet<_>>();
820        let buffer = self.topic_buffer()?;
821        let output_floor = Self::compact_data_output_floor(self.crdt.register, floor);
822        let elements = Self::compact_data_elements(
823            floor,
824            &removal_values,
825            Self::data_compaction_candidates(&self.data, buffer.values),
826        )?;
827        let tombstones = Self::compact_data_tombstones(output_floor, buffer.removes);
828        Ok(Self::materialize_elements(
829            self.did,
830            EntryKind::Data,
831            Some(output_floor),
832            elements,
833            tombstones,
834        ))
835    }
836}
837
838#[cfg(test)]
839mod test_entry;
rings_core/dht/entry.rs

rings_core/dht/
entry.rs
