subetha_pointers/
versioned_pointer.rs

1//! Versioned/MVCC pointers - time-travel addressing for snapshot
2//! isolation, immutable trees, and distributed clocks.
3//!
4//! Three pointer flavours sharing a common shape `(version, target)`:
5//!
6//! | Type                     | Version | Use case                                |
7//! |--------------------------|---------|-----------------------------------------|
8//! | [`VersionedPointer<T>`]  | u64     | Local MVCC, snapshot isolation          |
9//! | [`HlcVersionedPointer<T>`] | (u64 physical, u64 logical) | Distributed (CockroachDB-style HLC) |
10//! | [`VectorClockPointer<T, N>`] | `[u64; N]` per-node | Per-node causal ordering (Riak-style) |
11//!
12//! Plus a [`VersionedChain<T>`] linked-list of `VersionedNode<T>`s
13//! that retains all historical versions for time-travel queries.
14//!
15//! # The K_temporal / K_cascade interplay
16//!
17//! `VersionedPointer<T>` is a single-version snapshot.
18//! `VersionedPointer<VersionedPointer<T>>` is the K_cascade = 2 case:
19//! outer carries coarse (physical) time, inner carries fine (logical)
20//! counter. This is exactly the Hybrid Logical Clock pattern that
21//! CockroachDB and Spanner use - here exposed as a first-class
22//! typed primitive via [`HlcVersionedPointer<T>`].
23
24use std::cmp::Ordering;
25use std::sync::Arc;
26
27// =========================================================
28// Monotonic VersionedPointer<T>
29// =========================================================
30
31/// Pointer + monotonic u64 version. Used for snapshot-isolation
32/// reads: visible at a query snapshot when `self.version <= snapshot`.
33#[derive(Debug, Clone)]
34pub struct VersionedPointer<T> {
35    version: u64,
36    target: Arc<T>,
37}
38
39impl<T> VersionedPointer<T> {
40    /// Direction signature of `VersionedPointer<T>`. Engages the
41    /// `K_version` axis (u64 version counter stored at slot for
42    /// ABA-safe CAS and MVCC visibility checks).
43    pub const SIGNATURE: subetha_core::AxisMask = subetha_core::AxisMask::from_axes(
44        &[subetha_core::Axis::Version],
45    );
46
47    pub const fn new(target: Arc<T>, version: u64) -> Self {
48        Self { version, target }
49    }
50
51    #[inline]
52    pub const fn version(&self) -> u64 { self.version }
53
54    #[inline]
55    pub fn target(&self) -> &Arc<T> { &self.target }
56
57    /// True when this pointer is observable at `snapshot_version`.
58    /// Standard MVCC visibility: visible if its version is at or
59    /// before the snapshot.
60    #[inline]
61    pub const fn visible_at(&self, snapshot_version: u64) -> bool {
62        self.version <= snapshot_version
63    }
64
65    /// Return the target if it is visible at `snapshot_version`.
66    pub fn read_at(&self, snapshot_version: u64) -> Option<&T> {
67        if self.visible_at(snapshot_version) {
68            Some(&self.target)
69        } else {
70            None
71        }
72    }
73
74    /// Replace the target with a new version. Returns the previous
75    /// version. Panics if `new_version <= self.version` because
76    /// MVCC requires monotonic version growth.
77    pub fn replace(&mut self, new_target: Arc<T>, new_version: u64) -> u64 {
78        assert!(
79            new_version > self.version,
80            "MVCC version must be strictly monotonic: {} -> {}",
81            self.version, new_version
82        );
83        let old = self.version;
84        self.version = new_version;
85        self.target = new_target;
86        old
87    }
88}
89
90impl<T> PartialEq for VersionedPointer<T> {
91    fn eq(&self, other: &Self) -> bool {
92        self.version == other.version && Arc::ptr_eq(&self.target, &other.target)
93    }
94}
95
96impl<T> PartialOrd for VersionedPointer<T> {
97    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
98        Some(self.version.cmp(&other.version))
99    }
100}
101
102// =========================================================
103// HybridLogicalClock + HlcVersionedPointer<T>
104// =========================================================
105
106/// Hybrid Logical Clock: (physical timestamp, logical counter) pair.
107/// Combines wall-clock time (microsecond resolution typical) with a
108/// per-node monotonic counter that breaks ties between events
109/// recorded in the same physical instant.
110///
111/// This is the K_cascade = 2 case of versioning: the outer level
112/// (physical) is coarse; the inner level (logical) refines tied
113/// physical timestamps. Same architectural pattern as Umbra prefix
114/// + actual content, applied to time.
115#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
116pub struct HybridLogicalClock {
117    pub physical: u64,
118    pub logical: u64,
119}
120
121impl HybridLogicalClock {
122    pub const fn new(physical: u64, logical: u64) -> Self {
123        Self { physical, logical }
124    }
125
126    /// Construct from the current wall clock (microseconds since
127    /// UNIX epoch), with logical counter starting at 0.
128    pub fn now() -> Self {
129        let now = std::time::SystemTime::now()
130            .duration_since(std::time::UNIX_EPOCH)
131            .map(|d| d.as_micros() as u64)
132            .unwrap_or(0);
133        Self { physical: now, logical: 0 }
134    }
135
136    /// Advance: increment logical counter (same physical) or jump to
137    /// `new_physical` if larger. Used by an HLC source to record a
138    /// new event.
139    pub fn advance(&self, new_physical: u64) -> Self {
140        if new_physical > self.physical {
141            Self { physical: new_physical, logical: 0 }
142        } else {
143            Self { physical: self.physical, logical: self.logical + 1 }
144        }
145    }
146
147    /// Merge with a received event: take max physical; advance logical
148    /// if needed. This is the receiver-side HLC update.
149    pub fn merge(&self, received: &Self, local_physical: u64) -> Self {
150        let max_phys = self.physical.max(received.physical).max(local_physical);
151        let new_logical = if max_phys == self.physical && max_phys == received.physical {
152            self.logical.max(received.logical) + 1
153        } else if max_phys == self.physical {
154            self.logical + 1
155        } else if max_phys == received.physical {
156            received.logical + 1
157        } else {
158            0
159        };
160        Self { physical: max_phys, logical: new_logical }
161    }
162}
163
164impl PartialOrd for HybridLogicalClock {
165    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
166        Some(self.cmp(other))
167    }
168}
169
170impl Ord for HybridLogicalClock {
171    fn cmp(&self, other: &Self) -> Ordering {
172        // Lexicographic: physical first, then logical.
173        self.physical.cmp(&other.physical)
174            .then(self.logical.cmp(&other.logical))
175    }
176}
177
178/// Pointer + HLC. Composes [`VersionedPointer`] with the cascade
179/// structure of `(physical, logical)`. Two-level rejection on
180/// `visible_at`: physical mismatch rejects fast, logical compares
181/// only when physical ties.
182#[derive(Debug, Clone)]
183pub struct HlcVersionedPointer<T> {
184    clock: HybridLogicalClock,
185    target: Arc<T>,
186}
187
188impl<T> HlcVersionedPointer<T> {
189    /// Direction signature of `HlcVersionedPointer<T>`. Engages the
190    /// `K_version` axis (hybrid-logical-clock version stored at
191    /// slot for distributed MVCC).
192    pub const SIGNATURE: subetha_core::AxisMask = subetha_core::AxisMask::from_axes(
193        &[subetha_core::Axis::Version],
194    );
195
196    pub const fn new(target: Arc<T>, clock: HybridLogicalClock) -> Self {
197        Self { clock, target }
198    }
199
200    #[inline]
201    pub const fn clock(&self) -> HybridLogicalClock { self.clock }
202
203    #[inline]
204    pub fn target(&self) -> &Arc<T> { &self.target }
205
206    pub fn visible_at(&self, snapshot: HybridLogicalClock) -> bool {
207        self.clock <= snapshot
208    }
209
210    pub fn read_at(&self, snapshot: HybridLogicalClock) -> Option<&T> {
211        if self.visible_at(snapshot) { Some(&self.target) } else { None }
212    }
213}
214
215// =========================================================
216// VectorClock + VectorClockPointer<T, N>
217// =========================================================
218
219/// Per-node monotonic counters. `N` is the number of nodes in the
220/// system; `clock[i]` is node `i`'s observed event count. Causal
221/// ordering: `a` causally precedes `b` when every component of `a`
222/// is <= the corresponding component of `b`, with at least one
223/// strict less-than.
224#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
225pub struct VectorClock<const N: usize> {
226    pub clock: [u64; N],
227}
228
229impl<const N: usize> VectorClock<N> {
230    pub const fn zero() -> Self { Self { clock: [0; N] } }
231
232    pub fn increment(&mut self, node_idx: usize) {
233        self.clock[node_idx] += 1;
234    }
235
236    /// Causal-order check: returns `Less` when self happens-before
237    /// other (every component <= and at least one <), `Greater` when
238    /// other happens-before self, `Equal` when identical, `None`
239    /// when concurrent (incomparable).
240    pub fn causal_cmp(&self, other: &Self) -> Option<Ordering> {
241        let mut all_le = true;
242        let mut all_ge = true;
243        let mut strict = false;
244        for i in 0..N {
245            match self.clock[i].cmp(&other.clock[i]) {
246                Ordering::Less    => { all_ge = false; strict = true; }
247                Ordering::Greater => { all_le = false; strict = true; }
248                Ordering::Equal => {}
249            }
250        }
251        match (all_le, all_ge, strict) {
252            (true, true, false) => Some(Ordering::Equal),
253            (true, false, true) => Some(Ordering::Less),
254            (false, true, true) => Some(Ordering::Greater),
255            _ => None,
256        }
257    }
258
259    pub fn merge(&self, other: &Self) -> Self {
260        let mut out = Self::zero();
261        for i in 0..N {
262            out.clock[i] = self.clock[i].max(other.clock[i]);
263        }
264        out
265    }
266}
267
268/// Pointer + vector clock. Useful for distributed CRDT-style
269/// snapshot reads where causal-but-concurrent updates must be
270/// surfaced rather than ordered.
271#[derive(Debug, Clone)]
272pub struct VectorClockPointer<T, const N: usize> {
273    clock: VectorClock<N>,
274    target: Arc<T>,
275}
276
277impl<T, const N: usize> VectorClockPointer<T, N> {
278    pub const fn new(target: Arc<T>, clock: VectorClock<N>) -> Self {
279        Self { clock, target }
280    }
281
282    pub fn clock(&self) -> VectorClock<N> { self.clock }
283    pub fn target(&self) -> &Arc<T> { &self.target }
284
285    /// Returns the target if it causally precedes or equals
286    /// `snapshot`. Returns `None` for concurrent / future events.
287    pub fn read_at(&self, snapshot: VectorClock<N>) -> Option<&T> {
288        match self.clock.causal_cmp(&snapshot) {
289            Some(Ordering::Less | Ordering::Equal) => Some(&self.target),
290            _ => None,
291        }
292    }
293}
294
295// =========================================================
296// VersionedChain<T> - MVCC linked list of historical versions
297// =========================================================
298
299/// Linked list of `(version, value)` nodes ordered newest-first.
300/// Time-travel reads walk the chain until they find a version <=
301/// the query snapshot.
302pub struct VersionedChain<T: Clone> {
303    head: parking_lot::RwLock<Option<Arc<VersionNode<T>>>>,
304}
305
306struct VersionNode<T> {
307    version: u64,
308    value: T,
309    older: Option<Arc<VersionNode<T>>>,
310}
311
312impl<T: Clone> VersionedChain<T> {
313    pub fn new() -> Self {
314        Self { head: parking_lot::RwLock::new(None) }
315    }
316
317    /// Add a new version to the head. `new_version` must strictly
318    /// exceed the current head's version.
319    pub fn push(&self, value: T, new_version: u64) {
320        let mut h = self.head.write();
321        if let Some(cur) = h.as_ref() {
322            assert!(
323                new_version > cur.version,
324                "MVCC chain version must be strictly monotonic: {} -> {}",
325                cur.version, new_version
326            );
327        }
328        let older = h.take();
329        *h = Some(Arc::new(VersionNode { version: new_version, value, older }));
330    }
331
332    /// Read the value visible at `snapshot_version`. Walks back
333    /// through history until a node with version <= snapshot is found.
334    pub fn read_at(&self, snapshot_version: u64) -> Option<T> {
335        let h = self.head.read();
336        let mut cur = h.clone();
337        while let Some(node) = cur {
338            if node.version <= snapshot_version {
339                return Some(node.value.clone());
340            }
341            cur = node.older.clone();
342        }
343        None
344    }
345
346    /// Current (latest) version and value.
347    pub fn current(&self) -> Option<(u64, T)> {
348        self.head.read().as_ref().map(|n| (n.version, n.value.clone()))
349    }
350
351    /// Chain length (number of retained versions).
352    pub fn len(&self) -> usize {
353        let h = self.head.read();
354        let mut cur = h.clone();
355        let mut n = 0;
356        while let Some(node) = cur {
357            n += 1;
358            cur = node.older.clone();
359        }
360        n
361    }
362
363    pub fn is_empty(&self) -> bool { self.head.read().is_none() }
364}
365
366impl<T: Clone> Default for VersionedChain<T> {
367    fn default() -> Self { Self::new() }
368}
369
370#[cfg(test)]
371mod tests {
372    use super::*;
373
374    #[test]
375    fn versioned_pointer_visibility() {
376        let p = VersionedPointer::new(Arc::new("hello".to_string()), 100);
377        assert!(p.visible_at(100));
378        assert!(p.visible_at(101));
379        assert!(!p.visible_at(99));
380        assert_eq!(p.read_at(99), None);
381        assert_eq!(p.read_at(150).map(|s| s.as_str()), Some("hello"));
382    }
383
384    #[test]
385    fn versioned_pointer_replace_enforces_monotonic() {
386        let mut p = VersionedPointer::new(Arc::new(1u64), 10);
387        let old = p.replace(Arc::new(2), 11);
388        assert_eq!(old, 10);
389        assert_eq!(p.version(), 11);
390        assert_eq!(**p.target(), 2);
391    }
392
393    #[test]
394    #[should_panic(expected = "monotonic")]
395    fn versioned_pointer_replace_rejects_non_monotonic() {
396        let mut p = VersionedPointer::new(Arc::new(1u64), 10);
397        let _val = p.replace(Arc::new(2), 5);
398    }
399
400    #[test]
401    fn hlc_advances_logical_within_physical() {
402        let h0 = HybridLogicalClock::new(1000, 0);
403        let h1 = h0.advance(1000);
404        assert_eq!(h1.physical, 1000);
405        assert_eq!(h1.logical, 1);
406        let h2 = h1.advance(1001);
407        assert_eq!(h2.physical, 1001);
408        assert_eq!(h2.logical, 0);
409    }
410
411    #[test]
412    fn hlc_lexicographic_ordering() {
413        let a = HybridLogicalClock::new(100, 5);
414        let b = HybridLogicalClock::new(100, 7);
415        let c = HybridLogicalClock::new(101, 0);
416        assert!(a < b);
417        assert!(b < c);
418        assert!(a < c);
419        assert!(c > a);
420    }
421
422    #[test]
423    fn hlc_merge_takes_max_then_bumps_logical() {
424        let local = HybridLogicalClock::new(100, 5);
425        let received = HybridLogicalClock::new(100, 8);
426        let merged = local.merge(&received, 100);
427        assert_eq!(merged.physical, 100);
428        assert_eq!(merged.logical, 9);
429
430        let received2 = HybridLogicalClock::new(200, 0);
431        let merged2 = local.merge(&received2, 100);
432        assert_eq!(merged2.physical, 200);
433        assert_eq!(merged2.logical, 1);
434    }
435
436    #[test]
437    fn hlc_pointer_visibility() {
438        let snapshot = HybridLogicalClock::new(1000, 5);
439        let p1 = HlcVersionedPointer::new(Arc::new(1u64), HybridLogicalClock::new(999, 99));
440        let p2 = HlcVersionedPointer::new(Arc::new(2u64), HybridLogicalClock::new(1000, 5));
441        let p3 = HlcVersionedPointer::new(Arc::new(3u64), HybridLogicalClock::new(1000, 6));
442        assert_eq!(p1.read_at(snapshot).copied(), Some(1));
443        assert_eq!(p2.read_at(snapshot).copied(), Some(2));
444        assert_eq!(p3.read_at(snapshot), None);
445    }
446
447    #[test]
448    fn vector_clock_causal_ordering() {
449        // 3-node system.
450        let a = VectorClock::<3> { clock: [1, 0, 0] };
451        let b = VectorClock::<3> { clock: [1, 1, 0] };
452        let c = VectorClock::<3> { clock: [0, 0, 1] };
453        // a happens-before b (b extends a).
454        assert_eq!(a.causal_cmp(&b), Some(Ordering::Less));
455        assert_eq!(b.causal_cmp(&a), Some(Ordering::Greater));
456        // a equal to itself.
457        assert_eq!(a.causal_cmp(&a), Some(Ordering::Equal));
458        // a and c are concurrent.
459        assert_eq!(a.causal_cmp(&c), None);
460        assert_eq!(c.causal_cmp(&a), None);
461    }
462
463    #[test]
464    fn vector_clock_merge_takes_pointwise_max() {
465        let a = VectorClock::<3> { clock: [1, 0, 0] };
466        let b = VectorClock::<3> { clock: [0, 0, 1] };
467        let m = a.merge(&b);
468        assert_eq!(m.clock, [1, 0, 1]);
469    }
470
471    #[test]
472    fn vector_clock_pointer_concurrent_read_is_none() {
473        let snapshot = VectorClock::<3> { clock: [1, 1, 1] };
474        let visible = VectorClockPointer::new(
475            Arc::new(1u64),
476            VectorClock::<3> { clock: [1, 0, 0] },
477        );
478        let future = VectorClockPointer::new(
479            Arc::new(2u64),
480            VectorClock::<3> { clock: [2, 0, 0] },
481        );
482        assert!(visible.read_at(snapshot).is_some());
483        assert!(future.read_at(snapshot).is_none());
484    }
485
486    #[test]
487    fn versioned_chain_basic_push_and_read() {
488        let chain = VersionedChain::<u64>::new();
489        chain.push(10, 1);
490        chain.push(20, 2);
491        chain.push(30, 3);
492        assert_eq!(chain.len(), 3);
493        // Time-travel reads.
494        assert_eq!(chain.read_at(0), None);
495        assert_eq!(chain.read_at(1), Some(10));
496        assert_eq!(chain.read_at(2), Some(20));
497        assert_eq!(chain.read_at(3), Some(30));
498        assert_eq!(chain.read_at(99), Some(30));
499        assert_eq!(chain.current(), Some((3, 30)));
500    }
501
502    #[test]
503    #[should_panic(expected = "monotonic")]
504    fn versioned_chain_push_rejects_non_monotonic() {
505        let chain = VersionedChain::<u64>::new();
506        chain.push(10, 5);
507        chain.push(20, 3);  // earlier; must panic
508    }
509
510    #[test]
511    fn versioned_cascade_outer_inner_pattern() {
512        // K_cascade=2 demonstration: VersionedPointer<VersionedPointer<T>>.
513        // Outer carries a coarse version, inner carries a finer version
514        // (same shape as HLC but using monotonic instead of (phys, logic)).
515        let inner = VersionedPointer::new(Arc::new(42u64), 100);
516        let outer = VersionedPointer::new(Arc::new(inner.clone()), 1);
517
518        // Outer-visible at coarse snapshot 5.
519        assert!(outer.visible_at(5));
520        let inner_ref = outer.read_at(5).unwrap();
521        // Inner check is at the finer scale.
522        assert!(inner_ref.visible_at(101));
523        assert!(!inner_ref.visible_at(99));
524    }
525}
subetha_pointers/versioned_pointer.rs

subetha_pointers/
versioned_pointer.rs
