grit-core 0.2.4

//! Hybrid logical clock (Design Invariant 5): every op carries an HLC so merge
//! order across devices never depends on device wall clocks alone.
//!
//! Encoding is a fixed-width, lexicographically sortable string:
//! `{wall_ms:016x}-{counter:08x}-{device_id}`. String comparison == HLC order,
//! so SQLite can order and compare HLCs without custom collation.

use std::sync::Mutex;

use serde::{Deserialize, Serialize};

use crate::clock::{Clock, TimestampMs};
use crate::error::Error;

/// A single hybrid-logical-clock reading.
///
/// # Example
/// ```
/// use grit_core::Hlc;
/// let a = Hlc::new(1000, 0, "dev-a");
/// let b = Hlc::new(1000, 1, "dev-a");
/// assert!(a < b);
/// assert_eq!(a, a.encode().parse().unwrap());
/// ```
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct Hlc {
    /// Wall-clock component, ms since epoch.
    pub wall_ms: TimestampMs,
    /// Logical counter, bumps when wall time stalls or runs backwards.
    pub counter: u32,
    /// Tie-breaker: the originating device.
    pub device_id: String,
}

impl Hlc {
    /// Construct an HLC from parts.
    pub fn new(wall_ms: TimestampMs, counter: u32, device_id: impl Into<String>) -> Self {
        Self {
            wall_ms,
            counter,
            device_id: device_id.into(),
        }
    }

    /// Encode to the sortable string form stored in SQLite.
    pub fn encode(&self) -> String {
        format!(
            "{:016x}-{:08x}-{}",
            self.wall_ms, self.counter, self.device_id
        )
    }
}

impl std::fmt::Display for Hlc {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.encode())
    }
}

impl std::str::FromStr for Hlc {
    type Err = Error;

    fn from_str(s: &str) -> Result<Self, Error> {
        let bad = || Error::InvalidHlc(s.to_owned());
        let (wall, rest) = s.split_at_checked(16).ok_or_else(bad)?;
        let rest = rest.strip_prefix('-').ok_or_else(bad)?;
        let (counter, rest) = rest.split_at_checked(8).ok_or_else(bad)?;
        let device_id = rest.strip_prefix('-').ok_or_else(bad)?;
        Ok(Self {
            wall_ms: TimestampMs::from_str_radix(wall, 16).map_err(|_| bad())?,
            counter: u32::from_str_radix(counter, 16).map_err(|_| bad())?,
            device_id: device_id.to_owned(),
        })
    }
}

/// Issues monotonically increasing HLCs for one device, folding in the highest
/// HLC observed from remote ops so causality survives clock skew.
pub struct HlcGenerator {
    device_id: String,
    last: Mutex<(TimestampMs, u32)>,
}

impl HlcGenerator {
    /// A generator for `device_id` starting at wall time zero.
    pub fn new(device_id: impl Into<String>) -> Self {
        Self {
            device_id: device_id.into(),
            last: Mutex::new((0, 0)),
        }
    }

    /// Issue the next HLC, strictly greater than every previous one issued or
    /// observed, using `clock` for the wall component.
    ///
    /// Negative clock readings are clamped to 0 (the sortable encoding is
    /// only order-preserving for non-negative wall times — see [`Clock`]);
    /// if the logical counter is exhausted under a stalled clock, the wall
    /// component advances logically instead (HLC semantics permit the wall
    /// to run ahead of the physical clock).
    pub fn next(&self, clock: &dyn Clock) -> Hlc {
        let now = clock.now_ms().max(0);
        let mut last = self
            .last
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if now > last.0 {
            *last = (now, 0);
        } else if last.1 == u32::MAX {
            *last = (last.0 + 1, 0);
        } else {
            last.1 += 1;
        }
        Hlc::new(last.0, last.1, self.device_id.clone())
    }

    /// Fold in an HLC observed from a remote op so subsequently issued HLCs
    /// sort after it. Negative remote wall times are ignored (they are
    /// rejected at ingest by [`crate::Grit::apply_remote`]).
    pub fn observe(&self, remote: &Hlc) {
        if remote.wall_ms < 0 {
            return;
        }
        let mut last = self
            .last
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if (remote.wall_ms, remote.counter) > *last {
            *last = (remote.wall_ms, remote.counter);
        }
    }

    /// Test seam: force the internal (wall, counter) state, e.g. to exercise
    /// counter exhaustion without 2^32 calls.
    #[cfg(test)]
    fn set_state(&self, wall_ms: TimestampMs, counter: u32) {
        *self.last.lock().unwrap() = (wall_ms, counter);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::clock::ManualClock;

    #[test]
    fn encode_order_matches_semantic_order() {
        let hlcs = [
            Hlc::new(1, 0, "a"),
            Hlc::new(1, 1, "a"),
            Hlc::new(1, 1, "b"),
            Hlc::new(2, 0, "a"),
            Hlc::new(0x1_0000_0000, 0, "a"),
        ];
        for pair in hlcs.windows(2) {
            assert!(pair[0] < pair[1]);
            assert!(
                pair[0].encode() < pair[1].encode(),
                "{} !< {}",
                pair[0],
                pair[1]
            );
        }
    }

    #[test]
    fn roundtrip() {
        let h = Hlc::new(123_456_789, 42, "device-x");
        let parsed: Hlc = h.encode().parse().unwrap();
        assert_eq!(h, parsed);
    }

    #[test]
    fn generator_is_monotonic_under_stalled_clock() {
        let clock = ManualClock::new(100);
        let generator = HlcGenerator::new("dev");
        let a = generator.next(&clock);
        let b = generator.next(&clock);
        clock.set_ms(50); // clock runs backwards
        let c = generator.next(&clock);
        assert!(a < b && b < c);
    }

    #[test]
    fn negative_clock_is_clamped() {
        let clock = ManualClock::new(-5_000);
        let generator = HlcGenerator::new("dev");
        let a = generator.next(&clock);
        assert!(
            a.wall_ms >= 0,
            "negative wall would break the sortable encoding"
        );
        let b = generator.next(&clock);
        assert!(a < b);
        assert!(a.encode() < b.encode());
        // And negative remote HLCs never drag the generator backwards or
        // poison future encodings.
        generator.observe(&Hlc::new(-9_000, 7, "evil"));
        assert!(generator.next(&clock).wall_ms >= 0);
    }

    #[test]
    fn counter_exhaustion_advances_logical_wall() {
        let clock = ManualClock::new(100);
        let generator = HlcGenerator::new("dev");
        let before = generator.next(&clock);
        generator.set_state(100, u32::MAX);
        let after = generator.next(&clock);
        assert_eq!((after.wall_ms, after.counter), (101, 0));
        assert!(before < after);
        assert!(before.encode() < after.encode());
    }

    #[test]
    fn generator_respects_observed_remote() {
        let clock = ManualClock::new(100);
        let generator = HlcGenerator::new("dev");
        generator.observe(&Hlc::new(9_999, 3, "other"));
        assert!(generator.next(&clock) > Hlc::new(9_999, 3, "other"));
    }
}