minerva 0.2.0

extern crate alloc;

use alloc::boxed::Box;
use core::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use portable_atomic::AtomicU128;

mod config;
mod error;
mod fold;
mod local;
#[cfg(kani)]
mod proofs;
mod run;
mod state;

pub use config::{ClockConfig, ClockStats};
pub use error::{InvalidStationId, SkewExceeded};
pub use local::{DynLocalClock, LocalClock};
pub use run::{KairosRun, KairosRunIter};

use super::backoff::Backoff;
use super::stamp::{Kairos, ToU16};
use super::time::TimeSource;
use state::ClockState;

/// A lock-free Hybrid Logical Clock over an owned physical-time source.
///
/// `TS` is inferred from the source you pass, so a concrete source costs no
/// allocation. The parameter defaults to `Box<dyn TimeSource>` ([`DynClock`])
/// for heterogeneous sources behind one erased type.
pub struct Clock<TS: TimeSource = Box<dyn TimeSource>> {
    time_source: TS,
    /// Spatial identity, fixed at construction. Held here rather than re-derived
    /// from `state` so generation never reads identity from the mutable word.
    station_id: u32,
    /// Packed [`ClockState`]: `Option<(physical, logical)>` of the last mint. One
    /// atomic word, so generation advances with a single compare-and-swap.
    state: AtomicU128,
    max_forward_skew: AtomicU64,
    max_backward_skew: AtomicU64,
    peak_attempts: AtomicU32,
    config: ClockConfig,
}

impl<TS: TimeSource> Clock<TS> {
    /// Creates a clock with an explicit time source and configuration.
    ///
    /// # Errors
    ///
    /// Returns [`InvalidStationId`] if `station_id` is zero.
    pub fn new(
        time_source: TS,
        station_id: u32,
        config: ClockConfig,
    ) -> Result<Self, InvalidStationId> {
        if station_id == 0 {
            return Err(InvalidStationId);
        }
        Ok(Self {
            time_source,
            station_id,
            state: AtomicU128::new(ClockState::UNINIT.0),
            max_forward_skew: AtomicU64::new(0),
            max_backward_skew: AtomicU64::new(0),
            peak_attempts: AtomicU32::new(0),
            config,
        })
    }

    /// Creates a clock with [`ClockConfig::default`].
    ///
    /// # Errors
    ///
    /// Returns [`InvalidStationId`] if `station_id` is zero.
    pub fn with_default_config(time_source: TS, station_id: u32) -> Result<Self, InvalidStationId> {
        Self::new(time_source, station_id, ClockConfig::default())
    }

    /// Returns this clock's station identity, fixed at construction.
    ///
    /// Every stamp the clock mints carries this value in its `station_id` field; a
    /// holder can read its identity without minting a throwaway stamp. This is the
    /// identity a [`metis::Producer`](crate::metis::Producer) single-sources from its
    /// clock.
    #[must_use]
    pub const fn station_id(&self) -> u32 {
        self.station_id
    }

    /// Mints a stamp at the current reading: the local *send* step.
    ///
    /// Infallible. Contention is resolved by retrying and is observable as
    /// [`ClockStats::peak_attempts`] through [`stats`](Self::stats).
    pub fn now<T: ToU16>(&self, kairotic: T) -> Kairos {
        let (physical, logical) = self.advance(None);
        Kairos::new(physical, logical, self.station_id, kairotic)
    }

    /// Mints a stamp strictly after `previous`: receive-and-send fused into
    /// one step. Equivalent to [`observe`](Self::observe) then
    /// [`now`](Self::now), minus the extra logical tick and the second
    /// reading.
    pub fn after<T: ToU16>(&self, previous: Kairos, kairotic: T) -> Kairos {
        let (physical, logical) = self.advance(Some(previous));
        Kairos::new(physical, logical, self.station_id, kairotic)
    }

    /// Reserves `len` consecutive send positions in one commitment: the *run
    /// mint*, the bulk-ingest form of [`now`](Self::now).
    ///
    /// One time-source reading, one committed compare-and-swap, `len` stamps.
    /// The first member is exactly what `now` would have minted, each later
    /// one is the send step past its predecessor against that frozen reading,
    /// and the clock commits the run's *last* member, so every subsequent
    /// mint or receive strictly dominates every member.
    ///
    /// The run is therefore byte-for-byte what `len` individual `now` calls
    /// against a frozen reading would have committed, minus `len - 1`
    /// readings. Because the members chain under the `metis` chain law by
    /// construction, a woven run coalesces in the identity plane, the
    /// snapshot codec, and the order thread by birthright rather than by
    /// accident.
    ///
    /// All members share this clock's station and one `kairotic` tag, since
    /// the chain law requires both constant. Mint separately where the tag
    /// must vary.
    ///
    /// # Budget
    ///
    /// A run advances logical time by `len` positions at once, so bulk
    /// minting spends the logical space --- one physical unit per
    /// `u16::MAX` positions --- proportionally faster than typing. `len` is
    /// `u32`, so one reservation is bounded; the strict-ascent, chaining, and
    /// domination guarantees above hold below the `u64::MAX` physical
    /// saturation edge this door shares with every other mint, past which
    /// physical saturates and the logical cursor wraps.
    ///
    /// Recorded skews are exactly the maxima that many individual mints
    /// would have folded, so a unit-crossing run runs the committed physical
    /// ahead of its reading and the stats say so.
    pub fn now_run<T: ToU16>(&self, len: core::num::NonZeroU32, kairotic: T) -> KairosRun {
        let (first_physical, first_logical) =
            self.advance_run_from(self.time_source.now(), len.get());
        KairosRun::new(
            Kairos::new(first_physical, first_logical, self.station_id, kairotic),
            len.get(),
        )
    }

    /// Folds a remote stamp into this clock without minting one: the Hybrid Logical
    /// Clock *receive* rule.
    ///
    /// [`now`](Self::now) is the local *send* step and [`after`](Self::after)
    /// fuses receive-and-send into one mint; this is the pure *receive*. It
    /// advances the clock past `remote` so every subsequent mint strictly
    /// exceeds it, and returns nothing, because no event is created.
    ///
    /// Same fold as the minting doors, so infallible, lock-free, and
    /// recording skew identically. It is a receive *event*, not an idempotent
    /// merge: observing an already-dominated stamp still advances the clock,
    /// spending a logical tick where the physical component does not move.
    /// Hence against an unmoving reading, `observe` then `now` lands one tick
    /// above `after(remote)`; both strictly dominate `remote` either way.
    ///
    /// # Trust
    ///
    /// This door advances `physical` to dominate `remote`, so a far-future
    /// stamp pulls the clock forward by that whole gap --- and the jump
    /// cannot be bounded while the door stays both infallible and
    /// dominating. Benign for honest peers, where skew is recorded in
    /// [`stats`](Self::stats) rather than silently corrected, and an attack
    /// surface once stamps arrive from untrusted ones.
    /// [`try_observe`](Self::try_observe) is the bounded opt-in form.
    pub fn observe(&self, remote: Kairos) {
        let _ = self.advance(Some(remote));
    }

    /// The bounded, trust-aware [`observe`](Self::observe): folds `remote` in only
    /// if its forward skew is within `max_forward_skew`, else rejects it and leaves
    /// the clock untouched.
    ///
    /// The caller names how far ahead of the local reading a remote may
    /// legitimately be, and a stamp beyond that is refused.
    ///
    /// The bound is on *forward* skew only, which is the attack surface: the
    /// checked quantity is the remote's lead over a single time-source
    /// reading. A remote at or behind the reading has zero forward skew and
    /// is always admitted, since it cannot drag the clock forward, though it
    /// still ticks logical. The bound is inclusive, and caps what the
    /// *remote* contributes rather than re-bounding skew the clock already
    /// carries from its own history.
    ///
    /// On `Ok` the effect is exactly [`observe`](Self::observe); on `Err` the
    /// clock is unchanged, so the receive is all-or-nothing. Check and fold
    /// share one reading, so they cannot drift apart.
    ///
    /// # Errors
    ///
    /// [`SkewExceeded`] when forward skew over the current reading exceeds
    /// `max_forward_skew`; the clock is left unchanged.
    ///
    /// # Trust
    ///
    /// A skew bound defends only when the remote's `station_id` is
    /// authenticated. Without authentication an adversary simply rotates the
    /// claimed identity. Minerva owns the mechanism; the bound *value* and the
    /// policy for a rejected event are the caller's.
    pub fn try_observe(&self, remote: Kairos, max_forward_skew: u64) -> Result<(), SkewExceeded> {
        let current_physical = self.time_source.now();
        let observed_forward_skew = remote.physical().saturating_sub(current_physical);
        if observed_forward_skew > max_forward_skew {
            return Err(SkewExceeded {
                observed_forward_skew,
                max_forward_skew,
            });
        }
        let _ = self.advance_from(current_physical, Some(remote));
        Ok(())
    }

    /// The clock's health counters: the skew maxima, the configured warning
    /// threshold, and the contention high-water mark.
    ///
    /// Each field is read independently and relaxed, so under concurrent
    /// minting the values can come from different moments. Each is a
    /// monotone maximum and individually sound; only their mutual
    /// consistency is approximate.
    #[must_use]
    pub fn stats(&self) -> ClockStats {
        ClockStats {
            max_forward_skew_ns: self.max_forward_skew.load(Ordering::Relaxed),
            max_backward_skew_ns: self.max_backward_skew.load(Ordering::Relaxed),
            skew_warning_threshold_ns: self.config.max_skew_warning_threshold,
            peak_attempts: self.peak_attempts.load(Ordering::Relaxed),
        }
    }

    /// Samples the time source once and folds the optional remote through the
    /// lock-free [`advance_from`](Self::advance_from) heart. The mint and receive
    /// paths ([`now`](Self::now), [`after`](Self::after), [`observe`](Self::observe))
    /// all enter here; [`try_observe`](Self::try_observe) bypasses it to inspect the
    /// sampled reading before committing.
    fn advance(&self, previous: Option<Kairos>) -> (u64, u16) {
        self.advance_from(self.time_source.now(), previous)
    }

    /// The run mint's commitment shell: one `fold::advance_run` per attempt,
    /// committed exactly as [`advance_from`](Self::advance_from) commits a
    /// single step (the committed word is the run's last member), returning
    /// the run's *first* `(physical, logical)`. Skew is recorded per attempt
    /// off the one shared reading, as the single-step shell records it.
    fn advance_run_from(&self, current_physical: u64, len: u32) -> (u64, u16) {
        let mut attempt = 1;

        loop {
            let old_state = self.state.load(Ordering::Acquire);

            let advance = fold::advance_run(ClockState(old_state).last(), current_physical, len);
            if advance.forward_skew > 0 {
                let _ = self
                    .max_forward_skew
                    .fetch_max(advance.forward_skew, Ordering::Relaxed);
            }
            if advance.backward_skew > 0 {
                let _ = self
                    .max_backward_skew
                    .fetch_max(advance.backward_skew, Ordering::Relaxed);
            }

            let new_state = ClockState::minted(advance.physical, advance.logical).0;

            if self
                .state
                .compare_exchange(old_state, new_state, Ordering::SeqCst, Ordering::Acquire)
                .is_ok()
            {
                let _ = self.peak_attempts.fetch_max(attempt, Ordering::Relaxed);
                return (advance.first_physical, advance.first_logical);
            }

            // Same lock-free posture as `advance_from`: a committer is
            // guaranteed each round, backoff spins without touching the
            // stamp time source.
            self.config.backoff.backoff(attempt);
            attempt = attempt.saturating_add(1);
        }
    }

    /// The lock-free shell over the pure `fold::advance` heart: commits the
    /// fold's next `(physical, logical)` with a single compare-and-swap and
    /// retries on contention. Infallible. `current_physical` is sampled once by
    /// the caller and reused across retries, so a skew-checking caller
    /// ([`try_observe`](Self::try_observe)) commits against the very reading it
    /// bound-checked. The caller decides what to do with the committed position:
    /// wrap it in a [`Kairos`] (a mint) or discard it (an observe).
    ///
    /// The shell owns only commitment and bookkeeping: the time semantics live
    /// whole in `fold.rs` (shared with [`LocalClock`], proven by the S86
    /// harnesses). Skew observations are recorded per attempt, before the
    /// compare-and-swap, exactly as the pre-S188 inline fold recorded them.
    fn advance_from(&self, current_physical: u64, previous: Option<Kairos>) -> (u64, u16) {
        let remote = previous.map(|prev| (prev.physical(), prev.logical()));
        let mut attempt = 1;

        loop {
            let old_state = self.state.load(Ordering::Acquire);

            let advance = fold::advance(ClockState(old_state).last(), current_physical, remote);
            if advance.forward_skew > 0 {
                let _ = self
                    .max_forward_skew
                    .fetch_max(advance.forward_skew, Ordering::Relaxed);
            }
            if advance.backward_skew > 0 {
                let _ = self
                    .max_backward_skew
                    .fetch_max(advance.backward_skew, Ordering::Relaxed);
            }

            let new_state = ClockState::minted(advance.physical, advance.logical).0;

            if self
                .state
                .compare_exchange(old_state, new_state, Ordering::SeqCst, Ordering::Acquire)
                .is_ok()
            {
                let _ = self.peak_attempts.fetch_max(attempt, Ordering::Relaxed);
                return (advance.physical, advance.logical);
            }

            // CAS lost a race; another thread committed a newer state. Recompute
            // and retry. Generation is lock-free (not wait-free): a committer is
            // guaranteed each round, and backoff plus a finite contender count
            // bound this in practice. Backoff spins; it never touches the stamp
            // time source.
            self.config.backoff.backoff(attempt);
            // Saturating: the infallible loop must not panic on overflow even in
            // the impossible u32::MAX-retry limit.
            attempt = attempt.saturating_add(1);
        }
    }
}

/// A thread-safe clock with an erased, heap-allocated time source.
pub type DynClock = Clock<Box<dyn TimeSource>>;