minerva 0.2.0

Causal ordering for distributed systems
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

extern crate alloc;

use alloc::boxed::Box;
use alloc::sync::Arc;
use core::sync::atomic::{AtomicU64, Ordering};

/// Source of physical time for the HLC.
pub trait TimeSource: Send + Sync {
    /// Returns the current physical time as a `u64`, typically in nanoseconds.
    fn now(&self) -> u64;
}

impl<T: TimeSource + ?Sized> TimeSource for Box<T> {
    fn now(&self) -> u64 {
        (**self).now()
    }
}

/// Deterministic counter source: each read returns then increments.
#[derive(Debug, Default)]
pub struct TickCounter {
    counter: AtomicU64,
}

impl TickCounter {
    /// Creates a counter starting at zero.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            counter: AtomicU64::new(0),
        }
    }
}

impl TimeSource for TickCounter {
    fn now(&self) -> u64 {
        self.counter.fetch_add(1, Ordering::SeqCst)
    }
}

/// A caller-advanced virtual [`TimeSource`] for tests and deterministic
/// simulation.
///
/// Reading does not move it: [`now`](TimeSource::now) returns the same value until
/// a driver advances it with [`set`](Self::set) or [`advance`](Self::advance). That
/// is the difference from [`TickCounter`], which ticks on every read.
///
/// The reading lives behind an [`Arc`], so the type is a cheap shared handle: a
/// caller keeps one clone to drive time after handing another to a
/// [`Clock`](crate::kairos::Clock), which takes ownership of its source. Cloning
/// shares the same reading; advancing one handle moves them all. This is `no_std`
/// (an [`Arc`] plus an atomic), so it serves the `--no-default-features` tests as
/// well as `std` ones.
///
/// [`set`](Self::set) moves time in either direction, so a test can replay the
/// non-monotonic readings that exercise the clock's skew paths; the clock's own
/// output stays monotonic regardless. [`advance`](Self::advance) only moves forward,
/// saturating at the ceiling.
#[derive(Debug, Clone, Default)]
pub struct VirtualTimeSource {
    current: Arc<AtomicU64>,
}

impl VirtualTimeSource {
    /// Creates a virtual source reading `physical` until a driver advances it.
    #[must_use]
    pub fn new(physical: u64) -> Self {
        Self {
            current: Arc::new(AtomicU64::new(physical)),
        }
    }

    /// Sets the current reading to `physical`, in either direction.
    ///
    /// A backward set is deliberate: it feeds the clock the non-monotonic readings
    /// that drive its skew handling, which a forward-only source cannot.
    pub fn set(&self, physical: u64) {
        self.current.store(physical, Ordering::SeqCst);
    }

    /// Advances the reading forward by `delta`, saturating at [`u64::MAX`] rather
    /// than wrapping (the crate's "saturate, never wrap" stance). A command, not a
    /// query: read the result back with [`now`](TimeSource::now).
    pub fn advance(&self, delta: u64) {
        let mut current = self.current.load(Ordering::SeqCst);
        loop {
            let next = current.saturating_add(delta);
            match self.current.compare_exchange_weak(
                current,
                next,
                Ordering::SeqCst,
                Ordering::SeqCst,
            ) {
                Ok(_) => return,
                Err(observed) => current = observed,
            }
        }
    }
}

impl TimeSource for VirtualTimeSource {
    fn now(&self) -> u64 {
        self.current.load(Ordering::SeqCst)
    }
}