liminal-rs 0.12.0

//! A1 admission unit pins: the §3 aggregation table, the §3 delay hint, and
//! the §4 pre-append watermark predicate.

use std::time::Duration;

use super::{
    ChannelPressureConfig, DEFAULT_MAX_BUFFER_DEPTH, PressureAggregate,
    clamp_capacity_to_depth_cap, default_capacity, defer_after_append, defer_delay,
    watermark_reached,
};
use crate::pressure::{ConsumerCapacity, PressureSignal};

const CONFIG: ChannelPressureConfig = ChannelPressureConfig::DEFAULT;

fn accept() -> PressureSignal {
    PressureSignal::accept(3, 4)
}

fn defer(buffered: usize) -> PressureSignal {
    PressureSignal::defer(4, 4, buffered, 16)
}

fn reject() -> PressureSignal {
    PressureSignal::reject(4, 4, 16, 16)
}

/// §3, row 1: "all Accept (or zero matching subscribers) | **Accept**".
#[test]
fn zero_matching_subscribers_aggregates_to_accept() {
    let aggregate = PressureAggregate::new();
    assert_eq!(aggregate.matching(), 0);
    assert_eq!(aggregate.resolve(), PressureSignal::accept(0, 0));
}

#[test]
fn all_accept_aggregates_to_the_busiest_accept() {
    let mut aggregate = PressureAggregate::new();
    aggregate.record(PressureSignal::accept(1, 4));
    aggregate.record(PressureSignal::accept(3, 4));
    aggregate.record(PressureSignal::accept(2, 4));
    assert_eq!(
        aggregate.resolve(),
        PressureSignal::accept(3, 4),
        "every subscriber accepted, and the producer is told how close the busiest one is"
    );
}

/// §3, row 2: "any Defer ... | **Defer** (worst delay hint)".
#[test]
fn any_defer_aggregates_to_the_worst_defer() {
    let mut aggregate = PressureAggregate::new();
    aggregate.record(accept());
    aggregate.record(defer(2));
    aggregate.record(defer(11));
    aggregate.record(defer(5));
    assert_eq!(
        aggregate.resolve(),
        defer(11),
        "the fullest buffer band sets the hint the producer paces against"
    );
}

/// §3, row 2 again: "mixed Accept/Reject | **Defer**". This is the row that
/// keeps `Reject => delivered to nobody` exact — a message one subscriber took
/// must never be reported as shed.
#[test]
fn mixed_accept_and_reject_aggregates_to_defer_at_the_ceiling() {
    let mut aggregate = PressureAggregate::new();
    aggregate.record(accept());
    aggregate.record(reject());
    let resolved = aggregate.resolve();
    assert_eq!(
        resolved,
        PressureSignal::defer(4, 4, 16, 16),
        "one subscriber received it, so the producer is paced, not told to retry"
    );
    assert_eq!(
        defer_delay(&resolved, CONFIG),
        Some(Duration::from_millis(250)),
        "a shed sibling paces at the ceiling"
    );
}

/// §3, row 3: "all Reject | **Reject**".
#[test]
fn all_reject_aggregates_to_reject() {
    let mut aggregate = PressureAggregate::new();
    aggregate.record(reject());
    aggregate.record(reject());
    assert_eq!(aggregate.resolve(), reject());
}

/// §3: `delay = base + (max − base) × buffer_fill_fraction`, defaults 25/250.
#[test]
fn delay_hint_interpolates_across_the_buffer_band() {
    assert_eq!(defer_delay(&accept(), CONFIG), None, "Accept is not paced");
    assert_eq!(
        defer_delay(&defer(0), CONFIG),
        Some(Duration::from_millis(25)),
        "an empty buffer band hints the floor"
    );
    assert_eq!(
        defer_delay(&defer(8), CONFIG),
        Some(Duration::from_millis(25 + 225 / 2)),
        "a half-full band hints the midpoint"
    );
    assert_eq!(
        defer_delay(&defer(16), CONFIG),
        Some(Duration::from_millis(250)),
        "a full band hints the ceiling"
    );
    assert_eq!(
        defer_delay(&reject(), CONFIG),
        Some(Duration::from_millis(250)),
        "Reject is band-full by definition"
    );
}

/// §3: "escalation to Reject is structural, never time-based ... a resuming
/// consumer de-escalates the same way". The hint is a pure function of the
/// bands, so it is non-decreasing as the band fills and falls back as it drains.
#[test]
fn delay_hint_is_monotonic_in_buffer_fill() {
    let mut previous = Duration::ZERO;
    for buffered in 0..=16 {
        let signal = defer(buffered);
        assert!(
            defer_delay(&signal, CONFIG).is_some(),
            "a Defer always carries a hint"
        );
        let hint = defer_delay(&signal, CONFIG).unwrap_or_default();
        assert!(
            hint >= previous,
            "hint must not fall as the buffer band fills: {buffered} gave {hint:?} after {previous:?}"
        );
        previous = hint;
    }
}

/// §4: the only rewrite this direction. Accept and Defer pass through
/// untouched — a durable append does not make an accepted message deferred.
#[test]
fn defer_after_append_rewrites_only_reject() {
    assert_eq!(defer_after_append(accept()), accept());
    assert_eq!(defer_after_append(defer(3)), defer(3));
    assert_eq!(
        defer_after_append(reject()),
        PressureSignal::defer(4, 4, 16, 16),
        "an appended message is not shed, so Reject would be a lie"
    );
}

/// §4: the watermark is `queued >= bound × percent / 100`, and a channel with
/// no subscribers can never reach it.
#[test]
fn watermark_predicate_boundaries() {
    assert!(
        !watermark_reached(0, 0, 100),
        "no subscribers means no live buffer to fill, so nothing to bound"
    );
    assert!(
        !watermark_reached(9_999, 0, 100),
        "an unsubscribed durable channel is never watermark-rejected"
    );
    assert!(!watermark_reached(99, 100, 100), "below the watermark");
    assert!(watermark_reached(100, 100, 100), "at the watermark");
    assert!(watermark_reached(101, 100, 100), "above the watermark");
    assert!(!watermark_reached(49, 100, 50), "below a 50% watermark");
    assert!(watermark_reached(50, 100, 50), "at a 50% watermark");
}

/// An outcome that was never recorded at all — a non-matching subscriber, or a
/// closed inbox — cannot manufacture pressure.
#[test]
fn only_recorded_subscribers_count_toward_the_aggregate() {
    let mut aggregate = PressureAggregate::new();
    aggregate.record(accept());
    assert_eq!(aggregate.matching(), 1);
    assert_eq!(aggregate.dropped(), 0);
    assert_eq!(
        aggregate.resolve(),
        accept(),
        "a subscriber whose outcome was never recorded cannot manufacture pressure"
    );
}

/// §7 / verifier finding 2: **nobody matched is not everybody dropped.**
///
/// A §5 budget/fairness shed is recorded through `record_dropped`, which counts
/// it toward `matching`. So a publish every subscriber dropped resolves to
/// Reject — delivered to nobody — instead of borrowing the empty channel's
/// `Accept { 0, 0 }` sentinel and telling the producer the bus took custody.
#[test]
fn everybody_dropped_resolves_to_reject_not_the_empty_channel_sentinel() {
    let mut everybody = PressureAggregate::new();
    everybody.record_dropped(reject());
    everybody.record_dropped(reject());
    assert_eq!(everybody.matching(), 2);
    assert_eq!(everybody.dropped(), 2);
    assert_eq!(
        everybody.resolve(),
        reject(),
        "every matching subscriber dropped it: delivered to nobody"
    );

    let nobody = PressureAggregate::new();
    assert_eq!(
        nobody.resolve(),
        PressureSignal::accept(0, 0),
        "and the zero-subscriber sentinel is still reachable, from zero \
         subscribers only"
    );
}

/// A dropped sibling alongside an accepting one is §3's "mixed Accept/Reject":
/// Defer at the ceiling hint. At least one subscriber took custody, so the
/// publish IS admitted.
#[test]
fn a_dropped_sibling_alongside_an_accepting_one_defers() {
    let mut aggregate = PressureAggregate::new();
    aggregate.record(accept());
    aggregate.record_dropped(reject());
    assert_eq!(aggregate.matching(), 2);
    assert_eq!(aggregate.dropped(), 1);
    assert!(
        matches!(aggregate.resolve(), PressureSignal::Defer { .. }),
        "mixed accept/reject reports Defer, as it always has — got {:?}",
        aggregate.resolve()
    );
}

/// `every_match_dropped` is the fact the durable publish path reads to tell a
/// RECOVERABLE post-append shed from an unrecoverable one (round 3). Every
/// discriminating case, in one place: it must be false with nothing matching,
/// false when any matching subscriber answered through a door other than §5, and
/// true only when §5 took all of them.
#[test]
fn every_match_dropped_is_true_only_when_the_section_five_door_took_all_of_them() {
    let nobody = PressureAggregate::new();
    assert!(
        !nobody.every_match_dropped(),
        "zero matching subscribers is the empty-channel sentinel, not a drop"
    );

    let mut all_dropped = PressureAggregate::new();
    all_dropped.record_dropped(reject());
    all_dropped.record_dropped(reject());
    assert!(all_dropped.every_match_dropped());
    assert_eq!(
        all_dropped.resolve(),
        reject(),
        "and it implies the resolved verdict is Reject, which is what the \
         durable path's suppression relies on"
    );

    let mut with_accept = PressureAggregate::new();
    with_accept.record(accept());
    with_accept.record_dropped(reject());
    assert!(
        !with_accept.every_match_dropped(),
        "a subscriber took custody, so the message still reaches somebody"
    );

    // The load-bearing one: an A1 §4 PACING reject beside a §5 drop. Both are
    // Rejects and `resolve()` answers Reject for the pair, but the pacing shed
    // ARMS auto-catch-up, so the message is still on its way to that subscriber
    // and the durable Defer rewrite must stand.
    let mut with_pacing_reject = PressureAggregate::new();
    with_pacing_reject.record(reject());
    with_pacing_reject.record_dropped(reject());
    assert_eq!(with_pacing_reject.matching(), 2);
    assert_eq!(with_pacing_reject.dropped(), 1);
    assert!(
        !with_pacing_reject.every_match_dropped(),
        "an A1 pacing shed is recoverable: it is not a §5 drop and must not be \
         counted as one"
    );

    let mut with_defer = PressureAggregate::new();
    with_defer.record(defer(1));
    with_defer.record_dropped(reject());
    assert!(!with_defer.every_match_dropped());
}

// ---------------------------------------------------------------------------
// §5-vs-A1 ordering: the "A1 bites first" invariant, as a CONTROL
// ---------------------------------------------------------------------------

/// The invariant `clamp_capacity_to_depth_cap` exists to enforce, checked as a
/// predicate over the clamp's own output rather than against any literal: the
/// A1 bound must fit inside the §5 depth cap, or the terminal fairness shed
/// fires before the non-terminal A1 Reject ever can.
fn a1_bites_first(clamped: &ConsumerCapacity, depth_cap: usize) -> bool {
    clamped
        .max_in_flight
        .saturating_add(clamped.max_buffer_depth)
        <= depth_cap
        && clamped.max_in_flight > 0
}

/// Verifier finding 4: the record's "A1 bites first" rested on `128 + 1024 <
/// 4096` and nothing compared those numbers. Here they are compared, across
/// every shape the wire can present — including a declared window far above the
/// cap, which is exactly the case the wire path took verbatim.
#[test]
fn the_clamp_makes_a1_bite_first_at_every_declared_window() {
    // From 2 upward: below that no capacity with two positive bands can fit
    // (the minimum legal sum IS 2), so the invariant is unsatisfiable rather
    // than violated and the degenerate caps have their own pin below.
    for depth_cap in [2_usize, 3, 64, 1_024, 1_152, 4_096, 65_536, usize::MAX] {
        for declared_window in [1_usize, 16, 128, 1_024, 3_072, 4_096, 1_000_000] {
            let declared = ConsumerCapacity {
                max_in_flight: declared_window,
                max_buffer_depth: DEFAULT_MAX_BUFFER_DEPTH,
            };
            let clamped = clamp_capacity_to_depth_cap(declared, depth_cap);
            assert!(
                a1_bites_first(&clamped, depth_cap),
                "window {declared_window} under cap {depth_cap} clamped to \
                 {clamped:?}, whose A1 bound is above the cap"
            );
        }
    }
}

/// The clamp is not a blunt instrument: a declaration that already satisfies
/// the invariant is passed through untouched, and the shipped defaults are such
/// a declaration. Without this the previous test would pass on a clamp that
/// returned `(1, 0)` for everything.
#[test]
fn a_declaration_that_already_fits_is_untouched() {
    let shipped = ConsumerCapacity {
        // What the SDKs declare (`remote/tcp/subscription.rs`), against the
        // default `max_subscription_inbox_depth`.
        max_in_flight: 1_024,
        max_buffer_depth: DEFAULT_MAX_BUFFER_DEPTH,
    };
    assert_eq!(
        clamp_capacity_to_depth_cap(shipped, 4_096),
        shipped,
        "the property held today, and the clamp must not move it"
    );

    let library_defaults = default_capacity();
    assert_eq!(
        clamp_capacity_to_depth_cap(library_defaults, 4_096),
        library_defaults,
        "nor the library defaults"
    );
}

/// And it clamps the case that motivated it: a client declaring a window above
/// `depth_cap - max_buffer_depth` gets a window that is smaller than it asked
/// for, with the bus's buffer band intact.
#[test]
fn an_oversized_declared_window_is_cut_to_the_cap_and_the_policy_band_survives() {
    let declared = ConsumerCapacity {
        max_in_flight: 1_000_000,
        max_buffer_depth: DEFAULT_MAX_BUFFER_DEPTH,
    };
    let clamped = clamp_capacity_to_depth_cap(declared, 4_096);
    assert_eq!(
        clamped.max_buffer_depth, DEFAULT_MAX_BUFFER_DEPTH,
        "the buffer band is bus POLICY: the client's declaration yields first"
    );
    assert_eq!(
        clamped.max_in_flight,
        4_096 - DEFAULT_MAX_BUFFER_DEPTH,
        "and the window takes exactly the headroom left"
    );
}

/// **PIN (round 2) — the clamp never emits a capacity that
/// [`ConsumerCapacity::validate`] refuses.**
///
/// Round-2 verifier sub-finding. The clamp's output is not advice, it is what
/// `SubscriptionInbox::install_capacity` installs — and an invalid capacity is
/// not installed at all, so the inbox silently keeps the library defaults
/// (128 + 1024 = 1152) and the caller believes it clamped. That is the exact
/// invariant violation the clamp exists to prevent, reached through the clamp.
///
/// `a1_bites_first` did not catch it because it only requires
/// `max_in_flight > 0`: `(1, 0)` satisfies it and is illegal. A predicate that
/// cannot see the defect it is guarding is the vacuous shape, so the legality
/// of every output is asserted here directly, at every cap the other pin walks
/// PLUS the degenerate ones.
#[test]
fn the_clamp_never_emits_an_illegal_capacity() {
    for depth_cap in [
        0_usize,
        1,
        2,
        3,
        64,
        1_024,
        1_152,
        4_096,
        65_536,
        usize::MAX,
    ] {
        for declared_window in [1_usize, 16, 128, 1_024, 3_072, 4_096, 1_000_000] {
            let declared = ConsumerCapacity {
                max_in_flight: declared_window,
                max_buffer_depth: DEFAULT_MAX_BUFFER_DEPTH,
            };
            let clamped = clamp_capacity_to_depth_cap(declared, depth_cap);
            assert!(
                clamped.validate().is_ok(),
                "window {declared_window} under cap {depth_cap} clamped to \
                 {clamped:?}, which `validate` refuses — so it would not be \
                 installed at all and the inbox would keep its defaults"
            );
        }
    }
}

/// The degenerate caps, stated rather than glossed. No capacity with two
/// positive bands can sum to less than 2, so for `depth_cap < 2` the "A1 bites
/// first" invariant is UNSATISFIABLE — not violated, unsatisfiable. The clamp
/// returns the minimal legal capacity and the §5 fairness trip owns the shed.
/// Documented at the function; pinned here so the doc and the bytes cannot
/// drift apart again.
#[test]
fn a_degenerate_depth_cap_yields_the_minimal_legal_capacity() {
    for depth_cap in [0_usize, 1] {
        let clamped = clamp_capacity_to_depth_cap(default_capacity(), depth_cap);
        assert_eq!(
            clamped,
            ConsumerCapacity {
                max_in_flight: 1,
                max_buffer_depth: 1,
            },
            "cap {depth_cap} admits no capacity that fits, so the answer is the \
             smallest LEGAL one"
        );
        assert!(clamped.validate().is_ok());
    }
    // The first cap at which the invariant becomes satisfiable, and it holds.
    let clamped = clamp_capacity_to_depth_cap(default_capacity(), 2);
    assert!(a1_bites_first(&clamped, 2));
}