minerva 0.2.0

//! The pure Hybrid Logical Clock fold: the crate's time algebra, extracted
//! from the concurrency shell that runs it (S188, ruling R-19).
//!
//! [`advance`] is the *entire* semantics of a clock step: given the last
//! committed `(physical, logical)` (or none), one time-source reading, and an
//! optional remote stamp's `(physical, logical)`, it computes the next
//! committed pair and reports the skew it observed along the way. It is a
//! total, pure, heap-free function over machine words: no atomics, no time
//! source, no allocation, no `Kairos` (the caller unpacks the two fields it
//! folds), so the shells that share it stay free to choose their own
//! concurrency story and the checkers reach it whole.
//!
//! Two shells run this fold today: [`Clock`](super::Clock) commits it through
//! a compare-and-swap loop (thread-safe, lock-free, backoff on contention)
//! and [`LocalClock`](super::LocalClock) commits it through a `Cell` (single
//! threaded, no fences, no retries). Their agreement is not a test
//! obligation but a *construction*: there is one fold, and a shell only
//! decides how its result is committed. The S86 Kani harnesses prove the
//! fold's laws (strict domination, monotonicity under arbitrary skew, the
//! ceiling-logical rollover) through the `Clock` shell, whose single-threaded
//! first-attempt CAS is exactly one fold application; the shell-agreement
//! property test pins that `LocalClock` tells the same story on every tape.
//!
//! The fold is deliberately self-contained (plain integer arguments, no
//! sibling imports) so it stays eligible for the dual-homed pure-core proof
//! treatment (ruling R-18) without restructuring, if a session ever takes
//! the clock into the Creusot rung.

/// One advance's outcome: the pair to commit, and the skew observed while
/// computing it (zero when none). The shell owns what to do with each half:
/// commit the pair (CAS or set), fold the skews into its running maxima.
pub(super) struct Advance {
    /// The physical component of the next committed state.
    pub(super) physical: u64,
    /// The logical component of the next committed state.
    pub(super) logical: u16,
    /// How far the committed physical ran ahead of the sampled reading
    /// (remote drag or retained local lead); zero when it did not.
    pub(super) forward_skew: u64,
    /// How far the sampled reading ran behind the last committed physical
    /// (the source moved backward); zero when it did not.
    pub(super) backward_skew: u64,
}

/// One run reservation's outcome: the first stamp of the run, the pair to
/// commit (the run's last stamp, so every subsequent mint strictly dominates
/// every run member), and the skew observed by the single reading the run
/// consumed. The shell owns commitment and bookkeeping exactly as with
/// [`Advance`].
pub(super) struct AdvanceRun {
    /// The physical component of the run's first stamp.
    pub(super) first_physical: u64,
    /// The logical component of the run's first stamp.
    pub(super) first_logical: u16,
    /// The physical component of the next committed state (the run's last
    /// stamp).
    pub(super) physical: u64,
    /// The logical component of the next committed state (the run's last
    /// stamp).
    pub(super) logical: u16,
    /// As [`Advance::forward_skew`], observed once for the whole run.
    pub(super) forward_skew: u64,
    /// As [`Advance::backward_skew`], observed once for the whole run.
    pub(super) backward_skew: u64,
}

/// Rank slots per physical unit: logical values run `0..=u16::MAX - 1`
/// (`u16::MAX` is the rollover sentinel [`advance`] never commits), so each
/// physical unit carries exactly `u16::MAX as u64` successor positions.
const RANK_SLOTS: u64 = u16::MAX as u64;

/// The pair `n` send-step successors past `(physical, logical)`: the closed
/// form of iterating the fold's frozen-reading send step (which is the
/// snapshot codec's `rank_successor`; the S197 harness proves those equal)
/// `n` times. Total and heap-free; the physical carry saturates at the
/// `u64::MAX` ceiling exactly as the iterated step does (the documented pin
/// edge, where domination is arithmetically impossible), and the logical
/// cursor keeps its modular position either way, again as the iterated step.
/// `logical` must be a committed value (below the sentinel), which every
/// fold output is.
pub(super) const fn rank_offset(physical: u64, logical: u16, n: u64) -> (u64, u16) {
    // `logical < RANK_SLOTS` and `n` arrives from a `u32` run length, so the
    // sum stays far under `u64::MAX`; the arithmetic is total regardless via
    // the saturating carry.
    let total = logical as u64 + n;
    let carried = physical.saturating_add(total / RANK_SLOTS);
    (carried, (total % RANK_SLOTS) as u16)
}

/// The Hybrid Logical Clock *run* step: one fold application reserving `len`
/// consecutive send positions against a single time-source reading.
///
/// The first stamp is exactly `advance(last, reading, None)`; the
/// committed pair is the run's last stamp, `len - 1` send-step successors
/// past the first. Because a send step against a reading at or behind the
/// committed physical *is* the successor rule (the S197 harness's frozen
/// -reading identity), this equals folding [`advance`] `len` times over the
/// one frozen reading: the run members are the very stamps that many
/// individual mints would have committed, minus the re-reads.
///
/// The reported skews are the MAXIMA the iterated mints would have folded
/// into the shell's stats, not the first mint's alone: a run whose carry
/// crosses a physical unit runs the committed physical ahead of the one
/// reading (forward skew, the retained local lead), and every mint past the
/// crossing reads the source behind the committed physical (backward skew
/// off the penultimate member). Without this, `ClockStats` would
/// under-report exactly the gap the run created (the S211 gate review's
/// finding).
///
/// `len` arrives from a `NonZeroU32`, so it is at least one; the saturating
/// subtractions only keep the function total on the whole `u32` domain.
pub(super) const fn advance_run(last: Option<(u64, u16)>, reading: u64, len: u32) -> AdvanceRun {
    let first = advance(last, reading, None);
    let (physical, logical) = rank_offset(
        first.physical,
        first.logical,
        (len as u64).saturating_sub(1),
    );
    // Iterated send `i` computes BOTH skews off the physical it starts
    // from (the fold reads `last_physical` before any rollover commits a
    // higher one), so each mint past the first reports
    // `physical_{i-1} - reading`, and the maxima sit at the PENULTIMATE
    // member for forward and backward alike, joined with the first mint's
    // own observations (which alone see the pre-run state and any genuine
    // source regression). Not the final committed physical: when the last
    // member itself rolls a unit, no iterated mint ever read that value,
    // and the run must not report a skew the mints would not have (the
    // extended Kani harness found exactly this corner in the first cut of
    // this fix).
    let (penultimate, _) = rank_offset(
        first.physical,
        first.logical,
        (len as u64).saturating_sub(2),
    );
    let trailing = if len >= 2 {
        penultimate.saturating_sub(reading)
    } else {
        0
    };
    let forward_skew = if trailing > first.forward_skew {
        trailing
    } else {
        first.forward_skew
    };
    let backward_skew = if trailing > first.backward_skew {
        trailing
    } else {
        first.backward_skew
    };
    AdvanceRun {
        first_physical: first.physical,
        first_logical: first.logical,
        physical,
        logical,
        forward_skew,
        backward_skew,
    }
}

/// The Hybrid Logical Clock step. `last` is the shell's committed state
/// (`None` before the first mint), `reading` is one time-source sample, and
/// `remote` is the stamp being received, if any (its `(physical, logical)`;
/// send-only steps pass `None`).
///
/// The result strictly dominates both the last committed pair and any remote
/// below the physical ceiling: at `u64::MAX` the clock pins (saturates,
/// never wraps), the documented edge the proofs carry as a precondition. A
/// logical that would reach `u16::MAX` rolls into the next physical unit
/// instead (the S86 fold fix: the ceiling tie must not fall through to the
/// kairotic/station tiebreak).
pub(super) const fn advance(
    last: Option<(u64, u16)>,
    reading: u64,
    remote: Option<(u64, u16)>,
) -> Advance {
    match last {
        None => {
            // Uninitialized: adopt the larger of the local reading and any
            // remote physical, seeding logical from a remote at equal time.
            let new_physical = match remote {
                Some((remote_physical, _)) if remote_physical > reading => remote_physical,
                _ => reading,
            };
            let forward_skew = new_physical - reading;
            let new_logical = match remote {
                Some((remote_physical, remote_logical)) if remote_physical == new_physical => {
                    remote_logical.saturating_add(1)
                }
                _ => 0,
            };
            // `u16::MAX` is the overflow sentinel: it rolls into physical
            // here exactly as in the initialized arm, or a fresh clock
            // receiving a ceiling-logical stamp mints a tie whose
            // kairotic/station tiebreak can order the mint *below* the
            // remote (the S86 Kani finding).
            let (physical, logical) = if new_logical == u16::MAX {
                (new_physical.saturating_add(1), 0)
            } else {
                (new_physical, new_logical)
            };
            Advance {
                physical,
                logical,
                forward_skew,
                backward_skew: 0,
            }
        }
        Some((last_physical, last_logical)) => {
            // A reading behind the committed physical is backward skew:
            // reported, never adopted (the clock's output stays monotonic).
            let backward_skew = last_physical.saturating_sub(reading);
            let effective_physical = if reading < last_physical {
                last_physical
            } else {
                reading
            };
            let new_physical = match remote {
                Some((remote_physical, _)) if remote_physical > effective_physical => {
                    remote_physical
                }
                _ => effective_physical,
            };
            let forward_skew = new_physical.saturating_sub(reading);
            let new_logical = if new_physical == last_physical {
                let remote_logical = match remote {
                    Some((remote_physical, remote_logical)) if remote_physical == new_physical => {
                        remote_logical
                    }
                    _ => 0,
                };
                let base = if last_logical > remote_logical {
                    last_logical
                } else {
                    remote_logical
                };
                base.saturating_add(1)
            } else {
                match remote {
                    Some((remote_physical, remote_logical)) if remote_physical == new_physical => {
                        remote_logical.saturating_add(1)
                    }
                    _ => 0,
                }
            };
            let (physical, logical) = if new_logical == u16::MAX {
                (new_physical.saturating_add(1), 0)
            } else {
                (new_physical, new_logical)
            };
            Advance {
                physical,
                logical,
                forward_skew,
                backward_skew,
            }
        }
    }
}