safemesh_crdt/
wire.rs

1// SafeMesh — canonical wire codec implementations.
2// Copyright (C) 2026 Ben Cassie
3// SPDX-License-Identifier: Apache-2.0
4
5//! Canonical wire codec implementations for the CRDT types.
6
7use alloc::collections::{BTreeMap, BTreeSet};
8use alloc::string::String;
9use alloc::vec::Vec;
10
11use super::{
12    read_tag, write_bytes, write_len, write_u64, write_u8, CollectionLimits, DecodeLimits,
13    EnableWinsFlag, EnableWinsFlagDelta, EventLog, GCounterDelta, GSet, LwwMap, LwwMapDelta,
14    LwwRegister, LwwRegisterDelta, OrSet, OrSetDelta, PnCounterDelta, Record, RecordId, Rga,
15    RgaDelta, VersionVector, VersionVectorLimitError, VersionVectorLimits, WireCursor, WireDecode,
16    WireEncode, WireError, WireSchema, TAG_ENABLE_WINS_FLAG_DISABLE_U64,
17    TAG_ENABLE_WINS_FLAG_ENABLE_U64, TAG_ENABLE_WINS_FLAG_U64, TAG_GCOUNTER_DELTA, TAG_GSET_U64,
18    TAG_LWW_MAP_REMOVE_U64, TAG_LWW_MAP_SET_U64, TAG_LWW_MAP_U64, TAG_LWW_REGISTER_DELTA_U64,
19    TAG_LWW_REGISTER_U64, TAG_ORSET_ADD_STRING, TAG_ORSET_ADD_U64, TAG_ORSET_REMOVE_STRING,
20    TAG_ORSET_REMOVE_U64, TAG_ORSET_STRING, TAG_ORSET_U64, TAG_PNCOUNTER_DEC, TAG_PNCOUNTER_INC,
21    TAG_RECORD, TAG_RGA_DELETE_U64, TAG_RGA_INSERT_U64, TAG_RGA_U64, TAG_VERSION_VECTOR,
22};
23
24impl WireEncode for VersionVector {
25    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
26        write_u8(out, TAG_VERSION_VECTOR);
27        write_len(out, self.entries().len())?;
28        for (&replica, &prefix) in self.entries() {
29            write_u64(out, replica);
30            write_u64(out, prefix);
31        }
32        write_len(out, self.zero_replicas().len())?;
33        for &replica in self.zero_replicas() {
34            write_u64(out, replica);
35        }
36        Ok(())
37    }
38}
39
40impl VersionVector {
41    /// Decode with explicit author and sequence-zero acknowledgement budgets.
42    /// Claimed entry counts are checked before reading or allocating entries.
43    pub fn from_wire_bytes_with_limits(
44        bytes: &[u8],
45        limits: VersionVectorLimits,
46    ) -> Result<Self, WireError> {
47        let mut cursor = WireCursor::new(bytes);
48        let value = Self::decode_with_limits(&mut cursor, limits)?;
49        if cursor.is_empty() {
50            Ok(value)
51        } else {
52            Err(WireError::TrailingBytes)
53        }
54    }
55
56    fn decode_with_limits(
57        cursor: &mut WireCursor<'_>,
58        limits: VersionVectorLimits,
59    ) -> Result<Self, WireError> {
60        read_tag(cursor, TAG_VERSION_VECTOR)?;
61        let author_count = cursor.read_len()?;
62        if let Some(max_authors) = limits.max_authors {
63            if author_count > max_authors {
64                return Err(WireError::VersionAuthorLimitExceeded { max_authors });
65            }
66        }
67        let mut entries = BTreeMap::new();
68        for _ in 0..author_count {
69            let replica = cursor.read_u64()?;
70            let prefix = cursor.read_u64()?;
71            if prefix == 0 || entries.contains_key(&replica) {
72                return Err(WireError::NonCanonicalVersionVector);
73            }
74            check_state_entry(entries.last_key_value().map(|(key, _)| key), &replica)?;
75            entries.insert(replica, prefix);
76        }
77        let zero_count = cursor.read_len()?;
78        if let Some(max_zero_replicas) = limits.max_zero_replicas {
79            if zero_count > max_zero_replicas {
80                return Err(WireError::VersionZeroReplicaLimitExceeded { max_zero_replicas });
81            }
82        }
83        let mut zeros = BTreeSet::new();
84        for _ in 0..zero_count {
85            let replica = cursor.read_u64()?;
86            if zeros.contains(&replica) {
87                return Err(WireError::NonCanonicalVersionVector);
88            }
89            check_state_entry(zeros.last(), &replica)?;
90            zeros.insert(replica);
91        }
92        Self::from_peer_prefixes_with_limits(&entries, &zeros, limits).map_err(
93            |error| match error {
94                VersionVectorLimitError::AuthorLimitExceeded { max_authors } => {
95                    WireError::VersionAuthorLimitExceeded { max_authors }
96                }
97                VersionVectorLimitError::ZeroReplicaLimitExceeded { max_zero_replicas } => {
98                    WireError::VersionZeroReplicaLimitExceeded { max_zero_replicas }
99                }
100                VersionVectorLimitError::Prefix(_) => WireError::NonCanonicalVersionVector,
101            },
102        )
103    }
104}
105
106impl WireDecode for VersionVector {
107    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
108        Self::decode_with_limits(cursor, VersionVectorLimits::WIRE_DEFAULT)
109    }
110}
111
112impl WireEncode for GCounterDelta {
113    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
114        write_u8(out, TAG_GCOUNTER_DELTA);
115        write_u64(
116            out,
117            u64::try_from(self.replica).map_err(|_| WireError::LengthOverflow)?,
118        );
119        write_u64(out, self.tally);
120        Ok(())
121    }
122}
123
124impl WireDecode for GCounterDelta {
125    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
126        read_tag(cursor, TAG_GCOUNTER_DELTA)?;
127        let replica = usize::try_from(cursor.read_u64()?).map_err(|_| WireError::LengthOverflow)?;
128        let tally = cursor.read_u64()?;
129        Ok(GCounterDelta { replica, tally })
130    }
131}
132
133impl WireEncode for PnCounterDelta {
134    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
135        match self {
136            PnCounterDelta::Inc { replica, tally } => {
137                write_u8(out, TAG_PNCOUNTER_INC);
138                write_u64(
139                    out,
140                    u64::try_from(*replica).map_err(|_| WireError::LengthOverflow)?,
141                );
142                write_u64(out, *tally);
143            }
144            PnCounterDelta::Dec { replica, tally } => {
145                write_u8(out, TAG_PNCOUNTER_DEC);
146                write_u64(
147                    out,
148                    u64::try_from(*replica).map_err(|_| WireError::LengthOverflow)?,
149                );
150                write_u64(out, *tally);
151            }
152        }
153        Ok(())
154    }
155}
156
157impl WireDecode for PnCounterDelta {
158    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
159        let tag = cursor.read_u8()?;
160        let replica = usize::try_from(cursor.read_u64()?).map_err(|_| WireError::LengthOverflow)?;
161        let tally = cursor.read_u64()?;
162        match tag {
163            TAG_PNCOUNTER_INC => Ok(PnCounterDelta::Inc { replica, tally }),
164            TAG_PNCOUNTER_DEC => Ok(PnCounterDelta::Dec { replica, tally }),
165            _ => Err(WireError::InvalidTag),
166        }
167    }
168}
169
170impl WireEncode for GSet<u64> {
171    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
172        write_u8(out, TAG_GSET_U64);
173        write_len(out, self.elements.len())?;
174        for element in &self.elements {
175            write_u64(out, *element);
176        }
177        Ok(())
178    }
179}
180
181impl WireDecode for GSet<u64> {
182    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
183        Self::decode_with_limits(cursor, CollectionLimits::WIRE_DEFAULT)
184    }
185    fn decode_wire_with_collection_limits(
186        cursor: &mut WireCursor<'_>,
187        limits: CollectionLimits,
188    ) -> Result<Self, WireError> {
189        Self::decode_with_limits(cursor, limits)
190    }
191}
192
193// Validate state entries before insertion can erase non-canonical input.
194fn check_state_entry<T: Ord>(previous: Option<&T>, entry: &T) -> Result<(), WireError> {
195    if let Some(previous) = previous {
196        match previous.cmp(entry) {
197            core::cmp::Ordering::Equal => return Err(WireError::DuplicateEntry),
198            core::cmp::Ordering::Greater => return Err(WireError::NonCanonicalOrder),
199            core::cmp::Ordering::Less => {}
200        }
201    }
202    Ok(())
203}
204
205fn check_collection_count(count: usize, limits: CollectionLimits) -> Result<(), WireError> {
206    if let Some(max_elements) = limits.max_elements {
207        if count > max_elements {
208            return Err(WireError::CollectionElementLimitExceeded { max_elements });
209        }
210    }
211    Ok(())
212}
213
214impl GSet<u64> {
215    /// Decode with an explicit element count budget, checked before reading elements.
216    pub fn from_wire_bytes_with_limits(
217        bytes: &[u8],
218        limits: CollectionLimits,
219    ) -> Result<Self, WireError> {
220        let mut cursor = WireCursor::new(bytes);
221        let value = Self::decode_with_limits(&mut cursor, limits)?;
222        if cursor.is_empty() {
223            Ok(value)
224        } else {
225            Err(WireError::TrailingBytes)
226        }
227    }
228
229    fn decode_with_limits(
230        cursor: &mut WireCursor<'_>,
231        limits: CollectionLimits,
232    ) -> Result<Self, WireError> {
233        read_tag(cursor, TAG_GSET_U64)?;
234        let count = cursor.read_len()?;
235        check_collection_count(count, limits)?;
236        let mut set = GSet::new();
237        for _ in 0..count {
238            let element = cursor.read_u64()?;
239            check_state_entry(set.elements.last(), &element)?;
240            set.insert(element);
241        }
242        Ok(set)
243    }
244}
245
246impl WireEncode for OrSetDelta<u64, u64> {
247    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
248        match self {
249            OrSetDelta::Add { element, token } => {
250                write_u8(out, TAG_ORSET_ADD_U64);
251                write_u64(out, *element);
252                write_u64(out, *token);
253            }
254            OrSetDelta::Remove { tokens } => {
255                write_u8(out, TAG_ORSET_REMOVE_U64);
256                let mut sorted_tokens = tokens.clone();
257                sorted_tokens.sort_unstable();
258                sorted_tokens.dedup();
259                write_len(out, sorted_tokens.len())?;
260                for token in &sorted_tokens {
261                    write_u64(out, *token);
262                }
263            }
264        }
265        Ok(())
266    }
267}
268
269impl WireDecode for OrSetDelta<u64, u64> {
270    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
271        Self::decode_wire_with_collection_limits(cursor, CollectionLimits::WIRE_DEFAULT)
272    }
273
274    fn decode_wire_with_collection_limits(
275        cursor: &mut WireCursor<'_>,
276        limits: CollectionLimits,
277    ) -> Result<Self, WireError> {
278        match cursor.read_u8()? {
279            TAG_ORSET_ADD_U64 => Ok(OrSetDelta::Add {
280                element: cursor.read_u64()?,
281                token: cursor.read_u64()?,
282            }),
283            TAG_ORSET_REMOVE_U64 => {
284                let mut tokens = Vec::new();
285                let count = cursor.read_len()?;
286                check_collection_count(count, limits)?;
287                for _ in 0..count {
288                    tokens.push(cursor.read_u64()?);
289                }
290                Ok(OrSetDelta::Remove { tokens })
291            }
292            _ => Err(WireError::InvalidTag),
293        }
294    }
295}
296
297// UTF-8 delta tags are distinct from the u64 delta tags, including Remove.
298// Add carries a byte-length-prefixed UTF-8 element followed by a u64 token;
299// Remove carries a token count followed by sorted, unique u64 tokens.
300impl WireEncode for OrSetDelta<String, u64> {
301    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
302        match self {
303            OrSetDelta::Add { element, token } => {
304                write_u8(out, TAG_ORSET_ADD_STRING);
305                write_bytes(out, element.as_bytes())?;
306                write_u64(out, *token);
307            }
308            OrSetDelta::Remove { tokens } => {
309                write_u8(out, TAG_ORSET_REMOVE_STRING);
310                let mut sorted_tokens = tokens.clone();
311                sorted_tokens.sort_unstable();
312                sorted_tokens.dedup();
313                write_len(out, sorted_tokens.len())?;
314                for token in &sorted_tokens {
315                    write_u64(out, *token);
316                }
317            }
318        }
319        Ok(())
320    }
321}
322
323impl WireDecode for OrSetDelta<String, u64> {
324    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
325        Self::decode_wire_with_collection_limits(cursor, CollectionLimits::WIRE_DEFAULT)
326    }
327
328    fn decode_wire_with_collection_limits(
329        cursor: &mut WireCursor<'_>,
330        limits: CollectionLimits,
331    ) -> Result<Self, WireError> {
332        match cursor.read_u8()? {
333            TAG_ORSET_ADD_STRING => {
334                let len = cursor.read_len()?;
335                let element = core::str::from_utf8(cursor.read_exact(len)?)
336                    .map_err(|_| WireError::InvalidUtf8)?;
337                let token = cursor.read_u64()?;
338                Ok(OrSetDelta::Add {
339                    element: String::from(element),
340                    token,
341                })
342            }
343            TAG_ORSET_REMOVE_STRING => {
344                let mut tokens = Vec::new();
345                let count = cursor.read_len()?;
346                check_collection_count(count, limits)?;
347                for _ in 0..count {
348                    tokens.push(cursor.read_u64()?);
349                }
350                Ok(OrSetDelta::Remove { tokens })
351            }
352            _ => Err(WireError::InvalidTag),
353        }
354    }
355}
356
357impl WireEncode for OrSet<u64, u64> {
358    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
359        write_u8(out, TAG_ORSET_U64);
360        write_len(out, self.adds.len())?;
361        for (element, token) in &self.adds {
362            write_u64(out, *element);
363            write_u64(out, *token);
364        }
365        write_len(out, self.tombstones.len())?;
366        for token in &self.tombstones {
367            write_u64(out, *token);
368        }
369        Ok(())
370    }
371}
372
373impl WireDecode for OrSet<u64, u64> {
374    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
375        Self::decode_wire_with_collection_limits(cursor, CollectionLimits::WIRE_DEFAULT)
376    }
377
378    fn decode_wire_with_collection_limits(
379        cursor: &mut WireCursor<'_>,
380        limits: CollectionLimits,
381    ) -> Result<Self, WireError> {
382        read_tag(cursor, TAG_ORSET_U64)?;
383        let mut set = OrSet::new();
384        let count = cursor.read_len()?;
385        check_collection_count(count, limits)?;
386        for _ in 0..count {
387            let element = cursor.read_u64()?;
388            let token = cursor.read_u64()?;
389            let entry = (element, token);
390            check_state_entry(set.adds.last(), &entry)?;
391            set.add(entry.0, entry.1);
392        }
393        let mut tombstones = Vec::new();
394        let count = cursor.read_len()?;
395        check_collection_count(count, limits)?;
396        for _ in 0..count {
397            let token = cursor.read_u64()?;
398            check_state_entry(tombstones.last(), &token)?;
399            tombstones.push(token);
400        }
401        set.apply_remove(tombstones);
402        Ok(set)
403    }
404}
405
406impl WireEncode for OrSet<String, u64> {
407    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
408        write_u8(out, TAG_ORSET_STRING);
409        write_len(out, self.adds.len())?;
410        for (element, token) in &self.adds {
411            write_bytes(out, element.as_bytes())?;
412            write_u64(out, *token);
413        }
414        write_len(out, self.tombstones.len())?;
415        for token in &self.tombstones {
416            write_u64(out, *token);
417        }
418        Ok(())
419    }
420}
421
422impl WireDecode for OrSet<String, u64> {
423    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
424        Self::decode_wire_with_collection_limits(cursor, CollectionLimits::WIRE_DEFAULT)
425    }
426
427    fn decode_wire_with_collection_limits(
428        cursor: &mut WireCursor<'_>,
429        limits: CollectionLimits,
430    ) -> Result<Self, WireError> {
431        read_tag(cursor, TAG_ORSET_STRING)?;
432        let mut set = OrSet::new();
433        let count = cursor.read_len()?;
434        check_collection_count(count, limits)?;
435        for _ in 0..count {
436            let len = cursor.read_len()?;
437            let element = core::str::from_utf8(cursor.read_exact(len)?)
438                .map_err(|_| WireError::InvalidUtf8)?;
439            let element = String::from(element);
440            let token = cursor.read_u64()?;
441            let entry = (element, token);
442            check_state_entry(set.adds.last(), &entry)?;
443            if !set.adds.insert(entry) {
444                return Err(WireError::DuplicateEntry);
445            }
446        }
447        let mut tombstones = Vec::new();
448        let count = cursor.read_len()?;
449        check_collection_count(count, limits)?;
450        for _ in 0..count {
451            let token = cursor.read_u64()?;
452            check_state_entry(tombstones.last(), &token)?;
453            tombstones.push(token);
454        }
455        set.apply_remove(tombstones);
456        Ok(set)
457    }
458}
459
460impl WireEncode for RgaDelta<u64, u64> {
461    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
462        match self {
463            RgaDelta::Insert { position, value } => {
464                write_u8(out, TAG_RGA_INSERT_U64);
465                write_u64(out, *position);
466                write_u64(out, *value);
467            }
468            RgaDelta::Delete { position } => {
469                write_u8(out, TAG_RGA_DELETE_U64);
470                write_u64(out, *position);
471            }
472        }
473        Ok(())
474    }
475}
476
477impl WireDecode for RgaDelta<u64, u64> {
478    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
479        match cursor.read_u8()? {
480            TAG_RGA_INSERT_U64 => Ok(RgaDelta::Insert {
481                position: cursor.read_u64()?,
482                value: cursor.read_u64()?,
483            }),
484            TAG_RGA_DELETE_U64 => Ok(RgaDelta::Delete {
485                position: cursor.read_u64()?,
486            }),
487            _ => Err(WireError::InvalidTag),
488        }
489    }
490}
491
492impl WireEncode for Rga<u64, u64> {
493    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
494        write_u8(out, TAG_RGA_U64);
495        write_len(out, self.placed.len())?;
496        for (position, value) in &self.placed {
497            write_u64(out, *position);
498            write_u64(out, *value);
499        }
500        write_len(out, self.tombstones.len())?;
501        for position in &self.tombstones {
502            write_u64(out, *position);
503        }
504        Ok(())
505    }
506}
507
508impl WireDecode for Rga<u64, u64> {
509    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
510        Self::decode_with_limits(cursor, CollectionLimits::WIRE_DEFAULT)
511    }
512    fn decode_wire_with_collection_limits(
513        cursor: &mut WireCursor<'_>,
514        limits: CollectionLimits,
515    ) -> Result<Self, WireError> {
516        Self::decode_with_limits(cursor, limits)
517    }
518}
519
520impl Rga<u64, u64> {
521    /// Decode with an explicit per-list count budget for placements and tombstones.
522    pub fn from_wire_bytes_with_limits(
523        bytes: &[u8],
524        limits: CollectionLimits,
525    ) -> Result<Self, WireError> {
526        let mut cursor = WireCursor::new(bytes);
527        let value = Self::decode_with_limits(&mut cursor, limits)?;
528        if cursor.is_empty() {
529            Ok(value)
530        } else {
531            Err(WireError::TrailingBytes)
532        }
533    }
534
535    fn decode_with_limits(
536        cursor: &mut WireCursor<'_>,
537        limits: CollectionLimits,
538    ) -> Result<Self, WireError> {
539        read_tag(cursor, TAG_RGA_U64)?;
540        let placed_count = cursor.read_len()?;
541        check_collection_count(placed_count, limits)?;
542        let mut rga = Rga::new();
543        for _ in 0..placed_count {
544            let position = cursor.read_u64()?;
545            let value = cursor.read_u64()?;
546            check_state_entry(rga.placed.last(), &(position, value))?;
547            rga.insert(position, value);
548        }
549        let tombstone_count = cursor.read_len()?;
550        check_collection_count(tombstone_count, limits)?;
551        for _ in 0..tombstone_count {
552            let position = cursor.read_u64()?;
553            check_state_entry(rga.tombstones.last(), &position)?;
554            rga.delete(position);
555        }
556        Ok(rga)
557    }
558}
559
560impl<D: WireEncode> WireEncode for Record<D> {
561    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
562        write_u8(out, TAG_RECORD);
563        write_u64(out, self.id.replica);
564        write_u64(out, self.id.sequence);
565        write_bytes(out, &self.delta.to_wire_bytes()?)?;
566        Ok(())
567    }
568}
569
570impl<D: WireDecode> WireDecode for Record<D> {
571    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
572        Self::decode_wire_with_collection_limits(cursor, CollectionLimits::WIRE_DEFAULT)
573    }
574
575    fn decode_wire_with_collection_limits(
576        cursor: &mut WireCursor<'_>,
577        limits: CollectionLimits,
578    ) -> Result<Self, WireError> {
579        Self::decode_wire_with_limits(cursor, limits, None)
580    }
581
582    fn decode_wire_with_limits(
583        cursor: &mut WireCursor<'_>,
584        limits: CollectionLimits,
585        max_records: Option<usize>,
586    ) -> Result<Self, WireError> {
587        read_tag(cursor, TAG_RECORD)?;
588        let id = RecordId {
589            replica: cursor.read_u64()?,
590            sequence: cursor.read_u64()?,
591        };
592        let delta_len = cursor.read_len()?;
593        let delta_bytes = cursor.read_exact(delta_len)?;
594        let mut delta_cursor = WireCursor::new(delta_bytes);
595        let delta = D::decode_wire_with_limits(&mut delta_cursor, limits, max_records)?;
596        if !delta_cursor.is_empty() {
597            return Err(WireError::TrailingBytes);
598        }
599        Ok(Record { id, delta })
600    }
601}
602
603// Shape-bearing frame: tag, body length, complemented length,
604// body (shape marker, schema, arity, record count and length-prefixed records),
605// CRC of length fields + body. Tag 0x02 and unshaped 0x03 frames return
606// LegacyEventLogFrame; EventLog::migrate_legacy_wire_bytes_for re-encodes them.
607// u32::MAX cannot be the count of a valid old body within a u32 frame length.
608// Frame lengths, count and CRC are little-endian u32. Check the length pair before trusting it,
609// then verify the CRC before decoding any record or invoking a payload decoder.
610impl<D: WireEncode + WireSchema> WireEncode for EventLog<D> {
611    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
612        Self::encode_records(self.replica_count, &self.records, out)
613    }
614}
615
616impl<D: WireDecode + WireSchema + PartialEq> WireDecode for EventLog<D> {
617    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
618        Self::decode_with(
619            cursor,
620            CollectionLimits::WIRE_DEFAULT,
621            DecodeLimits::default().max_records,
622            |_| {},
623            |index| {
624                let max_records = DecodeLimits::DEFAULT_MAX_RECORDS;
625                if index >= max_records {
626                    return Err(WireError::DefaultRecordLimitExceeded { max_records });
627                }
628                Ok(())
629            },
630        )
631    }
632
633    fn decode_wire_with_collection_limits(
634        cursor: &mut WireCursor<'_>,
635        limits: CollectionLimits,
636    ) -> Result<Self, WireError> {
637        Self::decode_with(
638            cursor,
639            limits,
640            DecodeLimits::default().max_records,
641            |_| {},
642            |index| {
643                let max_records = DecodeLimits::DEFAULT_MAX_RECORDS;
644                if index >= max_records {
645                    return Err(WireError::DefaultRecordLimitExceeded { max_records });
646                }
647                Ok(())
648            },
649        )
650    }
651
652    fn decode_wire_with_limits(
653        cursor: &mut WireCursor<'_>,
654        limits: CollectionLimits,
655        max_records: Option<usize>,
656    ) -> Result<Self, WireError> {
657        Self::decode_with(
658            cursor,
659            limits,
660            max_records,
661            |_| {},
662            |index| {
663                if let Some(max_records) = max_records {
664                    if index >= max_records {
665                        return Err(WireError::NestedRecordLimitExceeded { max_records });
666                    }
667                }
668                Ok(())
669            },
670        )
671    }
672}
673
674impl WireEncode for LwwRegisterDelta<u64> {
675    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
676        write_u8(out, TAG_LWW_REGISTER_DELTA_U64);
677        write_u64(out, self.timestamp);
678        write_u64(out, self.replica);
679        write_u64(out, self.value);
680        Ok(())
681    }
682}
683
684impl WireDecode for LwwRegisterDelta<u64> {
685    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
686        read_tag(cursor, TAG_LWW_REGISTER_DELTA_U64)?;
687        Ok(LwwRegisterDelta {
688            timestamp: cursor.read_u64()?,
689            replica: cursor.read_u64()?,
690            value: cursor.read_u64()?,
691        })
692    }
693}
694
695impl WireEncode for LwwRegister<u64> {
696    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
697        write_u8(out, TAG_LWW_REGISTER_U64);
698        match self.entry() {
699            Some(entry) => {
700                write_u8(out, 1);
701                write_u64(out, entry.dot.timestamp);
702                write_u64(out, entry.dot.replica);
703                write_u64(out, entry.value);
704            }
705            None => write_u8(out, 0),
706        }
707        Ok(())
708    }
709}
710
711impl WireDecode for LwwRegister<u64> {
712    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
713        read_tag(cursor, TAG_LWW_REGISTER_U64)?;
714        let present = cursor.read_u8()?;
715        match present {
716            0 => Ok(LwwRegister::new()),
717            1 => {
718                let mut register = LwwRegister::new();
719                register.set(cursor.read_u64()?, cursor.read_u64()?, cursor.read_u64()?);
720                Ok(register)
721            }
722            _ => Err(WireError::InvalidTag),
723        }
724    }
725}
726
727impl WireEncode for EnableWinsFlagDelta<u64> {
728    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
729        match self {
730            EnableWinsFlagDelta::Enable { token } => {
731                write_u8(out, TAG_ENABLE_WINS_FLAG_ENABLE_U64);
732                write_u64(out, *token);
733            }
734            EnableWinsFlagDelta::Disable { tokens } => {
735                write_u8(out, TAG_ENABLE_WINS_FLAG_DISABLE_U64);
736                let mut sorted_tokens = tokens.clone();
737                sorted_tokens.sort_unstable();
738                sorted_tokens.dedup();
739                write_len(out, sorted_tokens.len())?;
740                for token in &sorted_tokens {
741                    write_u64(out, *token);
742                }
743            }
744        }
745        Ok(())
746    }
747}
748
749impl WireDecode for EnableWinsFlagDelta<u64> {
750    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
751        Self::decode_wire_with_collection_limits(cursor, CollectionLimits::WIRE_DEFAULT)
752    }
753
754    fn decode_wire_with_collection_limits(
755        cursor: &mut WireCursor<'_>,
756        limits: CollectionLimits,
757    ) -> Result<Self, WireError> {
758        let tag = cursor.read_u8()?;
759        match tag {
760            TAG_ENABLE_WINS_FLAG_ENABLE_U64 => Ok(EnableWinsFlagDelta::Enable {
761                token: cursor.read_u64()?,
762            }),
763            TAG_ENABLE_WINS_FLAG_DISABLE_U64 => {
764                let len = cursor.read_len()?;
765                check_collection_count(len, limits)?;
766                let mut tokens = Vec::new();
767                for _ in 0..len {
768                    tokens.push(cursor.read_u64()?);
769                }
770                Ok(EnableWinsFlagDelta::Disable { tokens })
771            }
772            _ => Err(WireError::InvalidTag),
773        }
774    }
775}
776
777impl WireEncode for EnableWinsFlag<u64> {
778    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
779        write_u8(out, TAG_ENABLE_WINS_FLAG_U64);
780        write_len(out, self.enables.len())?;
781        for token in self.enables.iter() {
782            write_u64(out, *token);
783        }
784        write_len(out, self.tombstones.len())?;
785        for token in self.tombstones.iter() {
786            write_u64(out, *token);
787        }
788        Ok(())
789    }
790}
791
792impl WireDecode for EnableWinsFlag<u64> {
793    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
794        Self::decode_wire_with_collection_limits(cursor, CollectionLimits::WIRE_DEFAULT)
795    }
796
797    fn decode_wire_with_collection_limits(
798        cursor: &mut WireCursor<'_>,
799        limits: CollectionLimits,
800    ) -> Result<Self, WireError> {
801        read_tag(cursor, TAG_ENABLE_WINS_FLAG_U64)?;
802        let enable_len = cursor.read_len()?;
803        check_collection_count(enable_len, limits)?;
804        let mut flag = EnableWinsFlag::new();
805        for _ in 0..enable_len {
806            let token = cursor.read_u64()?;
807            check_state_entry(flag.enables.last(), &token)?;
808            flag.enable(token);
809        }
810        let tombstone_len = cursor.read_len()?;
811        check_collection_count(tombstone_len, limits)?;
812        for _ in 0..tombstone_len {
813            let token = cursor.read_u64()?;
814            check_state_entry(flag.tombstones.last(), &token)?;
815            flag.disable([token]);
816        }
817        Ok(flag)
818    }
819}
820
821impl WireEncode for LwwMapDelta<u64, u64> {
822    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
823        match self {
824            LwwMapDelta::Set {
825                key,
826                timestamp,
827                replica,
828                value,
829            } => {
830                write_u8(out, TAG_LWW_MAP_SET_U64);
831                write_u64(out, *key);
832                write_u64(out, *timestamp);
833                write_u64(out, *replica);
834                write_u64(out, *value);
835            }
836            LwwMapDelta::Remove {
837                key,
838                timestamp,
839                replica,
840            } => {
841                write_u8(out, TAG_LWW_MAP_REMOVE_U64);
842                write_u64(out, *key);
843                write_u64(out, *timestamp);
844                write_u64(out, *replica);
845            }
846        }
847        Ok(())
848    }
849}
850
851impl WireDecode for LwwMapDelta<u64, u64> {
852    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
853        let tag = cursor.read_u8()?;
854        match tag {
855            TAG_LWW_MAP_SET_U64 => Ok(LwwMapDelta::Set {
856                key: cursor.read_u64()?,
857                timestamp: cursor.read_u64()?,
858                replica: cursor.read_u64()?,
859                value: cursor.read_u64()?,
860            }),
861            TAG_LWW_MAP_REMOVE_U64 => Ok(LwwMapDelta::Remove {
862                key: cursor.read_u64()?,
863                timestamp: cursor.read_u64()?,
864                replica: cursor.read_u64()?,
865            }),
866            _ => Err(WireError::InvalidTag),
867        }
868    }
869}
870
871impl WireEncode for LwwMap<u64, u64> {
872    fn encode_wire(&self, out: &mut Vec<u8>) -> Result<(), WireError> {
873        write_u8(out, TAG_LWW_MAP_U64);
874        write_len(out, self.entries.len())?;
875        for (key, entry) in self.entries.iter() {
876            write_u64(out, *key);
877            write_u64(out, entry.dot.timestamp);
878            write_u64(out, entry.dot.replica);
879            write_u64(out, entry.value);
880        }
881        write_len(out, self.removals.len())?;
882        for (key, dot) in self.removals.iter() {
883            write_u64(out, *key);
884            write_u64(out, dot.timestamp);
885            write_u64(out, dot.replica);
886        }
887        Ok(())
888    }
889}
890
891impl WireDecode for LwwMap<u64, u64> {
892    fn decode_wire(cursor: &mut WireCursor<'_>) -> Result<Self, WireError> {
893        Self::decode_wire_with_collection_limits(cursor, CollectionLimits::WIRE_DEFAULT)
894    }
895
896    fn decode_wire_with_collection_limits(
897        cursor: &mut WireCursor<'_>,
898        limits: CollectionLimits,
899    ) -> Result<Self, WireError> {
900        read_tag(cursor, TAG_LWW_MAP_U64)?;
901        let entry_len = cursor.read_len()?;
902        check_collection_count(entry_len, limits)?;
903        let mut map = LwwMap::new();
904        for _ in 0..entry_len {
905            let key = cursor.read_u64()?;
906            check_state_entry(map.entries.last_key_value().map(|(key, _)| key), &key)?;
907            map.set(
908                key,
909                cursor.read_u64()?,
910                cursor.read_u64()?,
911                cursor.read_u64()?,
912            );
913        }
914        let removal_len = cursor.read_len()?;
915        check_collection_count(removal_len, limits)?;
916        for _ in 0..removal_len {
917            let key = cursor.read_u64()?;
918            check_state_entry(map.removals.last_key_value().map(|(key, _)| key), &key)?;
919            map.remove(key, cursor.read_u64()?, cursor.read_u64()?);
920        }
921        Ok(map)
922    }
923}
924
925#[cfg(test)]
926mod version_vector_wire_tests {
927    use super::*;
928    use alloc::vec;
929
930    #[test]
931    fn test_module_follows_wire_implementations() {
932        let source = include_str!("wire.rs");
933        let test_module = source
934            .find("\n#[cfg(test)]\nmod version_vector_wire_tests")
935            .unwrap();
936        let last_impl = source.rfind("\nimpl ").unwrap();
937        assert!(last_impl < test_module);
938    }
939
940    #[test]
941    fn version_vector_round_trip_is_structural_and_deterministic() {
942        let first = VersionVector::from_peer_prefixes(
943            &BTreeMap::from([(9, 3), (2, 7), (5, u64::MAX)]),
944            &BTreeSet::from([9, 1, 5]),
945        )
946        .unwrap();
947        let second = VersionVector::from_peer_prefixes(
948            &[(5, u64::MAX), (9, 3), (2, 7)].into_iter().collect(),
949            &[5, 1, 9].into_iter().collect(),
950        )
951        .unwrap();
952        let bytes = first.to_wire_bytes().unwrap();
953        assert_eq!(bytes, second.to_wire_bytes().unwrap());
954        assert_eq!(VersionVector::from_wire_bytes(&bytes).unwrap(), first);
955    }
956
957    #[test]
958    fn version_vector_claimed_author_count_exceeding_default_is_rejected() {
959        let mut tampered = vec![TAG_VERSION_VECTOR];
960        write_len(
961            &mut tampered,
962            VersionVectorLimits::WIRE_DEFAULT.max_authors.unwrap() + 1,
963        )
964        .unwrap();
965        assert_eq!(
966            VersionVector::from_wire_bytes(&tampered),
967            Err(WireError::VersionAuthorLimitExceeded { max_authors: 4096 })
968        );
969        let valid = VersionVector::from_peer_prefixes(
970            &BTreeMap::from([(1, 2), (3, 4)]),
971            &BTreeSet::from([1]),
972        )
973        .unwrap();
974        assert_eq!(
975            VersionVector::from_wire_bytes(&valid.to_wire_bytes().unwrap()).unwrap(),
976            valid
977        );
978    }
979
980    #[test]
981    fn version_vector_custom_limits_reject_zeros_and_duplicate_authors() {
982        let value =
983            VersionVector::from_peer_prefixes(&BTreeMap::from([(1, 2)]), &BTreeSet::from([1, 2]))
984                .unwrap();
985        assert_eq!(
986            VersionVector::from_wire_bytes_with_limits(
987                &value.to_wire_bytes().unwrap(),
988                VersionVectorLimits {
989                    max_authors: Some(1),
990                    max_zero_replicas: Some(1)
991                },
992            ),
993            Err(WireError::VersionZeroReplicaLimitExceeded {
994                max_zero_replicas: 1
995            })
996        );
997        let mut duplicates = vec![TAG_VERSION_VECTOR];
998        write_len(&mut duplicates, 2).unwrap();
999        write_u64(&mut duplicates, 7);
1000        write_u64(&mut duplicates, 1);
1001        write_u64(&mut duplicates, 7);
1002        write_u64(&mut duplicates, 2);
1003        write_len(&mut duplicates, 0).unwrap();
1004        assert_eq!(
1005            VersionVector::from_wire_bytes(&duplicates),
1006            Err(WireError::NonCanonicalVersionVector)
1007        );
1008    }
1009}
safemesh_crdt/wire.rs

safemesh_crdt/
wire.rs
