minerva 0.2.0

extern crate alloc;

use alloc::boxed::Box;
use core::cell::Cell;

use super::super::stamp::{Kairos, ToU16};
use super::super::time::TimeSource;
use super::error::{InvalidStationId, SkewExceeded};
use super::fold;
use super::run::KairosRun;
use super::state::ClockState;
use super::{ClockConfig, ClockStats};

/// The single-threaded minting shell: the same Hybrid Logical Clock as
/// [`Clock`](super::Clock), over the same pure `fold` heart, committed
/// through a [`Cell`] instead of a compare-and-swap loop.
///
/// A [`Clock`](super::Clock) pays for thread-safety on every mint (atomic
/// load, sequentially consistent compare-and-swap, retry with backoff),
/// which buys nothing where contention is structurally impossible: a
/// single-threaded wasm deployment, a per-shard clock owned by one worker,
/// a simulation. `LocalClock` is the zero-cost form: one fold, one `Cell`
/// store, no fences, no retries. `!Sync` by construction, so the compiler,
/// not a convention, keeps it off shared paths; sending it between threads
/// remains fine.
///
/// ```compile_fail
/// // The concurrency posture is a type-level fact: a `LocalClock` cannot be
/// // shared across threads. Reach for `Clock` where sharing is the point.
/// fn shared<T: Sync>() {}
/// shared::<minerva::kairos::LocalClock>();
/// ```
///
/// # One fold, two shells
///
/// This is not a second clock implementation. The time semantics live
/// whole in `fold::advance`, the function [`Clock`](super::Clock) commits
/// through its CAS loop and the S86 Kani harnesses prove. A shell decides
/// only how the fold's result is committed and when skew readings are
/// recorded; shell agreement over any operation tape is pinned by
/// `prop_local_clock_agrees_with_atomic_clock_on_any_tape`.
///
/// # API parity
///
/// [`now`](Self::now) / [`after`](Self::after) / [`observe`](Self::observe) /
/// [`try_observe`](Self::try_observe) / [`stats`](Self::stats) /
/// [`station_id`](Self::station_id) carry the contracts documented on
/// [`Clock`](super::Clock), including `observe`'s per-call logical tick and
/// `try_observe`'s all-or-nothing bound check against a single time-source
/// reading. [`ClockStats::peak_attempts`] reads `1` once anything has been
/// committed: every commitment here succeeds on its first attempt, which is
/// the same quantity the atomic shell reports as contention-free operation.
pub struct LocalClock<TS: TimeSource = Box<dyn TimeSource>> {
    time_source: TS,
    /// Spatial identity, fixed at construction (the `Clock` posture).
    station_id: u32,
    /// Packed `ClockState`, committed by plain store: no rival writer
    /// exists by construction.
    state: Cell<u128>,
    max_forward_skew: Cell<u64>,
    max_backward_skew: Cell<u64>,
    max_skew_warning_threshold: u64,
}

impl<TS: TimeSource> LocalClock<TS> {
    /// Creates a single-threaded clock with the default skew-warning
    /// threshold (the [`ClockConfig`] default; there is no backoff to
    /// configure, because there is no contention to back off from).
    ///
    /// # Errors
    ///
    /// Returns [`InvalidStationId`] if `station_id` is zero.
    pub fn new(time_source: TS, station_id: u32) -> Result<Self, InvalidStationId> {
        Self::with_warning_threshold(
            time_source,
            station_id,
            ClockConfig::default().max_skew_warning_threshold,
        )
    }

    /// Creates a single-threaded clock with an explicit skew-warning
    /// threshold (nanoseconds in the time source's own units), the one
    /// [`ClockConfig`] knob that applies to this shell.
    ///
    /// # Errors
    ///
    /// Returns [`InvalidStationId`] if `station_id` is zero.
    pub fn with_warning_threshold(
        time_source: TS,
        station_id: u32,
        max_skew_warning_threshold: u64,
    ) -> Result<Self, InvalidStationId> {
        if station_id == 0 {
            return Err(InvalidStationId);
        }
        Ok(Self {
            time_source,
            station_id,
            state: Cell::new(ClockState::UNINIT.0),
            max_forward_skew: Cell::new(0),
            max_backward_skew: Cell::new(0),
            max_skew_warning_threshold,
        })
    }

    /// This clock's station identity, fixed at construction.
    #[must_use]
    pub const fn station_id(&self) -> u32 {
        self.station_id
    }

    /// Generates a new [`Kairos`] stamp: the local *send* step
    /// ([`Clock::now`](super::Clock::now)'s contract, single-threaded shell).
    pub fn now<T: ToU16>(&self, kairotic: T) -> Kairos {
        let (physical, logical) = self.advance(None);
        Kairos::new(physical, logical, self.station_id, kairotic)
    }

    /// Generates a stamp strictly after `previous`: receive-and-send fused
    /// ([`Clock::after`](super::Clock::after)'s contract).
    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 `Cell` store: the
    /// run mint ([`Clock::now_run`](super::Clock::now_run)'s contract,
    /// single-threaded shell). One reading, one commitment, `len` stamps
    /// whose succession, and whose recorded skew maxima, are exactly what
    /// `len` individual mints against a frozen reading would have
    /// committed and folded.
    pub fn now_run<T: ToU16>(&self, len: core::num::NonZeroU32, kairotic: T) -> KairosRun {
        let advance = fold::advance_run(
            ClockState(self.state.get()).last(),
            self.time_source.now(),
            len.get(),
        );
        if advance.forward_skew > self.max_forward_skew.get() {
            self.max_forward_skew.set(advance.forward_skew);
        }
        if advance.backward_skew > self.max_backward_skew.get() {
            self.max_backward_skew.set(advance.backward_skew);
        }
        self.state
            .set(ClockState::minted(advance.physical, advance.logical).0);
        KairosRun::new(
            Kairos::new(
                advance.first_physical,
                advance.first_logical,
                self.station_id,
                kairotic,
            ),
            len.get(),
        )
    }

    /// Folds a remote stamp in without minting: the pure *receive*
    /// ([`Clock::observe`](super::Clock::observe)'s contract, per-call logical tick and unbounded
    /// forward adoption included; the bounded form is
    /// [`try_observe`](Self::try_observe)).
    pub fn observe(&self, remote: Kairos) {
        let _ = self.advance(Some(remote));
    }

    /// The bounded, trust-aware receive ([`Clock::try_observe`](super::Clock::try_observe)'s contract):
    /// folds `remote` in only if its forward skew over a single time-source
    /// reading is within `max_forward_skew`, else leaves the clock untouched.
    ///
    /// # Errors
    /// [`SkewExceeded`] when the remote's forward skew over the current
    /// reading exceeds `max_forward_skew`; the clock is left unchanged.
    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 health statistics, shape-identical to [`Clock::stats`](super::Clock::stats).
    /// `peak_attempts` is `1` once anything has committed: a `Cell` store
    /// always succeeds on its first attempt.
    #[must_use]
    pub fn stats(&self) -> ClockStats {
        let attempts = u32::from(ClockState(self.state.get()).last().is_some());
        ClockStats {
            max_forward_skew_ns: self.max_forward_skew.get(),
            max_backward_skew_ns: self.max_backward_skew.get(),
            skew_warning_threshold_ns: self.max_skew_warning_threshold,
            peak_attempts: attempts,
        }
    }

    /// Samples the time source once and commits one fold (the
    /// `Clock::advance` mirror).
    fn advance(&self, previous: Option<Kairos>) -> (u64, u16) {
        self.advance_from(self.time_source.now(), previous)
    }

    /// The single-threaded shell over the shared pure fold: one
    /// [`fold::advance`], one `Cell` store, skew maxima folded in place. No
    /// loop: there is no rival writer to lose a race against.
    fn advance_from(&self, current_physical: u64, previous: Option<Kairos>) -> (u64, u16) {
        let remote = previous.map(|prev| (prev.physical(), prev.logical()));
        let advance = fold::advance(
            ClockState(self.state.get()).last(),
            current_physical,
            remote,
        );
        if advance.forward_skew > self.max_forward_skew.get() {
            self.max_forward_skew.set(advance.forward_skew);
        }
        if advance.backward_skew > self.max_backward_skew.get() {
            self.max_backward_skew.set(advance.backward_skew);
        }
        self.state
            .set(ClockState::minted(advance.physical, advance.logical).0);
        (advance.physical, advance.logical)
    }
}

/// A single-threaded clock with an erased, heap-allocated time source.
pub type DynLocalClock = LocalClock<Box<dyn TimeSource>>;