liminal-rs 0.10.1

//! A1 admission: the per-subscriber pressure decision and the fan-out
//! aggregation that turns it into ONE producer-visible signal.
//!
//! Spec: `docs/design/A1-DEFER-SEMANTICS.md` §2 (decision attachment), §3 (the
//! Defer lifecycle and the aggregation table), §4 (durable channels: the
//! pre-append hard watermark and Defer-not-Reject after append).
//!
//! This module exists because the design's graft §0.5 says so: `channel/types.rs`
//! was already over the 500-LOC budget before A1, so the decision and
//! aggregation vocabulary lands here rather than growing it.
//!
//! Nothing in this file touches a queue. The per-subscriber decision is taken
//! inside [`crate::channel::subscription::SubscriptionInbox::admit`], under the
//! one mutex that also owns the queue, so decide+push is atomic against a
//! concurrent pop; this module supplies the *rules* that decision and its
//! aggregation follow.
//!
//! All arithmetic here is exact integer arithmetic. The buffer-band fill is
//! carried as a rational `(filled, bound)` and compared by cross-multiplication
//! in `u128`, so "worst delay hint" is a total order with no floating-point
//! rounding anywhere on the publish path.

use crate::pressure::{ConsumerCapacity, PressureSignal};

use std::time::Duration;

/// Default in-flight window for an inbox whose subscriber declared no capacity
/// (A1 §2). Every inbox is bounded — there is no opt-out — so this is the
/// value plain `subscribe()` gets.
pub const DEFAULT_MAX_IN_FLIGHT: usize = 128;

/// Default buffer band for an inbox whose subscriber declared no capacity
/// (A1 §2). `max_buffer_depth` is bus policy, not a client declaration.
pub const DEFAULT_MAX_BUFFER_DEPTH: usize = 1_024;

/// The capacity every inbox starts with (A1 §2: "plain `subscribe()` gets
/// defaults ... so **every** inbox is bounded — no opt-out").
#[must_use]
pub const fn default_capacity() -> ConsumerCapacity {
    ConsumerCapacity {
        max_in_flight: DEFAULT_MAX_IN_FLIGHT,
        max_buffer_depth: DEFAULT_MAX_BUFFER_DEPTH,
    }
}

/// Per-channel pressure policy (A1 §3 delay hint, §4 hard watermark).
///
/// Carried on [`crate::channel::ChannelConfig`] so it is per-channel exactly as
/// the design says, and defaulted so no existing construction site changes.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ChannelPressureConfig {
    /// Delay hint floor in milliseconds, emitted at a just-full in-flight
    /// window with an empty buffer band (A1 §3, default 25ms).
    pub defer_delay_base_ms: u64,
    /// Delay hint ceiling in milliseconds, emitted at a full buffer band
    /// (A1 §3, default 250ms).
    pub defer_delay_max_ms: u64,
    /// Pre-append hard watermark for durable channels, as a percentage of the
    /// summed live-buffer bound of every registered subscriber (A1 §4, graft
    /// §0.3, default 100). At or above it a durable publish is Rejected
    /// BEFORE anything is appended, which is the only honest Reject a durable
    /// channel can emit.
    pub durable_reject_watermark_percent: u8,
}

impl ChannelPressureConfig {
    /// The design's defaults: 25ms/250ms hints, watermark at 100%.
    pub const DEFAULT: Self = Self {
        defer_delay_base_ms: 25,
        defer_delay_max_ms: 250,
        durable_reject_watermark_percent: 100,
    };
}

impl Default for ChannelPressureConfig {
    fn default() -> Self {
        Self::DEFAULT
    }
}

/// How full the buffer band was when a signal was taken, as the exact rational
/// `(filled, bound)` with `bound > 0` and `filled <= bound`.
///
/// Accept is `0/1` (the in-flight window still had credit, so no buffer was
/// touched); Reject is `1/1` (the buffer band is full by definition — that is
/// what makes it a Reject). Defer interpolates. This single ordering is what
/// "worst delay hint" in §3 means, so aggregation and the hint read the same
/// number.
const fn buffer_fill(signal: &PressureSignal) -> (u128, u128) {
    match *signal {
        PressureSignal::Accept { .. } => (0, 1),
        PressureSignal::Defer {
            current_buffer_depth,
            max_buffer_depth,
            ..
        } => {
            if max_buffer_depth == 0 || current_buffer_depth >= max_buffer_depth {
                (1, 1)
            } else {
                (current_buffer_depth as u128, max_buffer_depth as u128)
            }
        }
        PressureSignal::Reject { .. } => (1, 1),
    }
}

/// Whether `candidate` is at least as full as `held` — the "worst" comparison,
/// cross-multiplied so no division or float rounding is involved.
///
/// Both numerators and both denominators are `usize`-derived and therefore below
/// `2^64`, so each product is below `2^128` and cannot overflow.
const fn fill_at_least(candidate: &PressureSignal, held: &PressureSignal) -> bool {
    let (candidate_filled, candidate_bound) = buffer_fill(candidate);
    let (held_filled, held_bound) = buffer_fill(held);
    candidate_filled * held_bound >= held_filled * candidate_bound
}

/// The producer's advisory delay hint for a signal (A1 §3):
/// `base + (max − base) × buffer_fill_fraction`.
///
/// `None` for Accept — an accepted publish is not paced. Reject carries a hint
/// too: on a durable channel a post-append Reject is reported as Defer
/// ([`defer_after_append`]) and the producer needs the ceiling hint that goes
/// with a full buffer band.
#[must_use]
pub fn defer_delay(signal: &PressureSignal, config: ChannelPressureConfig) -> Option<Duration> {
    if matches!(*signal, PressureSignal::Accept { .. }) {
        return None;
    }
    let base = config.defer_delay_base_ms;
    let ceiling = config.defer_delay_max_ms.max(base);
    let span = ceiling.saturating_sub(base);
    let (filled, bound) = buffer_fill(signal);
    // `filled <= bound` and `bound > 0` by construction, so the quotient is at
    // most `span` and the conversion back cannot fail; the fallback is the
    // ceiling of that same range and so is honest even if it ever did.
    let scaled = u128::from(span) * filled / bound;
    let added = u64::try_from(scaled).unwrap_or(span);
    Some(Duration::from_millis(base.saturating_add(added)))
}

/// Rewrites a post-append Reject as a Defer with identical bands (A1 §4).
///
/// "Reject means shed, and an appended message is not shed — it sits at its
/// sequence position and every replay consumer sees it exactly once regardless
/// of what the live buffers did. Emitting Reject after a successful append
/// would be a lie."
#[must_use]
pub const fn defer_after_append(signal: PressureSignal) -> PressureSignal {
    match signal {
        PressureSignal::Reject {
            current_in_flight,
            max_in_flight,
            current_buffer_depth,
            max_buffer_depth,
        } => PressureSignal::defer(
            current_in_flight,
            max_in_flight,
            current_buffer_depth,
            max_buffer_depth,
        ),
        other => other,
    }
}

/// Clamps a declared consumer capacity so the A1 bands decide BEFORE the §5
/// depth cap — the "A1 bites first" invariant, enforced instead of asserted.
///
/// **The invariant, derived rather than asserted.** `SubscriptionInbox::admit`
/// takes the A1 decision first and refuses at `queued >= max_in_flight +
/// max_buffer_depth`; the §5 fairness trip is checked second and refuses at
/// `queued >= depth_cap`. So the A1 band decides at every queue length exactly
/// when `max_in_flight + max_buffer_depth <= depth_cap`. That matters because
/// the two refusals are not interchangeable: an A1 Reject paces one message,
/// while a §5 trip sets the sticky overflow marker and the whole subscription
/// is shed. A window large enough to put the A1 Reject band above the depth cap
/// turns "your consumer is behind" into "your subscription is gone".
///
/// **Why the client's window yields and the bus's band does not.** §2 makes
/// `max_in_flight` a client declaration and `max_buffer_depth` bus policy, and
/// the depth cap is server config. Policy therefore wins: the declared window
/// shrinks first, and the buffer band only gives way when the cap is smaller
/// than the policy band itself. Nothing here is hardcoded — every number comes
/// from the two arguments, so moving `depth_cap` or the default buffer band
/// moves the clamp with it.
///
/// **Degenerate caps (`depth_cap < 2`).** A legal [`ConsumerCapacity`] has two
/// positive bands, so the smallest legal sum is 2 and no capacity at all fits
/// under a cap of 0 or 1: the invariant is UNSATISFIABLE there, not violated.
/// The minimal legal capacity `(1, 1)` is returned and the §5 fairness trip
/// owns the shed. Neither band is ever zero, because a capacity with a zero
/// band is one [`ConsumerCapacity::validate`] refuses — and the output of this
/// function is what gets INSTALLED, so an invalid one is not merely useless,
/// it is silently discarded and the inbox keeps its defaults. `depth_cap == 0`
/// is unreachable from a validated server config; `depth_cap == 1` is
/// reachable and pathological.
#[must_use]
pub const fn clamp_capacity_to_depth_cap(
    declared: ConsumerCapacity,
    depth_cap: usize,
) -> ConsumerCapacity {
    let headroom = depth_cap.saturating_sub(declared.max_buffer_depth);
    let max_in_flight = if declared.max_in_flight < headroom {
        declared.max_in_flight
    } else {
        headroom
    };
    // Never zero: a zero window is not a legal declaration and would refuse
    // every publish outright.
    let max_in_flight = if max_in_flight == 0 { 1 } else { max_in_flight };
    let remaining = depth_cap.saturating_sub(max_in_flight);
    let max_buffer_depth = if declared.max_buffer_depth < remaining {
        declared.max_buffer_depth
    } else {
        remaining
    };
    // Never zero, for the same reason `max_in_flight` is never zero: this
    // capacity is INSTALLED, and `install_capacity` refuses an invalid one, so
    // emitting `(1, 0)` here does not produce a tight window — it produces no
    // window at all and leaves the inbox on the defaults this function exists
    // to clamp. Only reachable when `depth_cap < 2`, where nothing legal fits.
    let max_buffer_depth = if max_buffer_depth == 0 {
        1
    } else {
        max_buffer_depth
    };
    ConsumerCapacity {
        max_in_flight,
        max_buffer_depth,
    }
}

/// Whether a durable publish must be Rejected BEFORE anything is appended
/// (A1 §4, graft §0.3).
///
/// `total_queued` and `total_bound` are the channel-aggregate live-buffer
/// occupancy and its summed bound: coarse on purpose (no predicate is
/// evaluated, so this may refuse a message whose target subscriber is fast —
/// the design documents that imprecision and defers the per-cursor bound to
/// v2).
///
/// A channel with no registered subscribers has `total_bound == 0` and can
/// never reach the watermark: there is no live buffer to fill, so there is
/// nothing to bound. That guard is load-bearing — without it a `0 >= 0`
/// comparison would refuse every publish to an unsubscribed durable channel.
#[must_use]
pub const fn watermark_reached(total_queued: usize, total_bound: usize, percent: u8) -> bool {
    if total_bound == 0 {
        return false;
    }
    // Cross-multiplied in `u128` so the comparison is exact and cannot overflow
    // for any `usize` pair: `queued * 100 >= bound * percent`.
    let queued = (total_queued as u128) * 100;
    let threshold = (total_bound as u128) * (percent as u128);
    queued >= threshold
}

/// Folds per-subscriber outcomes into the one producer-visible signal (A1 §3).
///
/// | Matching-subscriber outcomes | Producer signal |
/// |---|---|
/// | all Accept (or zero matching subscribers) | **Accept** |
/// | any Defer, or mixed Accept/Reject | **Defer** (worst delay hint) |
/// | all Reject | **Reject** |
///
/// "The rule keeps `Reject ⇒ delivered to nobody` exact, which §5 depends on."
/// Only subscribers whose predicate MATCHED are offered here — a non-matching
/// subscriber never contributes false backpressure (§2).
#[derive(Clone, Debug, Default)]
pub struct PressureAggregate {
    matching: usize,
    accepted: usize,
    rejected: usize,
    /// Of the rejected, how many were dropped by the §5 memory-safety path
    /// rather than by an A1 pacing verdict. Counted separately so the two
    /// zero-delivery worlds stay distinguishable — see [`Self::resolve`].
    dropped: usize,
    /// Worst Accept seen, by in-flight occupancy: reported when every matching
    /// subscriber accepted, so the producer still learns how close the busiest
    /// consumer is to its window.
    worst_accept: Option<PressureSignal>,
    /// Worst non-Accept seen, by [`buffer_fill`]. Carries the hint.
    worst_pressured: Option<PressureSignal>,
}

impl PressureAggregate {
    /// A fresh aggregate with no subscriber outcomes recorded.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            matching: 0,
            accepted: 0,
            rejected: 0,
            dropped: 0,
            worst_accept: None,
            worst_pressured: None,
        }
    }

    /// Records one matching subscriber's decision.
    pub fn record(&mut self, signal: PressureSignal) {
        self.matching += 1;
        match signal {
            PressureSignal::Accept {
                current_in_flight, ..
            } => {
                self.accepted += 1;
                let replace = self
                    .worst_accept
                    .as_ref()
                    .is_none_or(|held| accept_in_flight(held) <= current_in_flight);
                if replace {
                    self.worst_accept = Some(signal);
                }
            }
            PressureSignal::Defer { .. } | PressureSignal::Reject { .. } => {
                if matches!(signal, PressureSignal::Reject { .. }) {
                    self.rejected += 1;
                }
                let replace = self
                    .worst_pressured
                    .as_ref()
                    .is_none_or(|held| fill_at_least(&signal, held));
                if replace {
                    self.worst_pressured = Some(signal);
                }
            }
        }
    }

    /// Records one matching subscriber that DROPPED the envelope through the
    /// §5 memory-safety path — the shared connection byte budget or the
    /// per-inbox fairness trip.
    ///
    /// A distinct entry point, and a distinct count, because the two worlds
    /// that produce zero deliveries must stay distinguishable: **nobody
    /// matched** (no consumer window was ever consulted — Accept, the
    /// sentinel) versus **everybody dropped it** (every consumer refused —
    /// Reject, delivered to nobody). Before this existed the §5 refusals were
    /// invisible here, so the second world resolved as the first and
    /// `ChannelDelivery::is_admitted()` claimed the bus had taken custody of a
    /// message it had thrown away.
    ///
    /// `signal` is a [`PressureSignal::Reject`] by construction (it comes from
    /// `InboxAdmission::shed_signal`), so folding it in through
    /// [`Self::record`] gives exactly §3's table: all-dropped ⇒ Reject, and a
    /// dropped sibling alongside an accepting one ⇒ Defer at the ceiling hint,
    /// which is what a shed sibling has always meant.
    pub fn record_dropped(&mut self, signal: PressureSignal) {
        self.dropped += 1;
        self.record(signal);
    }

    /// The number of matching subscribers whose decision was recorded.
    #[must_use]
    pub const fn matching(&self) -> usize {
        self.matching
    }

    /// How many matching subscribers dropped the envelope through the §5
    /// memory-safety path.
    #[must_use]
    pub const fn dropped(&self) -> usize {
        self.dropped
    }

    /// Whether EVERY matching subscriber dropped the envelope through the §5
    /// memory-safety path — the connection byte budget or the per-inbox fairness
    /// trip (A1 round 3).
    ///
    /// False when nothing matched, and false the moment ONE matching subscriber
    /// answered through any other door: an A1 §4 pacing shed beside a §5 drop
    /// still leaves a consumer that will get the message, because the pacing
    /// shed arms auto-catch-up.
    ///
    /// This is the ONE fact `ChannelHandle::publish_with_delivery` needs to tell
    /// a recoverable durable shed from an unrecoverable one, and it is derived
    /// from the two counts rather than tracked as a third: `dropped == matching`
    /// implies `rejected == matching` (every `record_dropped` folds a `Reject`
    /// in through [`Self::record`]), so this being true also means
    /// [`Self::resolve`] answered `Reject`.
    #[must_use]
    pub const fn every_match_dropped(&self) -> bool {
        self.matching > 0 && self.dropped == self.matching
    }

    /// The aggregated producer-visible signal, per the table above.
    ///
    /// With zero matching subscribers the answer is `Accept { 0, 0 }`: no
    /// consumer window was consulted, and a zero `max_in_flight` is not a legal
    /// [`ConsumerCapacity`], so it cannot be mistaken for a real one.
    ///
    /// That sentinel is reachable ONLY when nothing matched. A subscriber that
    /// matched and then dropped the envelope on the §5 path is recorded by
    /// [`Self::record_dropped`] and counts toward `matching`, so a publish
    /// every subscriber dropped resolves to Reject — delivered to nobody —
    /// and never borrows the empty channel's answer.
    #[must_use]
    pub fn resolve(&self) -> PressureSignal {
        if self.matching == 0 {
            return PressureSignal::accept(0, 0);
        }
        if self.rejected == self.matching {
            // All Reject. `worst_pressured` is a Reject here by construction:
            // every recorded signal was one.
            return self
                .worst_pressured
                .unwrap_or_else(|| PressureSignal::accept(0, 0));
        }
        if self.accepted == self.matching {
            return self
                .worst_accept
                .unwrap_or_else(|| PressureSignal::accept(0, 0));
        }
        // Any Defer, or mixed Accept/Reject: Defer at the worst hint. A Reject
        // in the mix contributes its full-band counts, so the producer is told
        // to pace at the ceiling — which is exactly what a shed sibling means.
        self.worst_pressured
            .map_or_else(|| PressureSignal::accept(0, 0), defer_after_append)
    }
}

/// The in-flight occupancy an Accept was taken at; `0` for any other variant
/// (never reached — this is only called on the Accept arm).
const fn accept_in_flight(signal: &PressureSignal) -> usize {
    match *signal {
        PressureSignal::Accept {
            current_in_flight, ..
        }
        | PressureSignal::Defer {
            current_in_flight, ..
        }
        | PressureSignal::Reject {
            current_in_flight, ..
        } => current_in_flight,
    }
}

#[cfg(test)]
mod tests;