slither 0.4.0

//! Endpoint configuration, and §16.5's injected wall clock.
//!
//! §16.5: *"`now: Instant` is an explicit argument on every mutating call;
//! the cores never read a clock. The initiation timestamp (§5.3) is the one
//! wall-clock read, behind a **clock service injected in the endpoint
//! config**."* That sentence is why a clock seam lives here and nowhere
//! else: `Instant` is the caller's to supply, and the single wall-clock
//! reading the protocol performs is the initiation timestamp.
//!
//! # `Rc`, not `Arc`
//!
//! [`Config`] holds `Rc<dyn WallClock>`, deliberately. `Arc<dyn WallClock +
//! Send + Sync>` is the shape a Rust programmer reaches for by default, and
//! it would quietly make `Config: Send` something consumers depend on — on
//! the one path §16.3 keeps free of `Send` so a hardware-backed static key
//! can drive it (S21).

use std::fmt;
use std::num::NonZeroU64;
use std::rc::Rc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::constants;
use crate::core::Timestamp;
use crate::error::ConfigError;

/// An endpoint's established-session liveness policy.
///
/// **[RATIFIED 2026/08/31 — ruling 282; amended by ruling 283]** The v1
/// values remain the default, but an application may choose a shorter or
/// longer coherent pair for all connections born from one endpoint. Nothing
/// is negotiated on the wire: deployments that rely on a custom profile's
/// loss/failover guarantees must configure their peers out of band.
///
/// This deliberately covers only the two coupled established-session clocks.
/// Handshake retransmission/give-up, introduction retention, ACK/PTO recovery,
/// close linger, and shell lateness remain the fixed v1 constants.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TimingProfile {
    passive_keepalive: Duration,
    dead_timeout: Duration,
}

impl TimingProfile {
    /// The ratified v1 liveness profile: 10 s passive keepalive, 25 s death.
    pub const V1: Self = Self {
        passive_keepalive: constants::KEEPALIVE_TIMEOUT,
        dead_timeout: constants::DEAD_TIMEOUT,
    };

    /// Construct a validated liveness profile.
    ///
    /// The passive interval must be at least
    /// [`PERSISTENT_KEEPALIVE_MIN`](crate::constants::PERSISTENT_KEEPALIVE_MIN)
    /// (1 s): below that, an uncongested keepalive becomes a load generator.
    /// The death timeout must be **strictly greater** than two passive
    /// intervals plus the protocol's initial RTT and twice the shell lateness
    /// bound. Each of two successive keepalive firings can independently be
    /// late because a firing re-anchors the next deadline from the actual send
    /// instant. This preserves one-lost-keepalive room under v1's initial RTT
    /// assumption and avoids making an arrival at the verdict instant depend
    /// on timer ordering. A value that cannot be added to the monotonic clock
    /// at construction is rejected as arithmetic overflow. Every later
    /// deadline derivation is checked again: reaching the platform horizon
    /// retains the logical timer but announces no fabricated earlier instant.
    ///
    /// No peer sees these values. A custom profile is deployment policy, not
    /// negotiation; mismatched peers remain wire-compatible but may make
    /// different liveness decisions.
    pub fn try_new(
        passive_keepalive: Duration,
        dead_timeout: Duration,
    ) -> Result<Self, TimingProfileError> {
        if passive_keepalive < constants::PERSISTENT_KEEPALIVE_MIN {
            return Err(TimingProfileError::KeepaliveTooShort);
        }

        let minimum_dead = passive_keepalive
            .checked_mul(2)
            .and_then(|twice| twice.checked_add(constants::K_INITIAL_RTT))
            .and_then(|with_rtt| with_rtt.checked_add(constants::SHELL_LATENESS_BOUND))
            .and_then(|with_first_lateness| {
                with_first_lateness.checked_add(constants::SHELL_LATENESS_BOUND)
            })
            .ok_or(TimingProfileError::ArithmeticOverflow)?;
        if dead_timeout <= minimum_dead {
            return Err(TimingProfileError::DeadTimeoutTooShort);
        }
        if Instant::now().checked_add(dead_timeout).is_none() {
            return Err(TimingProfileError::ArithmeticOverflow);
        }

        Ok(Self {
            passive_keepalive,
            dead_timeout,
        })
    }

    /// The passive keepalive cadence and contested-connection verdict.
    pub const fn passive_keepalive(&self) -> Duration {
        self.passive_keepalive
    }

    /// The receive-anchored dead-peer timeout.
    pub const fn dead_timeout(&self) -> Duration {
        self.dead_timeout
    }
}

impl Default for TimingProfile {
    fn default() -> Self {
        Self::V1
    }
}

/// Why a [`TimingProfile`] could not be constructed.
///
/// Non-exhaustive so validation can tighten without widening the exhaustive
/// [`ConfigError`] surface.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum TimingProfileError {
    /// The passive interval is below the 1 s load floor.
    #[error("the passive-keepalive interval is below the 1 s floor")]
    KeepaliveTooShort,
    /// The death timeout leaves no full one-loss keepalive margin.
    #[error(
        "the dead timeout is too short for two keepalives, initial RTT, and two shell-lateness allowances"
    )]
    DeadTimeoutTooShort,
    /// The relation or constructor-time monotonic deadline is not
    /// representable.
    #[error("the timing profile overflows duration or monotonic-instant arithmetic")]
    ArithmeticOverflow,
}

/// The one wall-clock reading the protocol performs (§5.3, §16.5).
///
/// Injected so a test can drive §17.2's monotonic forcing without waiting
/// on a real clock, and so an embedded host can supply whatever time source
/// it has. Nothing else in slither reads wall-clock time.
pub trait WallClock {
    /// The current wall-clock time, as §5.2's `secs ‖ nanos` pair.
    fn now(&self) -> Timestamp;
}

/// [`WallClock`] over `std::time::SystemTime`.
#[derive(Debug, Clone, Copy, Default)]
pub struct SystemClock;

impl WallClock for SystemClock {
    fn now(&self) -> Timestamp {
        // A clock before the epoch is not a protocol condition and has no
        // error path: §5.3 specifies no validation, and §17.2's forcing
        // makes even a zero reading emit a strictly-greater timestamp.
        match SystemTime::now().duration_since(UNIX_EPOCH) {
            Ok(d) => Timestamp::new(d.as_secs(), d.subsec_nanos()),
            Err(_) => Timestamp::new(0, 0),
        }
    }
}

/// An endpoint's configuration.
///
/// The two introduction-queue bounds are configurable because §6.3 says so
/// in as many words ("configurable in `Config`"). `INTRO_TTL` is **not** —
/// it remains a ratified admission timer. Ruling 282 separately admits a
/// validated [`TimingProfile`] for established-session liveness only.
#[derive(Clone)]
pub struct Config {
    intro_queue_cap: usize,
    intro_max_per_source: usize,
    epoch_size: NonZeroU64,
    stream_window: u64,
    connection_window: u64,
    timing_profile: TimingProfile,
    clock: Rc<dyn WallClock>,
}

impl fmt::Debug for Config {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Config")
            .field("intro_queue_cap", &self.intro_queue_cap)
            .field("intro_max_per_source", &self.intro_max_per_source)
            .field("epoch_size", &self.epoch_size)
            .field("stream_window", &self.stream_window)
            .field("connection_window", &self.connection_window)
            .field("timing_profile", &self.timing_profile)
            .finish_non_exhaustive()
    }
}

impl Default for Config {
    fn default() -> Self {
        Self {
            intro_queue_cap: constants::INTRO_QUEUE_CAP,
            intro_max_per_source: constants::INTRO_MAX_PER_SOURCE,
            epoch_size: Config::DEFAULT_EPOCH_SIZE,
            stream_window: Config::DEFAULT_STREAM_WINDOW,
            connection_window: Config::DEFAULT_CONNECTION_WINDOW,
            timing_profile: TimingProfile::default(),
            clock: Rc::new(SystemClock),
        }
    }
}

impl Config {
    /// §7.7's ratified epoch size: `REKEY_EPOCH_MSGS` messages per epoch.
    ///
    /// The production value, and the only one a shipped build uses.
    pub const DEFAULT_EPOCH_SIZE: NonZeroU64 = match NonZeroU64::new(constants::REKEY_EPOCH_MSGS) {
        Some(n) => n,
        None => panic!("REKEY_EPOCH_MSGS is nonzero"),
    };

    /// §10.2's ratified per-stream receive window:
    /// [`INITIAL_MAX_STREAM_DATA`](crate::constants::INITIAL_MAX_STREAM_DATA)
    /// (262 144 B).
    ///
    /// The production value, and what a shipped build advertises unless
    /// [`with_flow_windows`](Self::with_flow_windows) says otherwise.
    pub const DEFAULT_STREAM_WINDOW: u64 = constants::INITIAL_MAX_STREAM_DATA;

    /// §10.2's ratified connection-level receive window:
    /// [`INITIAL_MAX_DATA`](crate::constants::INITIAL_MAX_DATA)
    /// (1 048 576 B).
    ///
    /// The production value, and what a shipped build advertises unless
    /// [`with_flow_windows`](Self::with_flow_windows) says otherwise.
    pub const DEFAULT_CONNECTION_WINDOW: u64 = constants::INITIAL_MAX_DATA;

    /// The defaults: §6.3's ratified caps, §7.7's epoch size, §10.2's two
    /// receive windows, v1's liveness profile, and a `SystemTime` clock.
    pub fn new() -> Self {
        Self::default()
    }

    /// Override the endpoint-wide stage-0 queue cap (§6.3).
    ///
    /// The default is [`INTRO_QUEUE_CAP`](crate::constants::INTRO_QUEUE_CAP)
    /// (1024). This value is **not validated**, and two of the things it
    /// buys are load-bearing rather than advisory:
    ///
    /// - **`0` disables every inbound accept, permanently and silently.**
    ///   An arrival finds `entries.len() >= cap` with nothing evictable, so
    ///   it is dropped; no introduction is ever surfaced, `accept()` never
    ///   resolves, and there is no error, no event and no trace to say why.
    ///   Outbound `connect()` still works, which is what makes the
    ///   misconfiguration look like a peer problem.
    /// - **It is the numerator of §6.3's occupancy bound.** *"Filling the
    ///   queue needs ≥ 256 distinct sources"* is exactly this value divided
    ///   by [`with_intro_max_per_source`](Self::with_intro_max_per_source);
    ///   lowering one without the other lowers the number of sources an
    ///   attacker needs. §6.3 also prices sustained full occupancy at
    ///   ≈ 68 packets/second from `cap / INTRO_TTL`, so a larger cap costs
    ///   an attacker proportionally more bandwidth — and costs this
    ///   endpoint proportionally more memory (§17.5 bounds a parked chain at
    ///   ≈ 0.5–1 KB).
    #[must_use]
    pub fn with_intro_queue_cap(mut self, cap: usize) -> Self {
        self.intro_queue_cap = cap;
        self
    }

    /// Override the per-source chain cap (§6.3).
    ///
    /// The default is
    /// [`INTRO_MAX_PER_SOURCE`](crate::constants::INTRO_MAX_PER_SOURCE)
    /// (4). This value is **not validated**, and §6.3 calls what it buys
    /// *"the only occupant-shaped defence"* the protocol has until the
    /// deferred cookies/mac2 round (§19):
    ///
    /// - **`0` disables every inbound accept**, exactly as
    ///   [`with_intro_queue_cap(0)`](Self::with_intro_queue_cap) does and
    ///   for the same reason — the per-source check fires on the first
    ///   arrival and finds nothing to evict.
    /// - **Any value `>= intro_queue_cap` deletes the bound entirely.** The
    ///   per-source check can then never fire before the global one, so a
    ///   **single** source — one IP, one port, no spoofing capability,
    ///   since mac1's key is public data — can hold every slot in the
    ///   queue. §6.3's *"≥ 256 distinct sources"* is `cap /
    ///   max_per_source`, and at parity that number is 1.
    ///   [`constants`] pins
    ///   `INTRO_MAX_PER_SOURCE <= INTRO_QUEUE_CAP` as a compile-time
    ///   assertion for the **defaults**; nothing pins it for these two
    ///   builders.
    ///
    /// **There is a legitimate reason to raise it, which is why this note
    /// exists — and it turns on which key is which.** §6.3's *dedup* key is
    /// the full source address, so *"distinct initiators behind one NAT
    /// present distinct ports"* and each gets its own queue entry. This
    /// **cap** is keyed one level coarser: *"4 chains per source IP (per
    /// /64 for IPv6)"*. Every client behind one NAT therefore shares a
    /// single allowance of 4, and an operator serving many of them has a
    /// real reason to raise this knob. Doing so trades occupancy resistance
    /// for reachability, knowingly; raising it to or past the queue cap is
    /// not a trade, it is a removal.
    #[must_use]
    pub fn with_intro_max_per_source(mut self, cap: usize) -> Self {
        self.intro_max_per_source = cap;
        self
    }

    /// Override §7.7's epoch size — **a test-only facility**
    /// (**ruling 82**).
    ///
    /// §7.7's ratchet schedule is a pure function of the counter, so **both
    /// peers must pass the same value** or they disagree about which key
    /// opens which packet. Production uses
    /// [`REKEY_EPOCH_MSGS`](crate::constants::REKEY_EPOCH_MSGS), which is
    /// what [`Config::default`] supplies; this exists so a test can observe
    /// an epoch boundary without performing 65 536 seals to reach one.
    ///
    /// It carries §16.6's rule for the RNG seed verbatim: a build that
    /// accepts a caller-chosen epoch must be **feature-gated or documented
    /// as test-only**, and this is that documentation. Nothing in slither
    /// calls it outside tests, and a deployment that does has changed a
    /// ratified security parameter.
    #[must_use]
    pub fn with_epoch_size(mut self, epoch_size: NonZeroU64) -> Self {
        self.epoch_size = epoch_size;
        self
    }

    /// Raise §10.2's two advertised receive windows — **the knob raises,
    /// never lowers** (**ruling 259(viii)**).
    ///
    /// The defaults are [`DEFAULT_STREAM_WINDOW`](Self::DEFAULT_STREAM_WINDOW)
    /// (256 KiB) and
    /// [`DEFAULT_CONNECTION_WINDOW`](Self::DEFAULT_CONNECTION_WINDOW)
    /// (1 MiB), which are §10.2's ratified constants and what a `Config`
    /// nobody configured advertises. What they cost, measured (ruling 269 —
    /// this doc once credited the figure to ruling 247(a) as *measured*; it
    /// was derived, and derived with the wrong rule): one stream is capped
    /// at ≈ `stream_window / (2 × RTT)` — §10.3 re-grants at half the
    /// window consumed — about 1.4 MiB/s at 100 ms, and before this knob a
    /// consumer had no way to buy more.
    ///
    /// # What changes, and what does not
    ///
    /// **Nothing on the wire changes shape.** §10.2's initial windows are
    /// protocol constants that are never sent; MAX_DATA and
    /// MAX_STREAM_DATA already carry whatever the receiver has decided to
    /// advertise, so a raised window is announced through the frames §10.3
    /// already emits, at values §8.4 already admits. The ratified
    /// constants keep their values and stay the defaults.
    ///
    /// Three things this deliberately does **not** touch:
    ///
    /// - **The peer's limits.** A connection still assumes the *peer*
    ///   advertises §10.2's constants until a credit frame says otherwise.
    ///   The knob is one endpoint's receive policy; it is not negotiated
    ///   and the peer needs no matching build.
    /// - **[`MESSAGE_RECV_MAX`](crate::constants::MESSAGE_RECV_MAX).**
    ///   §9.8's message bound is a *cross-peer* contract checked on the
    ///   **send** side, and a sender cannot know what its receiver
    ///   configured. Raising it locally would make `send_message` emit a
    ///   payload that a default peer resets with `MESSAGE_OVERFLOW`. It
    ///   stays equal to
    ///   [`INITIAL_MAX_STREAM_DATA`](crate::constants::INITIAL_MAX_STREAM_DATA),
    ///   which is what `constants.rs` asserts and §9.8's table states.
    ///   Messages stay bounded at 256 KiB however wide this endpoint's
    ///   streams are; larger transfers use real streams, exactly as §9.8
    ///   says.
    /// - **§17.5's memory ceiling shape.** The receive commitment per
    ///   connection is the *connection* window, so raising it is a
    ///   deliberate purchase of memory: `connection_window` bytes per live
    ///   connection, plus reassembly metadata. Raise the pair, not one of
    ///   them, and size it against how many connections this endpoint
    ///   expects.
    ///
    /// # Errors
    ///
    /// - [`ConfigError::WindowTooSmall`] if either value is below its ratified
    ///   default. Lowering re-opens every sizing proof that rests on the
    ///   constants, so it is refused rather than clamped.
    /// - [`ConfigError::WindowTooLarge`] if either value exceeds `2^62 - 1`, the
    ///   largest offset a §8.1 varint carries.
    /// - [`ConfigError::StreamWindowAboveConnection`] if `stream` exceeds
    ///   `connection` — the relation `constants.rs` pins for the defaults.
    ///
    /// # Example
    ///
    /// ```
    /// use slither::prelude::*;
    ///
    /// // 2 MiB per stream, 8 MiB per connection: ~9 MiB/s measured on one
    /// // stream at a 100 ms RTT (ruling 269). Size ≈ 2 × RTT × target
    /// // rate, and not larger — past the optimum a bigger window buys
    /// // ack-path work, not throughput.
    /// let config = Config::new()
    ///     .with_flow_windows(2 * 1024 * 1024, 8 * 1024 * 1024)
    ///     .expect("a raise within the varint bound");
    /// assert_eq!(config.stream_window(), 2 * 1024 * 1024);
    /// ```
    pub fn with_flow_windows(mut self, stream: u64, connection: u64) -> Result<Self, ConfigError> {
        if stream < Self::DEFAULT_STREAM_WINDOW || connection < Self::DEFAULT_CONNECTION_WINDOW {
            return Err(ConfigError::WindowTooSmall);
        }
        if stream > crate::varint::VarInt::MAX_VALUE
            || connection > crate::varint::VarInt::MAX_VALUE
        {
            return Err(ConfigError::WindowTooLarge);
        }
        if stream > connection {
            return Err(ConfigError::StreamWindowAboveConnection);
        }
        self.stream_window = stream;
        self.connection_window = connection;
        Ok(self)
    }

    /// Set the endpoint-wide established-session liveness profile.
    ///
    /// The profile has already been validated by [`TimingProfile::try_new`],
    /// so installing it is infallible. It is copied into every connection
    /// born through either `connect()` or `accept()`.
    #[must_use]
    pub fn with_timing_profile(mut self, timing_profile: TimingProfile) -> Self {
        self.timing_profile = timing_profile;
        self
    }

    /// Replace the wall-clock service (§16.5).
    #[must_use]
    pub fn with_clock(mut self, clock: Rc<dyn WallClock>) -> Self {
        self.clock = clock;
        self
    }

    /// The endpoint-wide stage-0 queue cap.
    pub fn intro_queue_cap(&self) -> usize {
        self.intro_queue_cap
    }

    /// The per-source chain cap.
    pub fn intro_max_per_source(&self) -> usize {
        self.intro_max_per_source
    }

    /// §7.7's epoch size, for hiss's ratcheting datagram split.
    pub fn epoch_size(&self) -> NonZeroU64 {
        self.epoch_size
    }

    /// §10.2's per-stream receive window, as this endpoint advertises it.
    pub fn stream_window(&self) -> u64 {
        self.stream_window
    }

    /// §10.2's connection-level receive window, as this endpoint
    /// advertises it.
    pub fn connection_window(&self) -> u64 {
        self.connection_window
    }

    /// The endpoint-wide established-session liveness profile.
    pub fn timing_profile(&self) -> TimingProfile {
        self.timing_profile
    }

    /// The injected wall clock.
    pub fn clock(&self) -> &dyn WallClock {
        &*self.clock
    }
}

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

    /// Ruling 82 splits the boundary behaviour from the constant, so the
    /// constant needs its own pin: a `Config` nobody configured ratchets on
    /// §7.7's ratified schedule.
    #[test]
    fn the_default_epoch_size_is_rekey_epoch_msgs() {
        assert_eq!(
            Config::new().epoch_size().get(),
            constants::REKEY_EPOCH_MSGS
        );
        assert_eq!(
            Config::DEFAULT_EPOCH_SIZE.get(),
            constants::REKEY_EPOCH_MSGS
        );
    }

    #[test]
    fn the_epoch_size_override_takes_effect() {
        let config = Config::new().with_epoch_size(NonZeroU64::new(8).unwrap());
        assert_eq!(config.epoch_size().get(), 8);
    }

    /// **[ruling 259(viii)]** The knob's whole premise: the ratified
    /// constants stay the defaults. A `Config` nobody configured advertises
    /// §10.2's two values and nothing else.
    #[test]
    fn the_default_windows_are_the_ratified_constants() {
        let config = Config::new();
        assert_eq!(config.stream_window(), constants::INITIAL_MAX_STREAM_DATA);
        assert_eq!(config.connection_window(), constants::INITIAL_MAX_DATA);
        assert_eq!(
            Config::DEFAULT_STREAM_WINDOW,
            constants::INITIAL_MAX_STREAM_DATA
        );
        assert_eq!(
            Config::DEFAULT_CONNECTION_WINDOW,
            constants::INITIAL_MAX_DATA
        );
    }

    #[test]
    fn the_window_override_takes_effect() {
        let config = Config::new()
            .with_flow_windows(1 << 20, 1 << 23)
            .expect("a raise");
        assert_eq!(config.stream_window(), 1 << 20);
        assert_eq!(config.connection_window(), 1 << 23);
    }

    /// The knob raises, never lowers — on **either** value independently,
    /// and at the boundary the defaults themselves are accepted (a no-op
    /// raise), so the refusal is `< default`, not `<= default`.
    #[test]
    fn a_window_below_its_ratified_default_is_refused() {
        assert_eq!(
            Config::new()
                .with_flow_windows(
                    constants::INITIAL_MAX_STREAM_DATA - 1,
                    constants::INITIAL_MAX_DATA,
                )
                .unwrap_err(),
            ConfigError::WindowTooSmall,
        );
        assert_eq!(
            Config::new()
                .with_flow_windows(
                    constants::INITIAL_MAX_STREAM_DATA,
                    constants::INITIAL_MAX_DATA - 1,
                )
                .unwrap_err(),
            ConfigError::WindowTooSmall,
        );
        let at_the_defaults = Config::new()
            .with_flow_windows(
                constants::INITIAL_MAX_STREAM_DATA,
                constants::INITIAL_MAX_DATA,
            )
            .expect("the defaults restated are a legal no-op raise");
        assert_eq!(
            at_the_defaults.stream_window(),
            constants::INITIAL_MAX_STREAM_DATA
        );
    }

    /// The §8.1 varint ceiling, from both sides of the boundary: the
    /// largest encodable offset is accepted and one more is refused.
    #[test]
    fn a_window_past_the_varint_bound_is_refused() {
        let max = crate::varint::VarInt::MAX_VALUE;
        assert_eq!(
            Config::new()
                .with_flow_windows(max, max)
                .unwrap()
                .stream_window(),
            max,
            "the largest encodable absolute offset is admissible",
        );
        assert_eq!(
            Config::new()
                .with_flow_windows(max + 1, max + 1)
                .unwrap_err(),
            ConfigError::WindowTooLarge,
        );
        assert_eq!(
            Config::new().with_flow_windows(max, max + 1).unwrap_err(),
            ConfigError::WindowTooLarge,
            "the connection window is checked on its own, not only via the pair",
        );
    }

    #[test]
    fn the_stream_window_may_not_exceed_the_connection_window() {
        assert_eq!(
            Config::new()
                .with_flow_windows(1 << 23, 1 << 20)
                .unwrap_err(),
            ConfigError::StreamWindowAboveConnection,
        );
    }
}