safemesh_crdt/
event_log.rs

1// SafeMesh — delta-state CRDT convergence, built on crdt-lean.
2// Copyright (C) 2026 Ben Cassie
3// SPDX-License-Identifier: Apache-2.0
4
5use crate::{Crdt, Record, RecordId, VersionVector, WireError};
6use alloc::{collections::BTreeMap, vec::Vec};
7
8/// Full decoded payload equality distinguishes redelivery from an ID collision.
9#[derive(Clone, Copy, Debug, PartialEq, Eq)]
10pub enum Admission {
11    Accepted,
12    Duplicate,
13    Collision,
14    /// The log shape or record is incompatible with the supplied carrier.
15    Invalid(WireError),
16}
17
18#[derive(Clone, Copy, Debug, PartialEq, Eq)]
19pub enum AppendError {
20    SequenceExhausted,
21    InvalidRecord(WireError),
22}
23
24impl core::fmt::Display for AppendError {
25    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
26        match self {
27            Self::SequenceExhausted => {
28                f.write_str("cannot append: replica sequence number exhausted")
29            }
30            Self::InvalidRecord(error) => write!(f, "cannot append invalid record: {error}"),
31        }
32    }
33}
34
35impl core::error::Error for AppendError {
36    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
37        match self {
38            Self::InvalidRecord(error) => Some(error),
39            Self::SequenceExhausted => None,
40        }
41    }
42}
43
44/// Append-only, deduplicating event log for CRDT deltas.
45/// Equality compares log data, excluding the in-memory first-admission binding
46/// flag and the in-memory collision register.
47#[derive(Clone, Debug)]
48pub struct EventLog<D> {
49    pub(super) replica_count: Option<usize>,
50    pub(super) shape_bound: bool,
51    pub(super) records: Vec<Record<D>>,
52    pub(super) seen: BTreeMap<RecordId, usize>,
53    pub(super) version: VersionVector,
54    // Held IDs a peer holds with a different payload, with the first such
55    // payload seen. At most one entry per admitted record; never encoded.
56    pub(super) collisions: BTreeMap<RecordId, D>,
57}
58
59// Binding an empty unbounded log during decode must not change payload equality:
60// logs can themselves be nested in records and compared for deduplication.
61// The collision register is an alarm about peers, not log data.
62impl<D: PartialEq> PartialEq for EventLog<D> {
63    fn eq(&self, other: &Self) -> bool {
64        self.replica_count == other.replica_count
65            && self.records == other.records
66            && self.seen == other.seen
67            && self.version == other.version
68    }
69}
70
71impl<D: Eq> Eq for EventLog<D> {}
72
73impl<D> EventLog<D> {
74    /// Create an unbound log. The first accepted record binds its carrier shape.
75    /// Use `for_crdt` to declare a shape before any records are accepted.
76    pub fn new() -> Self {
77        EventLog {
78            replica_count: None,
79            shape_bound: false,
80            records: Vec::new(),
81            seen: BTreeMap::new(),
82            version: VersionVector::new(),
83            collisions: BTreeMap::new(),
84        }
85    }
86
87    /// Create a log for a fixed replica domain, including an empty domain.
88    pub fn with_replica_count(replica_count: usize) -> Self {
89        Self {
90            replica_count: Some(replica_count),
91            shape_bound: true,
92            ..Self::new()
93        }
94    }
95
96    /// Bind persistence metadata to the CRDT's actual domain before recording.
97    pub fn for_crdt<C: Crdt<Delta = D>>(state: &C) -> Self {
98        Self {
99            replica_count: state.replica_count(),
100            shape_bound: true,
101            ..Self::new()
102        }
103    }
104
105    pub fn replica_count(&self) -> Option<usize> {
106        self.replica_count
107    }
108
109    /// Allocate a fresh ID and admit a delta without applying it to `state`.
110    /// Returns an error if the sequence is exhausted or admission refuses the
111    /// record; in either case the log and state remain unchanged.
112    pub fn append<C: Crdt<Delta = D>>(
113        &mut self,
114        state: &mut C,
115        replica: u64,
116        delta: D,
117    ) -> Result<RecordId, AppendError>
118    where
119        D: PartialEq,
120    {
121        self.append_with(state, replica, delta, |_, _| {})
122    }
123
124    /// Allocate a fresh ID, then use the same gate as incoming records.
125    pub fn append_with<C, F>(
126        &mut self,
127        state: &mut C,
128        replica: u64,
129        delta: D,
130        apply: F,
131    ) -> Result<RecordId, AppendError>
132    where
133        D: PartialEq,
134        C: Crdt<Delta = D>,
135        F: FnOnce(&mut C, &D),
136    {
137        let sequence = self
138            .seen
139            .range(
140                RecordId {
141                    replica,
142                    sequence: 0,
143                }..=RecordId {
144                    replica,
145                    sequence: u64::MAX,
146                },
147            )
148            .next_back()
149            .map(|(id, _)| id.sequence)
150            .unwrap_or(0)
151            .checked_add(1)
152            .ok_or(AppendError::SequenceExhausted)?;
153        let id = RecordId { replica, sequence };
154        let outcome = self.admit_with(state, Record { id, delta }, apply);
155        if let Admission::Invalid(error) = outcome {
156            return Err(AppendError::InvalidRecord(error));
157        }
158        debug_assert_eq!(outcome, Admission::Accepted);
159        Ok(id)
160    }
161
162    pub fn merge_records<C: Crdt<Delta = D>, I>(&mut self, state: &C, records: I) -> Vec<Admission>
163    where
164        D: PartialEq,
165        I: IntoIterator<Item = Record<D>>,
166    {
167        records
168            .into_iter()
169            .map(|r| self.insert_record(state, r))
170            .collect()
171    }
172
173    /// Validate the log shape and record against the destination carrier, then
174    /// check record identity. An unbound log binds only on acceptance.
175    /// Only Accepted invokes `apply`, with the same carrier used for validation.
176    /// An invalid fresh record leaves the log, version, and carrier unchanged.
177    /// Duplicate/Collision retain their identity verdicts, but also run validation.
178    /// A Collision also raises the collision alarm for the ID
179    /// ([`Self::collisions`]); log, version and carrier stay unchanged.
180    /// The callback must be infallible and use the same delta interpretation as
181    /// replay. This is an in-memory transition, not a crash-durability guarantee.
182    #[must_use]
183    pub fn admit_with<C, F>(&mut self, state: &mut C, record: Record<D>, apply: F) -> Admission
184    where
185        D: PartialEq,
186        C: Crdt<Delta = D>,
187        F: FnOnce(&mut C, &D),
188    {
189        let outcome = self.admission(state, &record);
190        if outcome != Admission::Accepted {
191            self.raise_on_collision(outcome, record);
192            return outcome;
193        }
194        self.bind_shape(state);
195        self.commit_record(record);
196        apply(state, &self.records.last().expect("accepted record").delta);
197        Admission::Accepted
198    }
199
200    // Only called after admission, or by the inert wire decoder after identity checks.
201    pub(super) fn commit_record(&mut self, record: Record<D>) {
202        let id = record.id;
203        self.seen.insert(id, self.records.len());
204        self.records.push(record);
205        self.version.observe(id);
206        self.advance_contiguous_version(id.replica);
207    }
208
209    // Shared read-only decision for admission and transactional preflight.
210    // Consult the identity index without copying any admitted payloads.
211    pub(super) fn admission<C: Crdt<Delta = D>>(&self, state: &C, record: &Record<D>) -> Admission
212    where
213        D: PartialEq,
214    {
215        let identity = self.identity_admission(record);
216        let record_validation = state.validate_record(record.id, &record.delta);
217        let shape_validation = if self.shape_bound {
218            self.validate_shape(state)
219        } else {
220            Ok(())
221        };
222        let validation = shape_validation.and(record_validation);
223        match (identity, validation) {
224            (Admission::Accepted, Err(error)) => Admission::Invalid(error),
225            (outcome, _) => outcome,
226        }
227    }
228
229    // Also used by checked restore, where even an empty decoded log has a shape.
230    pub(super) fn validate_shape<C: Crdt<Delta = D>>(&self, state: &C) -> Result<(), WireError> {
231        match (state.replica_count(), self.replica_count) {
232            (Some(expected), Some(actual)) if expected != actual => {
233                Err(WireError::ReplicaCountMismatch { expected, actual })
234            }
235            (None, Some(_)) | (Some(_), None) => Err(WireError::ArityKindMismatch),
236            _ => Ok(()),
237        }
238    }
239
240    fn bind_shape<C: Crdt<Delta = D>>(&mut self, state: &C) {
241        if !self.shape_bound {
242            self.replica_count = state.replica_count();
243            self.shape_bound = true;
244        }
245    }
246
247    // Wire decoding has no carrier yet; checked loaders validate before replay.
248    pub(super) fn identity_admission(&self, record: &Record<D>) -> Admission
249    where
250        D: PartialEq,
251    {
252        match self.seen.get(&record.id) {
253            Some(&index) if self.records[index].delta == record.delta => Admission::Duplicate,
254            Some(_) => Admission::Collision,
255            None => Admission::Accepted,
256        }
257    }
258
259    pub fn version(&self) -> &VersionVector {
260        &self.version
261    }
262
263    pub fn records(&self) -> &[Record<D>] {
264        &self.records
265    }
266
267    /// Admit without applying; `state` must be the carrier used for later replay.
268    /// A Collision raises the collision alarm, as in [`Self::admit_with`].
269    pub fn insert_record<C: Crdt<Delta = D>>(&mut self, state: &C, record: Record<D>) -> Admission
270    where
271        D: PartialEq,
272    {
273        let outcome = self.admission(state, &record);
274        if outcome == Admission::Accepted {
275            self.bind_shape(state);
276            self.commit_record(record);
277        } else {
278            self.raise_on_collision(outcome, record);
279        }
280        outcome
281    }
282
283    // Keep the first differing payload offered for a held ID as the witness.
284    pub(crate) fn raise_on_collision(&mut self, outcome: Admission, record: Record<D>) {
285        if outcome == Admission::Collision {
286            self.collisions.entry(record.id).or_insert(record.delta);
287        }
288    }
289
290    fn advance_contiguous_version(&mut self, replica: u64) {
291        while let Some(next) = self.version.get(replica).checked_add(1) {
292            if self.seen.contains_key(&RecordId {
293                replica,
294                sequence: next,
295            }) {
296                self.version.set(replica, next);
297            } else {
298                break;
299            }
300        }
301    }
302}
303
304impl<D: Clone> EventLog<D> {
305    /// Return admitted records outside the peer's positive contiguous prefixes
306    /// and independently acknowledged sequence-zero records.
307    /// By design, a permanent gap offers all later records from that writer on every exchange; deliver the missing record.
308    pub fn since(&self, version: &VersionVector) -> Vec<Record<D>> {
309        self.records
310            .iter()
311            .filter(|record| !version.includes(record.id))
312            .cloned()
313            .collect()
314    }
315}
316
317impl<D> Default for EventLog<D> {
318    fn default() -> Self {
319        Self::new()
320    }
321}
safemesh_crdt/event_log.rs

safemesh_crdt/
event_log.rs
