yeti_types/

hlc.rs

//! Hybrid Logical Clock (HLC) — replication timestamp primitive.
//!
//! HLC is the foundational ordering primitive for yeti's multi-master
//! CRDT data plane. Every log entry is stamped with an HLC and applied
//! last-to-arrive. The richer mode (planned) stores per-record HLCs in
//! an `__hlc_meta` column family and switches the receive path to
//! HLC-LWW for plain fields plus CRDT merge for `@crdt`-marked fields.
//!
//! ## Why HLC instead of physical or logical alone
//!
//! - **Physical timestamps** (wall clock) drift between nodes and can
//!   regress if the clock is adjusted. Two writes with the same wall-
//!   clock millisecond can't be ordered.
//! - **Logical clocks** (Lamport) preserve causal order but lose any
//!   relationship to real time, so you can't say "the most recent
//!   write" in a way humans understand.
//! - **HLC** (Kulkarni et al. 2014) combines both: a physical
//!   component bounded by wall-clock that advances monotonically, plus
//!   a logical counter that breaks ties when physical time stalls or
//!   regresses.
//!
//! ## Wire shape
//!
//! An `HlcTimestamp` is a `u64` packed:
//! - High 48 bits: physical time, milliseconds since UNIX epoch
//!   (sufficient for ~8900 years past 1970)
//! - Low 16 bits: logical counter (65,535 events at the same
//!   physical millisecond before overflow — at which point physical
//!   time will have advanced)
//!
//! Total ordering on `HlcTimestamp` is the natural u64 ordering.
//! Two timestamps from different nodes that happen to collide are
//! broken by appending `node_id` at the call site (the HLC itself is
//! node-agnostic; the node-id tie-break is part of the replication
//! envelope, not the timestamp).
//!
//! ## Skew tolerance
//!
//! HLC-LWW assumes bounded clock skew between nodes. A peer whose
//! wall clock is far ahead of ours could write timestamps that look
//! arbitrarily-in-the-future from our perspective; if we accept them
//! into our HLC state, subsequent local writes will be perpetually
//! "behind" until our wall clock catches up. The skew tolerance
//! caps how far in the future we accept incoming timestamps.
//!
//! Default 250ms (CockroachDB precedent). Configurable per-cluster
//! via `replication.hlc_skew_tolerance_ms` in `yeti-config.yaml`.

use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH};

use serde::{Deserialize, Serialize};

/// Default HLC skew tolerance — the maximum amount we'll let a peer's
/// timestamp exceed our local wall clock before rejecting it.
///
/// Configurable per-cluster via `yeti-config.yaml`
/// (`replication.hlc_skew_tolerance_ms`). 250ms is the `CockroachDB`
/// precedent — tight enough to limit poisoning damage from a badly-
/// skewed peer, loose enough to absorb normal NTP-disciplined drift.
pub const DEFAULT_SKEW_TOLERANCE_MS: u64 = 250;

const PHYSICAL_BITS: u32 = 48;
const LOGICAL_BITS: u32 = 16;
const LOGICAL_MASK: u64 = (1 << LOGICAL_BITS) - 1;
const PHYSICAL_MASK: u64 = !LOGICAL_MASK;

/// A 64-bit hybrid logical clock timestamp.
///
/// Layout: `[physical_ms : 48][logical : 16]`. Total ordering is the
/// natural `u64` ordering on the packed value. Tie-breaking across
/// nodes (when two HLC values are equal) is the caller's concern —
/// the replication envelope appends `node_id` as a secondary key.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct HlcTimestamp(u64);

impl HlcTimestamp {
    /// Zero / sentinel value. Less than any real timestamp; useful
    /// for "no last-applied yet" cases.
    pub const ZERO: Self = Self(0);

    /// The maximum representable timestamp.
    pub const MAX: Self = Self(u64::MAX);

    /// Compose from physical milliseconds and logical counter.
    ///
    /// Panics if `physical_ms` overflows 48 bits (year ~10880) or
    /// `logical` overflows 16 bits.
    #[must_use]
    pub const fn new(physical_ms: u64, logical: u16) -> Self {
        assert!(
            physical_ms < (1 << PHYSICAL_BITS),
            "HLC physical component exceeds 48 bits"
        );
        Self((physical_ms << LOGICAL_BITS) | (logical as u64))
    }

    /// Physical milliseconds since UNIX epoch.
    #[must_use]
    pub const fn physical_ms(&self) -> u64 {
        (self.0 & PHYSICAL_MASK) >> LOGICAL_BITS
    }

    /// Logical counter component.
    #[must_use]
    pub const fn logical(&self) -> u16 {
        (self.0 & LOGICAL_MASK) as u16
    }

    /// Underlying packed `u64`. Useful for storage / wire formats
    /// that want a single integer.
    #[must_use]
    pub const fn as_u64(&self) -> u64 {
        self.0
    }

    /// Reconstruct from the packed `u64`.
    #[must_use]
    pub const fn from_u64(packed: u64) -> Self {
        Self(packed)
    }
}

impl std::fmt::Display for HlcTimestamp {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}.{}", self.physical_ms(), self.logical())
    }
}

/// Error from [`HybridLogicalClock::update`] when an incoming
/// timestamp exceeds the configured skew tolerance.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[error(
    "HLC skew tolerance exceeded: remote {remote_physical_ms}ms > local {local_physical_ms}ms + {tolerance_ms}ms tolerance"
)]
pub struct SkewExceeded {
    /// The remote timestamp's physical component (ms since UNIX epoch).
    pub remote_physical_ms: u64,
    /// The local wall clock's reading at the time of the rejection.
    pub local_physical_ms: u64,
    /// The skew-tolerance bound that was exceeded.
    pub tolerance_ms: u64,
}

/// A node-local hybrid logical clock.
///
/// One instance per node. Thread-safe (internal `Mutex` on the
/// last-emitted timestamp). All HLC reads / updates go through this
/// type so the invariant "every emitted timestamp is strictly greater
/// than every previously-emitted timestamp on this node" is
/// preserved.
///
/// ## Usage
///
/// - [`now`](Self::now) — call before writing a record locally. The
///   returned timestamp is guaranteed strictly greater than any
///   previously-emitted timestamp by this clock.
/// - [`update`](Self::update) — call when receiving a record from a
///   peer. The returned timestamp is greater than both the local
///   last-emitted and the incoming remote timestamp. Errors with
///   [`SkewExceeded`] if the remote's physical component is more
///   than `skew_tolerance_ms` ahead of our wall clock.
/// - [`last_emitted`](Self::last_emitted) — current high-water mark;
///   used by replication for catch-up bookmarks.
pub struct HybridLogicalClock {
    /// The last timestamp this clock emitted. Strictly monotonic.
    last: Mutex<HlcTimestamp>,
    /// Wall-clock source. Injected for testability; production uses
    /// [`SystemTime::now`].
    wall_clock: Box<dyn Fn() -> u64 + Send + Sync>,
    /// Maximum allowed `(remote_physical - local_physical)` before we
    /// reject the incoming timestamp.
    skew_tolerance_ms: u64,
}

impl std::fmt::Debug for HybridLogicalClock {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let last = self.last.lock().map_or(HlcTimestamp::ZERO, |g| *g);
        f.debug_struct("HybridLogicalClock")
            .field("last_emitted", &last)
            .field("skew_tolerance_ms", &self.skew_tolerance_ms)
            .field("wall_clock", &"<dyn Fn() -> u64>")
            .finish()
    }
}

impl HybridLogicalClock {
    /// Construct with default wall clock and default skew tolerance.
    #[must_use]
    pub fn new() -> Self {
        Self::with_skew_tolerance_ms(DEFAULT_SKEW_TOLERANCE_MS)
    }

    /// Construct with a custom skew-tolerance bound.
    #[must_use]
    pub fn with_skew_tolerance_ms(skew_tolerance_ms: u64) -> Self {
        Self {
            last: Mutex::new(HlcTimestamp::ZERO),
            wall_clock: Box::new(default_wall_clock_ms),
            skew_tolerance_ms,
        }
    }

    /// Construct with an injected wall clock (for testing). The clock
    /// must return milliseconds since some fixed epoch.
    #[must_use]
    pub fn with_wall_clock<F>(wall_clock: F, skew_tolerance_ms: u64) -> Self
    where
        F: Fn() -> u64 + Send + Sync + 'static,
    {
        Self {
            last: Mutex::new(HlcTimestamp::ZERO),
            wall_clock: Box::new(wall_clock),
            skew_tolerance_ms,
        }
    }

    /// Emit a fresh timestamp. Strictly greater than every previously-
    /// emitted timestamp by this clock.
    ///
    /// Algorithm:
    /// 1. Read wall clock `wall`.
    /// 2. If `wall > last.physical`, return `(wall, 0)` and update.
    /// 3. Else, return `(last.physical, last.logical + 1)` and update
    ///    (logical-counter advance because wall clock didn't move
    ///    past the last emit).
    ///
    /// Logical overflow (65,536 events at the same wall-clock
    /// millisecond) panics — in practice this requires emitting
    /// 65,536 timestamps within ~1ms, which would mean a writer is
    /// generating > 65M events/sec on a single clock; a real
    /// scenario at that rate needs a re-architecture, not a wider
    /// counter.
    pub fn now(&self) -> HlcTimestamp {
        let wall = (self.wall_clock)();
        // Recover the inner value on poison: HLC monotonicity is preserved
        // even if a panicking writer held the lock — the last-emitted
        // timestamp is still the high-water mark and remains valid.
        let mut last = self
            .last
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let new = if wall > last.physical_ms() {
            HlcTimestamp::new(wall, 0)
        } else {
            // Wall didn't advance — bump logical. Overflow at >65k events
            // in one millisecond is a documented unrecoverable state
            // (single writer generating >65M events/sec); the panic
            // message is the intended failure mode.
            #[allow(clippy::expect_used)]
            let next_logical = last
                .logical()
                .checked_add(1)
                .expect("HLC logical counter overflow within a single ms");
            HlcTimestamp::new(last.physical_ms(), next_logical)
        };
        *last = new;
        new
    }

    /// Integrate a remote timestamp and emit a fresh local timestamp
    /// strictly greater than both the local state and the remote.
    ///
    /// Algorithm:
    /// 1. Read wall clock `wall`.
    /// 2. Reject if `remote.physical > wall + skew_tolerance`. This
    ///    prevents a misbehaving peer from poisoning our HLC state
    ///    with a far-future timestamp.
    /// 3. Compute `new_physical = max(local.physical, remote.physical, wall)`.
    /// 4. Compute `new_logical`:
    ///    - both `local.physical` and `remote.physical` == `new_physical`:
    ///      `max(local.logical, remote.logical) + 1`
    ///    - only `local.physical` == `new_physical`: `local.logical + 1`
    ///    - only `remote.physical` == `new_physical`: `remote.logical + 1`
    ///    - neither (wall is new winner): `0`
    /// 5. Update local and return.
    ///
    /// # Errors
    ///
    /// Returns [`SkewExceeded`] when the remote's physical component
    /// is more than `skew_tolerance_ms` ahead of our wall clock.
    pub fn update(&self, remote: HlcTimestamp) -> Result<HlcTimestamp, SkewExceeded> {
        let wall = (self.wall_clock)();
        if remote.physical_ms() > wall.saturating_add(self.skew_tolerance_ms) {
            return Err(SkewExceeded {
                remote_physical_ms: remote.physical_ms(),
                local_physical_ms: wall,
                tolerance_ms: self.skew_tolerance_ms,
            });
        }

        let mut last = self
            .last
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let local_phys = last.physical_ms();
        let remote_phys = remote.physical_ms();
        let new_phys = local_phys.max(remote_phys).max(wall);

        // Logical overflow paths are documented unrecoverable invariants
        // (see `now()` for rationale).
        #[allow(clippy::expect_used)]
        let new_logical = if new_phys == local_phys && new_phys == remote_phys {
            last.logical()
                .max(remote.logical())
                .checked_add(1)
                .expect("HLC logical overflow on update (local + remote same ms)")
        } else if new_phys == local_phys {
            last.logical()
                .checked_add(1)
                .expect("HLC logical overflow on update (local advanced)")
        } else if new_phys == remote_phys {
            remote
                .logical()
                .checked_add(1)
                .expect("HLC logical overflow on update (remote advanced)")
        } else {
            0
        };

        let new = HlcTimestamp::new(new_phys, new_logical);
        *last = new;
        drop(last);
        Ok(new)
    }

    /// Read the current high-water mark without emitting a new
    /// timestamp. Used by replication to record per-peer `last_applied`
    /// bookmarks.
    #[must_use]
    pub fn last_emitted(&self) -> HlcTimestamp {
        *self
            .last
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// The skew tolerance this clock was configured with.
    #[must_use]
    pub const fn skew_tolerance_ms(&self) -> u64 {
        self.skew_tolerance_ms
    }
}

impl Default for HybridLogicalClock {
    fn default() -> Self {
        Self::new()
    }
}

fn default_wall_clock_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        // Pre-1970 timestamps shouldn't happen; if the clock is set to
        // before the epoch, fall back to 0 so we still emit *something*
        // sensible (the logical counter will dominate).
        //
        // u128 → u64 saturates at u64::MAX (~584 million years post-epoch);
        // realistic wall-clock millis fit in u64 forever.
        .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU64, Ordering};

    /// Build an HLC with a controllable wall clock.
    fn new_test_clock(initial_wall_ms: u64) -> (HybridLogicalClock, Arc<AtomicU64>) {
        let wall = Arc::new(AtomicU64::new(initial_wall_ms));
        let wall_for_clock = Arc::clone(&wall);
        let clock = HybridLogicalClock::with_wall_clock(
            move || wall_for_clock.load(Ordering::SeqCst),
            DEFAULT_SKEW_TOLERANCE_MS,
        );
        (clock, wall)
    }

    #[test]
    fn timestamp_packing_round_trip() {
        let ts = HlcTimestamp::new(1_700_000_000_000, 42);
        assert_eq!(ts.physical_ms(), 1_700_000_000_000);
        assert_eq!(ts.logical(), 42);
        assert_eq!(HlcTimestamp::from_u64(ts.as_u64()), ts);
    }

    #[test]
    fn natural_ordering_is_physical_then_logical() {
        let a = HlcTimestamp::new(100, 5);
        let b = HlcTimestamp::new(100, 6);
        let c = HlcTimestamp::new(101, 0);
        assert!(a < b);
        assert!(b < c);
        assert!(a < c);
    }

    #[test]
    fn now_advances_monotonically_when_wall_stalls() {
        let (clock, _wall) = new_test_clock(1000);
        let t1 = clock.now();
        let t2 = clock.now();
        let t3 = clock.now();
        assert_eq!(t1.physical_ms(), 1000);
        assert_eq!(t1.logical(), 0);
        assert_eq!(t2.physical_ms(), 1000);
        assert_eq!(t2.logical(), 1);
        assert_eq!(t3.physical_ms(), 1000);
        assert_eq!(t3.logical(), 2);
        assert!(t1 < t2);
        assert!(t2 < t3);
    }

    #[test]
    fn now_takes_wall_when_wall_advances() {
        let (clock, wall) = new_test_clock(1000);
        let t1 = clock.now();
        wall.store(2000, Ordering::SeqCst);
        let t2 = clock.now();
        assert_eq!(t2.physical_ms(), 2000);
        assert_eq!(t2.logical(), 0);
        assert!(t1 < t2);
    }

    #[test]
    fn now_holds_at_last_when_wall_regresses() {
        // Operator adjusts the wall clock backwards (NTP step). The HLC
        // must NOT regress — the physical component holds at the last-
        // emitted value and the logical counter advances.
        let (clock, wall) = new_test_clock(5000);
        let t1 = clock.now();
        wall.store(3000, Ordering::SeqCst);
        let t2 = clock.now();
        assert_eq!(t1.physical_ms(), 5000);
        assert_eq!(
            t2.physical_ms(),
            5000,
            "HLC must not regress on wall stepback"
        );
        assert_eq!(t2.logical(), 1);
        assert!(t1 < t2);
    }

    #[test]
    fn update_with_strictly_greater_remote() {
        // Wall starts at 1000; we advance it to 2000 so a remote at
        // (2000, 5) is inside the skew tolerance window.
        let (clock, wall) = new_test_clock(1000);
        let _ = clock.now(); // local at (1000, 0)
        wall.store(2000, Ordering::SeqCst);
        let remote = HlcTimestamp::new(2000, 5);
        let merged = clock.update(remote).unwrap();
        assert_eq!(merged.physical_ms(), 2000);
        assert_eq!(
            merged.logical(),
            6,
            "remote.logical + 1 when remote phys wins"
        );
    }

    #[test]
    fn update_with_equal_physical_takes_max_logical_plus_one() {
        let (clock, _wall) = new_test_clock(1000);
        let _ = clock.now();
        let _ = clock.now(); // local at (1000, 1)
        let remote = HlcTimestamp::new(1000, 10);
        let merged = clock.update(remote).unwrap();
        assert_eq!(merged.physical_ms(), 1000);
        assert_eq!(merged.logical(), 11);
    }

    #[test]
    fn update_with_lesser_remote_advances_local_logical() {
        let (clock, _wall) = new_test_clock(5000);
        let _ = clock.now(); // local (5000, 0)
        let remote = HlcTimestamp::new(3000, 0);
        let merged = clock.update(remote).unwrap();
        // local.physical > remote.physical, so new_phys = local.physical;
        // new_logical = local.logical + 1.
        assert_eq!(merged.physical_ms(), 5000);
        assert_eq!(merged.logical(), 1);
    }

    #[test]
    fn update_with_wall_winning_resets_logical() {
        let (clock, wall) = new_test_clock(1000);
        let _ = clock.now(); // local (1000, 0)
        wall.store(10_000, Ordering::SeqCst);
        let remote = HlcTimestamp::new(2000, 99);
        let merged = clock.update(remote).unwrap();
        // wall(10000) > local(1000), wall > remote(2000) → new_phys = wall, logical = 0.
        assert_eq!(merged.physical_ms(), 10_000);
        assert_eq!(merged.logical(), 0);
    }

    #[test]
    fn update_rejects_remote_beyond_skew_tolerance() {
        let (clock, _wall) = new_test_clock(1000);
        let far_future = HlcTimestamp::new(1000 + DEFAULT_SKEW_TOLERANCE_MS + 1, 0);
        let err = clock.update(far_future).unwrap_err();
        assert_eq!(err.remote_physical_ms, 1000 + DEFAULT_SKEW_TOLERANCE_MS + 1);
        assert_eq!(err.local_physical_ms, 1000);
        assert_eq!(err.tolerance_ms, DEFAULT_SKEW_TOLERANCE_MS);
    }

    #[test]
    fn update_accepts_remote_at_skew_tolerance_boundary() {
        let (clock, _wall) = new_test_clock(1000);
        // Exactly at the boundary: remote_phys == wall + tolerance.
        let at_boundary = HlcTimestamp::new(1000 + DEFAULT_SKEW_TOLERANCE_MS, 0);
        let merged = clock.update(at_boundary).unwrap();
        assert_eq!(merged.physical_ms(), 1000 + DEFAULT_SKEW_TOLERANCE_MS);
    }

    #[test]
    fn skew_tolerance_is_configurable() {
        // A tight cluster (PTP-disciplined) can tighten the bound.
        let wall = Arc::new(AtomicU64::new(1000));
        let wall_clone = wall;
        let strict =
            HybridLogicalClock::with_wall_clock(move || wall_clone.load(Ordering::SeqCst), 10);
        let too_far = HlcTimestamp::new(1100, 0);
        assert!(strict.update(too_far).is_err());

        let just_inside = HlcTimestamp::new(1010, 0);
        assert!(strict.update(just_inside).is_ok());
    }

    #[test]
    fn now_and_update_interleave_preserve_monotonicity() {
        // Realistic scenario: local writes interleaved with incoming
        // remote replications; every emitted timestamp must be > every
        // prior emitted timestamp on this clock.
        let (clock, wall) = new_test_clock(1000);
        let t1 = clock.now();
        let t2 = clock.update(HlcTimestamp::new(1100, 0)).unwrap();
        let t3 = clock.now();
        wall.store(1200, Ordering::SeqCst);
        let t4 = clock.now();
        let t5 = clock.update(HlcTimestamp::new(1150, 50)).unwrap();
        let t6 = clock.now();
        let series = [t1, t2, t3, t4, t5, t6];
        for w in series.windows(2) {
            assert!(
                w[0] < w[1],
                "monotonicity violated: {:?} >= {:?}",
                w[0],
                w[1]
            );
        }
    }

    #[test]
    fn last_emitted_reflects_latest_without_advancing() {
        let (clock, _wall) = new_test_clock(1000);
        let t1 = clock.now();
        assert_eq!(clock.last_emitted(), t1);
        let t2 = clock.now();
        assert_eq!(clock.last_emitted(), t2);
        // Reading last_emitted does not advance state.
        assert_eq!(clock.last_emitted(), t2);
    }

    #[test]
    fn display_format() {
        let ts = HlcTimestamp::new(1_700_000_000_000, 42);
        assert_eq!(format!("{ts}"), "1700000000000.42");
    }

    #[test]
    fn zero_is_less_than_anything_real() {
        let ts = HlcTimestamp::new(1, 0);
        assert!(HlcTimestamp::ZERO < ts);
    }
}