batpak 0.9.0

use crate::store::StoreError;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::Instant;

/// Runtime clock source for store timestamping, evidence, and deterministic tests.
///
/// Boundary: this seam owns every clock value that lands on disk, appears in a
/// public report or receipt, or participates in identity such as UUIDv7 wall
/// bits. Process-local wait deadlines still use `Instant`: cursor pull waits,
/// frontier waits, and writer idle backoff compute elapsed time for local
/// scheduling only, and never become durable bytes or receipt/report identity.
///
/// Production uses [`SystemClock`]. Tests and embeddings that need repeatable
/// store behavior can provide a custom implementation and install it with
/// [`crate::store::StoreConfig::with_clock`].
pub trait Clock: Send + Sync {
    /// Return microseconds since the Unix epoch.
    fn now_us(&self) -> i64;
    /// Return nanoseconds since the Unix epoch, saturating on overflow.
    fn now_wall_ns(&self) -> i64;
    /// Return process-local monotonic nanoseconds.
    fn now_mono_ns(&self) -> i64;
    /// Return the process-epoch marker for monotonic metadata.
    fn process_boot_ns(&self) -> u64;
}

/// Returns microseconds since Unix epoch, saturating to `i64::MAX` if the system
/// clock is beyond year ~292,277 (treat the max value as a clock-malfunction
/// signal). No panic; cache staleness checks downstream see a saturated value
/// and force a replay rather than poisoning the process.
fn system_now_us() -> i64 {
    let micros = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_micros();
    i64::try_from(micros).unwrap_or(i64::MAX)
}

/// Convert a public clock reading to persisted wall-clock milliseconds.
///
/// Custom clocks must report microseconds since Unix epoch as a non-negative
/// `i64`. Negative values are rejected as invalid caller input rather than
/// panicking in append/batch hot paths.
pub(crate) fn wall_ms_from_timestamp_us(timestamp_us: i64) -> Result<u64, StoreError> {
    if timestamp_us < 0 {
        return Err(StoreError::InvalidClock {
            timestamp_us,
            reason: "timestamp_us must be >= 0 microseconds since Unix epoch".into(),
        });
    }
    Ok((timestamp_us / 1000).cast_unsigned())
}

fn system_now_wall_ns_saturating() -> i64 {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    i64::try_from(nanos).unwrap_or(i64::MAX)
}

/// Process-wide monotonic anchor. Captured on first call; subsequent calls read
/// the elapsed nanoseconds from `Instant::now()` relative to this anchor.
///
/// The anchor couples two facts:
///   1. `anchor_instant`: the `Instant` captured at first call.
///   2. `anchor_boot_ns`: a u64 marker that identifies *this* process's
///      monotonic epoch. Any cached monotonic value persisted to disk and then
///      read back by a different process MUST compare its `process_boot_ns`
///      against this value — mismatch means the monotonic value belongs to a
///      different process's clock and cannot be trusted.
pub(crate) struct MonotonicAnchor {
    anchor_instant: Instant,
    anchor_boot_ns: u64,
}

impl MonotonicAnchor {
    fn get() -> &'static Self {
        static ANCHOR: OnceLock<MonotonicAnchor> = OnceLock::new();
        ANCHOR.get_or_init(|| {
            // The boot marker is the wall-clock time at anchor creation, encoded
            // as nanoseconds since Unix epoch and saturated to u64. Two processes
            // booting in the same nanosecond on the same machine would collide,
            // which is acceptable (they would both re-project on mismatch anyway).
            let wall_ns = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_nanos();
            let anchor_boot_ns = u64::try_from(wall_ns).unwrap_or(u64::MAX);
            MonotonicAnchor {
                anchor_instant: Instant::now(),
                anchor_boot_ns,
            }
        })
    }

    fn now_mono_ns(&self) -> i64 {
        let elapsed = self.anchor_instant.elapsed().as_nanos();
        i64::try_from(elapsed).unwrap_or(i64::MAX)
    }
}

/// Returns monotonic nanoseconds since the process-wide anchor. Guaranteed
/// non-decreasing within a single process; meaningless across processes
/// (use [`Clock::process_boot_ns`] to detect cross-process comparisons).
///
/// Saturates to `i64::MAX` if the process has been alive for more than
/// ~292 years.
#[derive(Clone)]
pub struct SystemClock {
    anchor: &'static MonotonicAnchor,
}

impl SystemClock {
    /// Create a production clock backed by system wall time and process-local monotonic time.
    pub fn new() -> Self {
        Self {
            anchor: MonotonicAnchor::get(),
        }
    }
}

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

impl Clock for SystemClock {
    fn now_us(&self) -> i64 {
        system_now_us()
    }

    fn now_wall_ns(&self) -> i64 {
        system_now_wall_ns_saturating()
    }

    fn now_mono_ns(&self) -> i64 {
        self.anchor.now_mono_ns()
    }

    fn process_boot_ns(&self) -> u64 {
        self.anchor.anchor_boot_ns
    }
}

struct FnClock {
    inner: Arc<dyn Fn() -> i64 + Send + Sync>,
    anchor: &'static MonotonicAnchor,
}

impl FnClock {
    fn new(inner: Arc<dyn Fn() -> i64 + Send + Sync>) -> Self {
        Self {
            inner,
            anchor: MonotonicAnchor::get(),
        }
    }
}

impl Clock for FnClock {
    fn now_us(&self) -> i64 {
        (self.inner)()
    }

    fn now_wall_ns(&self) -> i64 {
        self.now_us().saturating_mul(1000)
    }

    fn now_mono_ns(&self) -> i64 {
        self.anchor.now_mono_ns()
    }

    fn process_boot_ns(&self) -> u64 {
        self.anchor.anchor_boot_ns
    }
}

pub(crate) fn clock_from_fn(inner: Arc<dyn Fn() -> i64 + Send + Sync>) -> Arc<dyn Clock> {
    Arc::new(FnClock::new(inner))
}

/// Non-decreasing wrapper around a clock source.
///
/// A user clock that regresses (e.g. NTP jump, manual reset) would poison age
/// comparisons — a slot cached at `now=1000` and read at `now=500` would look
/// like it's `-500` µs old, and a naive check can misclassify it. This wrapper
/// clamps each observed value to `max(last, new)`: once we see a value, we
/// never return anything smaller. Regressions emit `tracing::error!` with the
/// previous and new values and return the previous value — the user's clock
/// is broken, but the store keeps running.
#[derive(Clone)]
pub(crate) struct MonotonicClock {
    inner: Arc<dyn Clock>,
    last: Arc<AtomicI64>,
    last_wall_ns: Arc<AtomicI64>,
}

impl MonotonicClock {
    /// Wrap a clock. The returned handle is cloneable
    /// and stores shared state (`AtomicI64`) in an `Arc`, so clones observe the
    /// same non-decreasing sequence.
    ///
    /// `now_us` (microseconds) and `now_wall_ns` (nanoseconds) are each clamped
    /// against their own atomic. They are independent sequences — `SystemClock`
    /// derives them from two separate `SystemTime::now()` calls — so sharing a
    /// single atomic would cross-contaminate the two.
    pub(crate) fn wrap(inner: Arc<dyn Clock>) -> Self {
        Self {
            inner,
            last: Arc::new(AtomicI64::new(i64::MIN)),
            last_wall_ns: Arc::new(AtomicI64::new(i64::MIN)),
        }
    }

    /// Clamp `raw` to be non-decreasing relative to the highest value ever
    /// installed in `slot`. Returns `raw` when it advances the slot, otherwise
    /// logs a regression at `error` level and returns the previously installed
    /// value. The memory ordering (Acquire load, AcqRel/Acquire CAS) is the
    /// audited pattern shared by every non-decreasing clock sequence.
    fn clamp_non_decreasing(slot: &AtomicI64, raw: i64, what: &str) -> i64 {
        // Compare-and-swap loop: install `raw` if it's newer than the slot,
        // otherwise report a regression and keep the old value.
        loop {
            let prev = slot.load(Ordering::Acquire);
            if raw >= prev {
                match slot.compare_exchange(prev, raw, Ordering::AcqRel, Ordering::Acquire) {
                    Ok(_) => return raw,
                    Err(_) => continue, // another thread stored a newer value; retry
                }
            } else {
                tracing::error!("user clock regressed ({}): prev={} new={}", what, prev, raw);
                return prev;
            }
        }
    }

    /// Sample the wrapped clock and return a value that is never smaller than
    /// any value previously returned by this [`MonotonicClock`] (or any clone
    /// of it). A regression is logged at `error` level.
    pub(crate) fn now_us(&self) -> i64 {
        Self::clamp_non_decreasing(&self.last, self.inner.now_us(), "us")
    }
}

impl Clock for MonotonicClock {
    fn now_us(&self) -> i64 {
        MonotonicClock::now_us(self)
    }

    fn now_wall_ns(&self) -> i64 {
        Self::clamp_non_decreasing(&self.last_wall_ns, self.inner.now_wall_ns(), "wall_ns")
    }

    fn now_mono_ns(&self) -> i64 {
        self.inner.now_mono_ns()
    }

    fn process_boot_ns(&self) -> u64 {
        self.inner.process_boot_ns()
    }
}

#[cfg(test)]
mod tests {
    use super::{Clock, FnClock, MonotonicClock, SystemClock};
    use std::sync::atomic::{AtomicI64, Ordering};
    use std::sync::Arc;

    /// A mutable test clock whose `now_us` and `now_wall_ns` are independently
    /// programmable, so we can inject backward jumps into either sequence.
    struct AdjustableClock {
        us: AtomicI64,
        wall_ns: AtomicI64,
    }

    impl AdjustableClock {
        fn new(us: i64, wall_ns: i64) -> Arc<Self> {
            Arc::new(Self {
                us: AtomicI64::new(us),
                wall_ns: AtomicI64::new(wall_ns),
            })
        }
        fn set_us(&self, v: i64) {
            self.us.store(v, Ordering::SeqCst);
        }
        fn set_wall_ns(&self, v: i64) {
            self.wall_ns.store(v, Ordering::SeqCst);
        }
    }

    impl Clock for AdjustableClock {
        fn now_us(&self) -> i64 {
            self.us.load(Ordering::SeqCst)
        }
        fn now_wall_ns(&self) -> i64 {
            self.wall_ns.load(Ordering::SeqCst)
        }
        fn now_mono_ns(&self) -> i64 {
            0
        }
        fn process_boot_ns(&self) -> u64 {
            0
        }
    }

    #[test]
    fn fn_clock_preserves_negative_wall_values_but_not_monotonic_time() {
        // Anchor a monotonic floor FIRST: FnClock scripts only the wall clock,
        // so its monotonic reading must come from the same process-wide
        // MonotonicAnchor that SystemClock reads. The bounded spin (at most one
        // clock-resolution tick) pushes the floor past the trivial constants
        // (0, 1) a body-stubbing mutant substitutes for `now_mono_ns`.
        let system = SystemClock::new();
        let mut floor = system.now_mono_ns();
        while floor <= 1 {
            floor = system.now_mono_ns();
        }

        let clock = FnClock::new(Arc::new(|| -7));

        assert_eq!(
            clock.now_us(),
            -7,
            "PROPERTY: FnClock must expose malformed caller wall time for validation"
        );
        assert_eq!(
            clock.now_wall_ns(),
            -7_000,
            "PROPERTY: wall nanoseconds come from the caller wall clock, not the monotonic anchor"
        );
        let mono = clock.now_mono_ns();
        let ceiling = system.now_mono_ns();
        assert!(
            floor <= mono && mono <= ceiling,
            "PROPERTY: process-local monotonic evidence reads the shared process anchor \
             (floor {floor} <= mono {mono} <= ceiling {ceiling}), never an echo of the \
             negative caller wall clock or a stubbed constant"
        );
    }

    #[test]
    fn now_wall_ns_clamps_on_regression() {
        let inner = AdjustableClock::new(1_000, 5_000_000_000);
        let mono = MonotonicClock::wrap(Arc::clone(&inner) as Arc<dyn Clock>);

        assert_eq!(
            mono.now_wall_ns(),
            5_000_000_000,
            "PROPERTY: forward wall clock passes through unchanged"
        );
        // Inner wall clock jumps backward (NTP step-back / manual reset).
        inner.set_wall_ns(1_000_000_000);
        assert_eq!(
            mono.now_wall_ns(),
            5_000_000_000,
            "PROPERTY: a regressing wall clock stalls at the highest value seen, never moving backward"
        );
        // A subsequent forward move past the high-water mark advances again.
        inner.set_wall_ns(6_000_000_000);
        assert_eq!(
            mono.now_wall_ns(),
            6_000_000_000,
            "PROPERTY: forward progress past the high-water mark resumes"
        );
    }

    #[test]
    fn now_us_and_now_wall_ns_clamp_states_are_independent() {
        let inner = AdjustableClock::new(1_000, 5_000_000_000);
        let mono = MonotonicClock::wrap(Arc::clone(&inner) as Arc<dyn Clock>);

        // Prime both sequences.
        assert_eq!(mono.now_us(), 1_000);
        assert_eq!(mono.now_wall_ns(), 5_000_000_000);

        // Regress only the microsecond clock; wall_ns must still advance freely.
        inner.set_us(500);
        inner.set_wall_ns(9_000_000_000);
        assert_eq!(
            mono.now_us(),
            1_000,
            "PROPERTY: regressed us sequence stalls at its own high-water mark"
        );
        assert_eq!(
            mono.now_wall_ns(),
            9_000_000_000,
            "PROPERTY: wall_ns is not contaminated by the us regression (separate atomic)"
        );

        // Regress only the wall clock; us must still advance freely.
        inner.set_us(2_000);
        inner.set_wall_ns(1_000_000_000);
        assert_eq!(
            mono.now_us(),
            2_000,
            "PROPERTY: us is not contaminated by the wall_ns regression (separate atomic)"
        );
        assert_eq!(
            mono.now_wall_ns(),
            9_000_000_000,
            "PROPERTY: regressed wall_ns sequence stalls at its own high-water mark"
        );
    }

    #[test]
    fn system_clock_now_wall_ns_reports_real_wall_time() {
        // The production clock must report a genuine UNIX-epoch wall-clock
        // reading in nanoseconds, not a constant stand-in. 1_600_000_000 s
        // (~2020-09-13) is far below any real "now" yet far above the trivial
        // values (`0`, `1`) a body-stubbing mutant would substitute.
        const EARLIEST_PLAUSIBLE_WALL_NS: i64 = 1_600_000_000_000_000_000;

        let reading = SystemClock::new().now_wall_ns();
        assert!(
            reading > EARLIEST_PLAUSIBLE_WALL_NS,
            "PROPERTY: SystemClock::now_wall_ns must read the real wall clock \
             (>{EARLIEST_PLAUSIBLE_WALL_NS} ns), got {reading}"
        );
    }

    #[test]
    fn monotonic_clock_process_boot_ns_delegates_real_anchor() {
        // MonotonicClock must forward the wrapped clock's process-epoch marker
        // rather than synthesize a value. The static MonotonicAnchor encodes the
        // wall-clock nanoseconds captured at first use, so any real boot marker is
        // far above the trivial constants (`0`, `1`) a body-stubbing mutant
        // substitutes. 1_600_000_000 s (~2020-09-13) is below any real "now" yet
        // far above those stubs. No blocking wait: process_boot_ns is a direct
        // return, so a divergent mutant fails an assertion immediately.
        const EARLIEST_PLAUSIBLE_BOOT_NS: u64 = 1_600_000_000_000_000_000;

        let inner = Arc::new(SystemClock::new());
        let inner_boot = inner.process_boot_ns();
        let mono = MonotonicClock::wrap(Arc::clone(&inner) as Arc<dyn Clock>);
        let wrapped_boot = mono.process_boot_ns();

        let mut failures: Vec<String> = Vec::new();
        if wrapped_boot != inner_boot {
            failures.push(format!(
                "PROPERTY: MonotonicClock::process_boot_ns must delegate to the \
                 wrapped clock unchanged: wrapped={wrapped_boot}, inner={inner_boot}"
            ));
        }
        if wrapped_boot <= EARLIEST_PLAUSIBLE_BOOT_NS {
            failures.push(format!(
                "PROPERTY: the process boot marker is the real anchor wall-ns \
                 (>{EARLIEST_PLAUSIBLE_BOOT_NS}), never a trivial stub: got {wrapped_boot}"
            ));
        }
        assert!(failures.is_empty(), "{failures:?}");
    }
}