safemesh_crdt/
collision.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
5//! The record-ID collision alarm and its wire report.
6//!
7//! Two replicas can each admit one [`RecordId`] with a different payload.
8//! Their version vectors are then equal, so `since` offers neither payload and
9//! ordinary anti-entropy never repairs or reveals the fork. When a peer does
10//! offer the other payload, admission refuses it as
11//! [`Admission::Collision`](crate::Admission::Collision) and also raises the
12//! alarm in the receiving log ([`EventLog::collisions`]). The receiver then
13//! sends [`EventLog::collision_report_bytes`] back to the peer, whose
14//! [`EventLog::merge_collision_report_bytes`] raises the same alarm on its side.
15//!
16//! The alarm never admits, applies or replaces a payload; it only names the
17//! fork. It is not part of the encoded log or log equality. A durable
18//! replica stores the alarm separately and reloads it on restart; a restored
19//! plain log drops it. A report payload is decoded under the log's schema but
20//! is not validated against a carrier, because it is never applied.
21
22use crate::{
23    codec::{frame_crc32, read_checked_body, read_tag},
24    write_bytes, write_len, write_u32, write_u64, write_u8, CollectionLimits, DecodeError,
25    DecodeLimits, EventLog, RecordId, WireCursor, WireDecode, WireEncode, WireError, WireSchema,
26};
27use alloc::{borrow::Cow, vec::Vec};
28
29/// First byte of a [`CollisionReport`] frame. No earlier decoder accepts it:
30/// every record, log, version and state decoder returns
31/// [`WireError::InvalidTag`] for it before reading further.
32pub(super) const TAG_COLLISION_REPORT: u8 = 0x05;
33
34/// One record ID held with two different payloads.
35///
36/// In [`EventLog::collisions`], `local` is the payload this log admitted and
37/// `remote` the payload a peer holds. In a [`CollisionReport`] the perspective
38/// is the reporter's: `local` is the reporter's payload and `remote` is the
39/// payload it was offered.
40#[derive(Clone, Debug, PartialEq, Eq)]
41pub struct RecordCollision<D> {
42    pub id: RecordId,
43    pub local: D,
44    pub remote: D,
45}
46
47/// The wire message that carries collision alarms to a peer.
48///
49/// Frame: tag `0x05`, little-endian `u32` body length, its complement, the
50/// body, and a CRC-32 of both length fields and the body, as for an EventLog
51/// frame. The body is the length-prefixed schema
52/// `safemesh/collision-report/v1/<payload schema>`, a `u32` entry count, then
53/// for each entry the author and sequence as `u64` and the reporter's payload
54/// and the offered payload, each length-prefixed in its own wire encoding.
55#[derive(Clone, Debug, PartialEq, Eq)]
56pub struct CollisionReport<D> {
57    pub collisions: Vec<RecordCollision<D>>,
58}
59
60/// What merging one report entry did to the receiving log.
61#[derive(Clone, Copy, Debug, PartialEq, Eq)]
62#[non_exhaustive]
63pub enum CollisionVerdict {
64    /// This log holds the ID, the entry names a payload that differs from the
65    /// one held, and the alarm for the ID is now raised here.
66    Recorded,
67    /// The alarm for the ID was already raised here. Nothing changed.
68    Known,
69    /// This log does not hold the ID. Nothing changed.
70    Unheld,
71    /// Both payloads the entry names equal the one held: no fork at this ID
72    /// here. Nothing changed.
73    Agrees,
74}
75
76impl<D: WireSchema> WireSchema for CollisionReport<D> {
77    fn wire_schema() -> Cow<'static, [u8]> {
78        let mut schema = b"safemesh/collision-report/v1/".to_vec();
79        schema.extend_from_slice(D::wire_schema().as_ref());
80        Cow::Owned(schema)
81    }
82}
83
84impl<D: WireEncode + WireSchema> WireEncode for CollisionReport<D> {
85    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
86        let mut body = Vec::new();
87        write_bytes(&mut body, Self::wire_schema().as_ref())?;
88        write_len(&mut body, self.collisions.len())?;
89        for entry in &self.collisions {
90            write_u64(&mut body, entry.id.replica);
91            write_u64(&mut body, entry.id.sequence);
92            write_bytes(&mut body, &entry.local.to_wire_bytes()?)?;
93            write_bytes(&mut body, &entry.remote.to_wire_bytes()?)?;
94        }
95        let len = u32::try_from(body.len()).map_err(|_| WireError::LengthOverflow)?;
96        write_u8(out, TAG_COLLISION_REPORT);
97        let start = out.len();
98        write_u32(out, len);
99        write_u32(out, !len);
100        out.extend_from_slice(&body);
101        let checksum = frame_crc32(&out[start..]);
102        write_u32(out, checksum);
103        Ok(())
104    }
105}
106
107impl<D: WireDecode + WireSchema> CollisionReport<D> {
108    /// Decode a report with an entry budget and nested payload budgets.
109    ///
110    /// Checks the tag, length pair and CRC before reading the body, then the
111    /// schema, and stops before entry `max_records + 1`. Any failure returns
112    /// no partial report.
113    pub fn from_wire_bytes_with_limits(
114        bytes: &[u8],
115        limits: DecodeLimits,
116    ) -> Result<Self, DecodeError> {
117        let mut cursor = WireCursor::new(bytes);
118        let report = Self::decode_limited(
119            &mut cursor,
120            CollectionLimits {
121                max_elements: limits
122                    .max_collection_elements
123                    .or(CollectionLimits::WIRE_DEFAULT.max_elements),
124            },
125            limits.max_records,
126            limits.max_records,
127        )?;
128        if !cursor.is_empty() {
129            return Err(WireError::TrailingBytes.into());
130        }
131        Ok(report)
132    }
133
134    fn decode_limited(
135        cursor: &mut WireCursor<'_>,
136        collections: CollectionLimits,
137        max_entries: Option<usize>,
138        nested_max_records: Option<usize>,
139    ) -> Result<Self, DecodeError> {
140        read_tag(cursor, TAG_COLLISION_REPORT)?;
141        let mut body = WireCursor::new(read_checked_body(cursor)?);
142        let schema_len = body.read_len()?;
143        if body.read_exact(schema_len)? != Self::wire_schema().as_ref() {
144            return Err(WireError::DeltaTypeMismatch.into());
145        }
146        let mut collisions = Vec::new();
147        for index in 0..body.read_len()? {
148            if let Some(max_records) = max_entries {
149                if index >= max_records {
150                    return Err(DecodeError::RecordLimitExceeded { max_records });
151                }
152            }
153            let id = RecordId {
154                replica: body.read_u64()?,
155                sequence: body.read_u64()?,
156            };
157            let local = decode_payload(&mut body, collections, nested_max_records)?;
158            let remote = decode_payload(&mut body, collections, nested_max_records)?;
159            collisions.push(RecordCollision { id, local, remote });
160        }
161        if !body.is_empty() {
162            return Err(WireError::TrailingBytes.into());
163        }
164        Ok(Self { collisions })
165    }
166}
167
168fn decode_payload<D: WireDecode>(
169    body: &mut WireCursor<'_>,
170    collections: CollectionLimits,
171    max_records: Option<usize>,
172) -> Result<D, WireError> {
173    let len = body.read_len()?;
174    let mut cursor = WireCursor::new(body.read_exact(len)?);
175    let payload = D::decode_wire_with_limits(&mut cursor, collections, max_records)?;
176    if !cursor.is_empty() {
177        return Err(WireError::TrailingBytes);
178    }
179    Ok(payload)
180}
181
182impl<D: WireDecode + WireSchema> WireDecode for CollisionReport<D> {
183    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
184        Self::decode_wire_with_collection_limits(cursor, CollectionLimits::WIRE_DEFAULT)
185    }
186
187    fn decode_wire_with_collection_limits(
188        cursor: &mut WireCursor<'_>,
189        limits: CollectionLimits,
190    ) -> Result<Self, WireError> {
191        Self::decode_limited(
192            cursor,
193            limits,
194            DecodeLimits::default().max_records,
195            DecodeLimits::default().max_records,
196        )
197        .map_err(|error| WireError::from(error.with_default_record_limit()))
198    }
199
200    fn decode_wire_with_limits(
201        cursor: &mut WireCursor<'_>,
202        collections: CollectionLimits,
203        max_records: Option<usize>,
204    ) -> Result<Self, WireError> {
205        Self::decode_limited(cursor, collections, max_records, max_records).map_err(|error| {
206            match error {
207                DecodeError::Wire(error) => error,
208                DecodeError::RecordLimitExceeded { max_records }
209                | DecodeError::DefaultRecordLimitExceeded { max_records } => {
210                    WireError::NestedRecordLimitExceeded { max_records }
211                }
212            }
213        })
214    }
215}
216
217impl<D> EventLog<D> {
218    /// Every held ID whose alarm is raised, in ID order, with the payload held
219    /// here and the first differing payload a peer offered or reported.
220    /// Empty when no fork has been seen. Durable replicas store it separately.
221    pub fn collisions(&self) -> Vec<RecordCollision<D>>
222    where
223        D: Clone,
224    {
225        self.collisions
226            .iter()
227            .map(|(&id, remote)| RecordCollision {
228                id,
229                local: self.records[self.seen[&id]].delta.clone(),
230                remote: remote.clone(),
231            })
232            .collect()
233    }
234
235    /// The report a peer needs to raise the same alarms: every entry of
236    /// [`Self::collisions`], from this log's perspective.
237    pub fn collision_report(&self) -> CollisionReport<D>
238    where
239        D: Clone,
240    {
241        CollisionReport {
242            collisions: self.collisions(),
243        }
244    }
245
246    /// Merge a peer's report. Each entry is judged against the payload held
247    /// here and returns its verdict in input order. Only
248    /// [`CollisionVerdict::Recorded`] changes anything, and it changes only the
249    /// alarm: never the records, version or a carrier.
250    pub fn merge_collision_report(
251        &mut self,
252        report: CollisionReport<D>,
253    ) -> Vec<(RecordId, CollisionVerdict)>
254    where
255        D: PartialEq,
256    {
257        report
258            .collisions
259            .into_iter()
260            .map(|entry| (entry.id, self.merge_collision_entry(entry)))
261            .collect()
262    }
263
264    fn merge_collision_entry(&mut self, entry: RecordCollision<D>) -> CollisionVerdict
265    where
266        D: PartialEq,
267    {
268        let Some(&index) = self.seen.get(&entry.id) else {
269            return CollisionVerdict::Unheld;
270        };
271        let held = &self.records[index].delta;
272        // The reporter's own payload is the usual witness; the offered payload
273        // is normally this log's own.
274        let witness = if entry.local != *held {
275            entry.local
276        } else if entry.remote != *held {
277            entry.remote
278        } else {
279            return CollisionVerdict::Agrees;
280        };
281        if self.collisions.contains_key(&entry.id) {
282            return CollisionVerdict::Known;
283        }
284        self.collisions.insert(entry.id, witness);
285        CollisionVerdict::Recorded
286    }
287}
288
289impl<D: Clone + WireEncode + WireSchema> EventLog<D> {
290    /// Encode [`Self::collision_report`] for the peer, or `None` when no alarm
291    /// is raised here. Send it back after merging a peer's batch.
292    pub fn collision_report_bytes(&self) -> Result<Option<Vec<u8>>, WireError> {
293        if self.collisions.is_empty() {
294            return Ok(None);
295        }
296        self.collision_report().to_wire_bytes().map(Some)
297    }
298}
299
300impl<D: PartialEq + WireDecode + WireSchema> EventLog<D> {
301    /// Decode a peer's report under `limits`, then merge it. A decode failure
302    /// refuses the whole message and changes nothing.
303    pub fn merge_collision_report_bytes(
304        &mut self,
305        bytes: &[u8],
306        limits: DecodeLimits,
307    ) -> Result<Vec<(RecordId, CollisionVerdict)>, DecodeError> {
308        let report = CollisionReport::from_wire_bytes_with_limits(bytes, limits)?;
309        Ok(self.merge_collision_report(report))
310    }
311}
safemesh_crdt/collision.rs

safemesh_crdt/
collision.rs
