blockworx/log/
id.rs

1//! Identities and the total order that decides every merge.
2//!
3//! The order is the whole convergence argument, so it is spelled out here
4//! rather than reconstructed at each comparison site: a register keeps the
5//! write with the greatest [`WriteOrder`] it has ever seen, and `max` over a
6//! total order is commutative, associative and idempotent. Delivery order,
7//! grouping and duplicates therefore cannot change the result.
8
9use std::fmt;
10
11use serde::{Deserialize, Deserializer, Serialize, Serializer};
12use uuid::Uuid;
13
14/// One element of the document — a block, pin, route, annotation. Random v4 at
15/// creation and never reused, so two replicas creating elements concurrently
16/// cannot collide, and a duplicated subtree is genuinely independent of its
17/// source rather than sharing its identity.
18#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
19pub struct ElementId(Uuid);
20
21impl ElementId {
22    /// The containment tree's root. Not a drawable element: it is the parent
23    /// the top block points at, so that "every element has exactly one parent"
24    /// holds without the root being a special case in the fold.
25    pub const DOCUMENT: ElementId = ElementId(Uuid::nil());
26
27    pub fn new() -> Self {
28        Self(Uuid::new_v4())
29    }
30
31    pub const fn from_uuid(id: Uuid) -> Self {
32        Self(id)
33    }
34
35    pub const fn as_uuid(self) -> Uuid {
36        self.0
37    }
38
39    pub fn is_document(self) -> bool {
40        self == Self::DOCUMENT
41    }
42}
43
44impl Default for ElementId {
45    fn default() -> Self {
46        Self::new()
47    }
48}
49
50/// One replica — an install, not a human. A user editing from two machines is
51/// two actors, which is correct: their edits are genuinely concurrent.
52#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
53pub struct ActorId(Uuid);
54
55impl ActorId {
56    pub fn new() -> Self {
57        Self(Uuid::new_v4())
58    }
59
60    pub const fn from_uuid(id: Uuid) -> Self {
61        Self(id)
62    }
63
64    pub const fn as_uuid(self) -> Uuid {
65        self.0
66    }
67}
68
69impl Default for ActorId {
70    fn default() -> Self {
71        Self::new()
72    }
73}
74
75/// A logical clock reading. Counts causality, not time; wall time is display
76/// metadata and never load-bearing.
77#[derive(
78    Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Default, Serialize, Deserialize,
79)]
80pub struct Lamport(u64);
81
82impl Lamport {
83    pub const ZERO: Lamport = Lamport(0);
84
85    pub const fn new(n: u64) -> Self {
86        Self(n)
87    }
88
89    pub const fn get(self) -> u64 {
90        self.0
91    }
92
93    /// Saturating so a replica that somehow reached `u64::MAX` degrades into
94    /// ties broken by actor rather than wrapping into the distant past.
95    fn next(self) -> Self {
96        Self(self.0.saturating_add(1))
97    }
98}
99
100/// When a change was authored, in the only sense that matters for merging.
101///
102/// `lamport` first, `actor` as the tiebreak: two replicas can reach the same
103/// Lamport value independently, but no two distinct actors share an id, so the
104/// order is total across every pair of changes from different replicas. One
105/// actor never issues two changes at the same Lamport value.
106#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Serialize, Deserialize)]
107pub struct Stamp {
108    pub lamport: Lamport,
109    pub actor: ActorId,
110}
111
112/// A command's position within its change's batch.
113#[derive(
114    Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Default, Serialize, Deserialize,
115)]
116pub struct BatchIndex(u32);
117
118impl BatchIndex {
119    pub const FIRST: BatchIndex = BatchIndex(0);
120
121    pub const fn new(n: u32) -> Self {
122        Self(n)
123    }
124
125    pub const fn get(self) -> u32 {
126        self.0
127    }
128}
129
130/// The register comparison key — the full total order on writes.
131///
132/// [`Stamp`] alone is *not* enough: every command in one change shares that
133/// change's stamp, so a change writing the same register twice would leave the
134/// two writes unordered. The batch index is the final tiebreak.
135#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug, Serialize, Deserialize)]
136pub struct WriteOrder {
137    pub stamp: Stamp,
138    pub batch: BatchIndex,
139}
140
141impl WriteOrder {
142    pub const fn new(stamp: Stamp, batch: BatchIndex) -> Self {
143        Self { stamp, batch }
144    }
145}
146
147/// Content address of a change: blake3 over its canonical encoding, which
148/// includes its parent hashes, so a hash commits to the entire history behind
149/// it.
150#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
151pub struct ChangeHash([u8; 32]);
152
153impl ChangeHash {
154    pub const fn from_bytes(bytes: [u8; 32]) -> Self {
155        Self(bytes)
156    }
157
158    pub const fn as_bytes(&self) -> &[u8; 32] {
159        &self.0
160    }
161}
162
163/// Content address of an image payload. Deliberately content-derived rather
164/// than random: an asset is a value, so two replicas importing the same image
165/// must reference one asset. Truncated to 64 bits, matching the `<hash>.<ext>`
166/// filenames already written into a container's `assets/`.
167#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
168pub struct AssetId([u8; 8]);
169
170impl AssetId {
171    pub const fn from_bytes(bytes: [u8; 8]) -> Self {
172        Self(bytes)
173    }
174
175    pub const fn as_bytes(&self) -> &[u8; 8] {
176        &self.0
177    }
178
179    pub fn of(content: &[u8]) -> Self {
180        let full = blake3::hash(content);
181        let mut truncated = [0u8; 8];
182        truncated.copy_from_slice(&full.as_bytes()[..8]);
183        Self(truncated)
184    }
185}
186
187/// This replica's Lamport clock.
188///
189/// Held by the session that appends changes. It is not part of document state
190/// and never enters the fold — a clock inside a pure replay would make replay
191/// depend on when it ran.
192#[derive(Clone, Copy, Debug)]
193pub struct Clock {
194    actor: ActorId,
195    lamport: Lamport,
196}
197
198impl Clock {
199    pub const fn new(actor: ActorId) -> Self {
200        Self {
201            actor,
202            lamport: Lamport::ZERO,
203        }
204    }
205
206    pub const fn actor(self) -> ActorId {
207        self.actor
208    }
209
210    pub const fn lamport(self) -> Lamport {
211        self.lamport
212    }
213
214    /// Take account of a change authored elsewhere. Every ingested change must
215    /// pass through here, or a later local change could be stamped as though it
216    /// had not seen it.
217    pub fn observe(&mut self, seen: Lamport) {
218        self.lamport = self.lamport.max(seen);
219    }
220
221    /// Advance and stamp a new local change.
222    pub fn tick(&mut self) -> Stamp {
223        self.lamport = self.lamport.next();
224        Stamp {
225            lamport: self.lamport,
226            actor: self.actor,
227        }
228    }
229}
230
231fn write_hex(f: &mut fmt::Formatter<'_>, bytes: &[u8]) -> fmt::Result {
232    for byte in bytes {
233        write!(f, "{byte:02x}")?;
234    }
235    Ok(())
236}
237
238impl fmt::Display for ElementId {
239    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
240        if self.is_document() {
241            return f.write_str("document");
242        }
243        // The first four bytes are enough to read a log by eye; the full uuid
244        // is `as_uuid`.
245        write_hex(f, &self.0.as_bytes()[..4])
246    }
247}
248
249impl fmt::Display for ActorId {
250    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
251        write_hex(f, &self.0.as_bytes()[..4])
252    }
253}
254
255impl fmt::Display for ChangeHash {
256    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
257        write_hex(f, &self.0[..8])
258    }
259}
260
261impl fmt::Display for AssetId {
262    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
263        write_hex(f, &self.0)
264    }
265}
266
267impl fmt::Debug for ElementId {
268    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
269        write!(f, "{self}")
270    }
271}
272
273impl fmt::Debug for ActorId {
274    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
275        write!(f, "{self}")
276    }
277}
278
279impl fmt::Debug for ChangeHash {
280    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
281        write!(f, "{self}")
282    }
283}
284
285impl fmt::Debug for AssetId {
286    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
287        write!(f, "{self}")
288    }
289}
290
291/// Hex text rather than an array of numbers, so a log is legible when the
292/// storage format is. Fixed width, so the encoding is one-to-one.
293fn hex_serialize<S: Serializer>(bytes: &[u8], s: S) -> Result<S::Ok, S::Error> {
294    use std::fmt::Write as _;
295    let mut out = String::with_capacity(bytes.len() * 2);
296    for byte in bytes {
297        let _ = write!(out, "{byte:02x}");
298    }
299    s.serialize_str(&out)
300}
301
302fn hex_deserialize<'de, D: Deserializer<'de>, const N: usize>(d: D) -> Result<[u8; N], D::Error> {
303    use serde::de::Error as _;
304    let text = String::deserialize(d)?;
305    if text.len() != N * 2 {
306        return Err(D::Error::custom(format!(
307            "expected {} hex characters, found {}",
308            N * 2,
309            text.len()
310        )));
311    }
312    let mut out = [0u8; N];
313    for (slot, pair) in out.iter_mut().zip(text.as_bytes().chunks_exact(2)) {
314        let digits = std::str::from_utf8(pair).map_err(D::Error::custom)?;
315        *slot = u8::from_str_radix(digits, 16).map_err(D::Error::custom)?;
316    }
317    Ok(out)
318}
319
320impl Serialize for ChangeHash {
321    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
322        hex_serialize(&self.0, s)
323    }
324}
325
326impl<'de> Deserialize<'de> for ChangeHash {
327    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
328        hex_deserialize::<D, 32>(d).map(Self)
329    }
330}
331
332impl Serialize for AssetId {
333    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
334        hex_serialize(&self.0, s)
335    }
336}
337
338impl<'de> Deserialize<'de> for AssetId {
339    fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
340        hex_deserialize::<D, 8>(d).map(Self)
341    }
342}
343
344#[cfg(test)]
345mod tests {
346    use super::*;
347
348    fn actor(byte: u8) -> ActorId {
349        ActorId::from_uuid(Uuid::from_bytes([byte; 16]))
350    }
351
352    /// The property the whole merge rests on: any two writes from distinct
353    /// actors are ordered, so `max` over them is well defined.
354    #[test]
355    fn equal_lamports_are_broken_by_actor() {
356        let (low, high) = (actor(1), actor(2));
357        assert!(low < high, "the fixture's actors must differ to order them");
358
359        let a = Stamp {
360            lamport: Lamport::new(7),
361            actor: low,
362        };
363        let b = Stamp {
364            lamport: Lamport::new(7),
365            actor: high,
366        };
367        assert!(a < b);
368    }
369
370    /// Lamport dominates the actor tiebreak — otherwise a replica with a
371    /// high-sorting id would win every race regardless of causality.
372    #[test]
373    fn lamport_outranks_actor() {
374        let earlier = Stamp {
375            lamport: Lamport::new(7),
376            actor: actor(9),
377        };
378        let later = Stamp {
379            lamport: Lamport::new(8),
380            actor: actor(1),
381        };
382        assert!(earlier < later);
383    }
384
385    /// Two writes in one change share a stamp, so only the batch index
386    /// separates them.
387    #[test]
388    fn one_changes_writes_are_ordered_by_batch_index() {
389        let stamp = Stamp {
390            lamport: Lamport::new(3),
391            actor: actor(4),
392        };
393        let first = WriteOrder::new(stamp, BatchIndex::new(0));
394        let second = WriteOrder::new(stamp, BatchIndex::new(1));
395        assert_eq!(
396            first.stamp, second.stamp,
397            "the fixture must share a stamp or the batch index isn't what's ordering them"
398        );
399        assert!(first < second);
400    }
401
402    #[test]
403    fn observing_a_remote_clock_never_moves_time_backwards() {
404        let mut clock = Clock::new(actor(1));
405        clock.observe(Lamport::new(10));
406        assert_eq!(clock.tick().lamport, Lamport::new(11));
407
408        clock.observe(Lamport::new(2));
409        assert_eq!(
410            clock.tick().lamport,
411            Lamport::new(12),
412            "a stale remote reading must not rewind the clock"
413        );
414    }
415
416    #[test]
417    fn a_fresh_element_id_is_not_the_document_root() {
418        let id = ElementId::new();
419        assert!(!id.is_document());
420        assert!(ElementId::DOCUMENT.is_document());
421        assert_ne!(id, ElementId::new());
422    }
423
424    #[test]
425    fn asset_ids_are_content_derived() {
426        assert_eq!(AssetId::of(b"one"), AssetId::of(b"one"));
427        assert_ne!(AssetId::of(b"one"), AssetId::of(b"two"));
428        assert_eq!(AssetId::of(b"one").to_string().len(), 16);
429    }
430}
blockworx/log/id.rs

blockworx/log/
id.rs
