safemesh_crdt/version_vector.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::RecordId;
6use alloc::collections::{BTreeMap, BTreeSet};
7
8/// Per-replica contiguous prefixes for anti-entropy pulls.
9///
10/// `get(replica) == n` means every sequence `1..=n` for that replica is known.
11/// Later records that arrive before earlier records must not advance this
12/// prefix, otherwise `since` could hide gaps. Sequence zero is acknowledged
13/// independently by `zero_replicas`; it is not implied by a positive prefix.
14/// The built-in wire codec carries both [`Self::entries`] and
15/// [`Self::zero_replicas`]. Custom version exchanges must carry both too. A
16/// receiver can rebuild the vector with [`Self::from_peer_prefixes`] after
17/// checking application-owned author and zero-acknowledgement budgets, as below;
18/// record payloads are not required.
19/// Legacy prefix-only exchanges cannot acknowledge zeros and will keep receiving
20/// them until upgraded.
21///
22/// These are claims about the sending peer's log. The caller is responsible for
23/// their truth: neither `observe` nor `since` checks possession of record payloads.
24/// Fabricated IDs can acknowledge missing records and cause `since` to omit them.
25/// Use [`crate::EventLog::version`] to derive claims from records actually admitted.
26///
27/// Checked reconstruction validates each positive prefix, then clones both
28/// collections: for `r` authors and `z` zero acknowledgements it takes O(r + z)
29/// time and O(r + z) additional space, independently of the claimed sequences.
30/// Bound both collection sizes before reconstruction and enforce transport byte
31/// limits before decoding to bound the input allocation itself.
32/// Zero-valued prefix entries are noncanonical and are rejected by the checked
33/// constructor; only `zero_replicas()` acknowledges sequence zero.
34///
35/// ```
36/// use safemesh_crdt::{EventLog, VersionVector};
37/// use std::collections::{BTreeMap, BTreeSet};
38///
39/// fn receive(
40/// prefixes: &BTreeMap<u64, u64>, zeros: &BTreeSet<u64>,
41/// author_budget: usize, zero_budget: usize,
42/// ) -> Result<VersionVector, &'static str> {
43/// if prefixes.len() > author_budget || zeros.len() > zero_budget {
44/// return Err("peer version exceeds application budget");
45/// }
46/// VersionVector::from_peer_prefixes(prefixes, zeros).map_err(|_| "invalid prefix")
47/// }
48///
49/// let mut peer = EventLog::new();
50/// peer.append(&mut safemesh_crdt::GSet::new(), 7, 42u64).unwrap();
51/// // Carry both collections over the application's transport.
52/// let prefixes = peer.version().entries().clone();
53/// let zeros = peer.version().zero_replicas().clone();
54/// // This example has one publisher and allocates no sequence-zero records.
55/// let received = receive(&prefixes, &zeros, 1, 0).unwrap();
56/// assert_eq!(&received, peer.version());
57/// assert_eq!(peer.since(&received), peer.since(peer.version()));
58/// ```
59#[derive(Clone, Debug, Default, PartialEq, Eq)]
60pub struct VersionVector {
61 entries: BTreeMap<u64, u64>,
62 zero_replicas: BTreeSet<u64>,
63}
64
65/// Optional collection budgets for [`VersionVector::from_peer_prefixes_with_limits`].
66#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
67pub struct VersionVectorLimits {
68 /// Maximum positive-prefix entries. `None` is unbounded; `Some(0)` requires none.
69 pub max_authors: Option<usize>,
70 /// Maximum sequence-zero acknowledgements, independently of prefix entries.
71 /// `None` is unbounded; `Some(0)` requires none.
72 pub max_zero_replicas: Option<usize>,
73}
74
75impl VersionVectorLimits {
76 /// Default wire entry ceilings. Callers can supply tighter or larger budgets
77 /// with [`VersionVector::from_wire_bytes_with_limits`].
78 pub const WIRE_DEFAULT: Self = Self {
79 max_authors: Some(4096),
80 max_zero_replicas: Some(4096),
81 };
82}
83
84/// Failure from opt-in bounded peer-version reconstruction.
85/// Separate from [`VersionVectorError`] to preserve existing exhaustive matches.
86#[derive(Clone, Copy, Debug, PartialEq, Eq)]
87pub enum VersionVectorLimitError {
88 Prefix(VersionVectorError),
89 AuthorLimitExceeded { max_authors: usize },
90 ZeroReplicaLimitExceeded { max_zero_replicas: usize },
91}
92
93impl core::fmt::Display for VersionVectorLimitError {
94 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
95 match self {
96 Self::Prefix(error) => error.fmt(f),
97 Self::AuthorLimitExceeded { max_authors } => {
98 write!(f, "peer version exceeds author limit {max_authors}")
99 }
100 Self::ZeroReplicaLimitExceeded { max_zero_replicas } => {
101 write!(
102 f,
103 "peer version exceeds zero-acknowledgement limit {max_zero_replicas}"
104 )
105 }
106 }
107 }
108}
109
110impl core::error::Error for VersionVectorLimitError {
111 fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
112 match self {
113 Self::Prefix(error) => Some(error),
114 _ => None,
115 }
116 }
117}
118
119#[derive(Clone, Copy, Debug, PartialEq, Eq)]
120pub enum VersionVectorError {
121 /// Positive prefixes are canonical; sequence zero belongs in zero_replicas.
122 ZeroPrefix { replica: u64 },
123}
124
125impl core::fmt::Display for VersionVectorError {
126 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
127 match self {
128 VersionVectorError::ZeroPrefix { replica } => {
129 write!(f, "replica {replica} has a noncanonical zero prefix")
130 }
131 }
132 }
133}
134
135impl core::error::Error for VersionVectorError {}
136
137impl VersionVector {
138 pub fn new() -> Self {
139 VersionVector {
140 entries: BTreeMap::new(),
141 zero_replicas: BTreeSet::new(),
142 }
143 }
144
145 /// Reconstruct a vector from a peer's canonical positive prefixes and
146 /// independent sequence-zero acknowledgements.
147 ///
148 /// Accepts every positive `u64` prefix, including `u64::MAX`, without
149 /// iterating over sequences. Zero remains noncanonical: use `zero_replicas`.
150 /// This method imposes no collection budget. Use
151 /// [`Self::from_peer_prefixes_with_limits`] to enforce author and
152 /// zero-acknowledgement budgets before validation or cloning.
153 /// Callers must retain three application-owned input limits:
154 /// - author count bounds prefix validation and map cloning;
155 /// - zero-acknowledgement count bounds set cloning;
156 /// - encoded byte size, enforced before decoding, bounds input allocation
157 /// and parsing (including duplicate entries before collection deduplication).
158 ///
159 /// The built-in wire decoder has default entry ceilings; callers can choose
160 /// their own limits. This constructor imposes no universal application budget.
161 /// Reconstruction takes O(entries.len() + zero_replicas.len()) time and space.
162 pub fn from_peer_prefixes(
163 entries: &BTreeMap<u64, u64>,
164 zero_replicas: &BTreeSet<u64>,
165 ) -> Result<Self, VersionVectorError> {
166 for (&replica, &prefix) in entries {
167 if prefix == 0 {
168 return Err(VersionVectorError::ZeroPrefix { replica });
169 }
170 }
171
172 Ok(Self {
173 entries: entries.clone(),
174 zero_replicas: zero_replicas.clone(),
175 })
176 }
177
178 /// Reconstruct a peer version with caller-supplied collection budgets.
179 ///
180 /// Checks author count first, then zero-acknowledgement count, before
181 /// validating prefixes or cloning either collection. If both budgets are
182 /// exceeded, returns [`VersionVectorLimitError::AuthorLimitExceeded`].
183 /// No partial vector is returned. Exact limits are accepted.
184 /// [`VersionVectorLimits::default`] preserves the unbounded constructor's
185 /// values and wraps its errors in [`VersionVectorLimitError::Prefix`].
186 /// These checks do not bound decoding or input allocation: callers must
187 /// enforce encoded-byte and entry budgets at their decoding boundary too.
188 pub fn from_peer_prefixes_with_limits(
189 entries: &BTreeMap<u64, u64>,
190 zero_replicas: &BTreeSet<u64>,
191 limits: VersionVectorLimits,
192 ) -> Result<Self, VersionVectorLimitError> {
193 if let Some(max_authors) = limits.max_authors {
194 if entries.len() > max_authors {
195 return Err(VersionVectorLimitError::AuthorLimitExceeded { max_authors });
196 }
197 }
198 if let Some(max_zero_replicas) = limits.max_zero_replicas {
199 if zero_replicas.len() > max_zero_replicas {
200 return Err(VersionVectorLimitError::ZeroReplicaLimitExceeded {
201 max_zero_replicas,
202 });
203 }
204 }
205 Self::from_peer_prefixes(entries, zero_replicas).map_err(VersionVectorLimitError::Prefix)
206 }
207
208 pub fn get(&self, replica: u64) -> u64 {
209 self.entries.get(&replica).copied().unwrap_or(0)
210 }
211
212 /// Advance this prefix only when `id` is the next contiguous sequence.
213 /// A missing sequence keeps later records outside the prefix until it arrives.
214 ///
215 /// Use `EventLog` to ingest out-of-order records; the log remembers gaps
216 /// and advances this vector once the prefix is complete.
217 pub fn observe(&mut self, id: RecordId) {
218 if id.sequence == 0 {
219 self.zero_replicas.insert(id.replica);
220 } else if self.get(id.replica).checked_add(1) == Some(id.sequence) {
221 self.set(id.replica, id.sequence);
222 }
223 }
224
225 /// Return true for an acknowledged zero or a sequence in the writer's contiguous prefix.
226 pub fn includes(&self, id: RecordId) -> bool {
227 if id.sequence == 0 {
228 self.zero_replicas.contains(&id.replica)
229 } else {
230 self.get(id.replica) >= id.sequence
231 }
232 }
233
234 /// Positive contiguous prefixes only; does not describe zero possession.
235 pub fn entries(&self) -> &BTreeMap<u64, u64> {
236 &self.entries
237 }
238
239 /// Authors whose sequence-zero record has been observed.
240 pub fn zero_replicas(&self) -> &BTreeSet<u64> {
241 &self.zero_replicas
242 }
243
244 pub(super) fn set(&mut self, replica: u64, sequence: u64) {
245 if sequence == 0 {
246 self.entries.remove(&replica);
247 } else {
248 self.entries.insert(replica, sequence);
249 }
250 }
251}