liminal/channel/

subscription.rs

//! LIM-002 R2/R3: subscriptions backed by real beamr processes.
//!
//! Each subscription owns a real, scheduler-supervised beamr native process
//! (a [`SubscriberProcess`]) plus the in-memory inbox the channel actor delivers
//! matching envelopes into. The channel actor LINKS to this process's pid on
//! `Subscribe`; when the [`SubscriptionHandle`] is dropped (or the caller
//! unsubscribes) the process is terminated, the link fires an `{EXIT, pid, _}`
//! signal, and the trapping channel actor removes the dead subscriber from its
//! fan-out list. There is NO weak-Arc polling: liveness is observed structurally
//! through the beamr link/EXIT path, exactly as the conversation actor observes
//! its participants (`conversation/actor/beam.rs`).
//!
//! R3 predicates live INSIDE the channel actor process: a [`SubscriptionPredicate`]
//! is a boxed `Fn(&Envelope) -> bool` owned by the actor's subscriber
//! registration and evaluated at delivery time. This mirrors the participant
//! `behaviour` pattern (a boxed trait object the process owns); for an in-memory
//! ephemeral channel there is no need for a serialisable predicate, so a closure
//! the actor holds is the simplest faithful design.

use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, OnceLock};

use beamr::native::native_process::{NativeContext, NativeHandler, NativeOutcome};
use beamr::process::ExitReason;
use beamr::scheduler::Scheduler;
use beamr::term::binary_ref::BinaryRef;

use crate::channel::admission::{default_capacity, defer_after_append};
use crate::channel::schema::SchemaId;
use crate::channel::wire::{decode_envelope, encode_envelope};
use crate::durability::bridge::block_on;
use crate::durability::{ConsumerCursor, DurableStore, MessageEnvelope, replay_range};
use crate::envelope::{Envelope, PublisherId};
use crate::error::LiminalError;
use crate::pressure::{CapacityError, CapacityTracker, ConsumerCapacity, PressureSignal};

/// A shared, cloneable in-memory inbox a subscriber receives delivered envelopes
/// on. See [`SubscriptionInbox`].
pub(crate) type SubscriberInbox = Arc<SubscriptionInbox>;

/// A wake callback fired on EVERY envelope admitted to the inbox (R3, §1.2(2)).
///
/// The server installs one that fires the CONNECTION scheduler's `READY` marker,
/// so a publish into a parked connection's inbox wakes it. It is called from the
/// PUBLISHING actor's slice (the channel actor for local delivery, the subscriber
/// process for a remote frame), so it must be cheap and non-blocking — a single
/// `enqueue_atom_message`. `None` (no notifier installed) is the standalone
/// library / test case: nothing to wake, delivery still lands in the inbox.
pub type InboxNotifier = Arc<dyn Fn() + Send + Sync>;

/// One shared inbox-byte budget per connection (§5).
///
/// Spent across ALL that connection's subscription inboxes. The accounting unit is
/// serialized envelope bytes AS ADMITTED — charged at enqueue, released at dequeue
/// — so the signed 4 MiB product is exact and envelope-size-independent, not a
/// per-inbox count bounding a variable the design does not control.
#[derive(Debug)]
pub struct ConnectionInboxBudget {
    used: AtomicUsize,
    cap: usize,
}

impl ConnectionInboxBudget {
    /// Creates a shared budget with `cap` bytes of headroom across all the
    /// connection's inboxes.
    #[must_use]
    pub fn new(cap: usize) -> Arc<Self> {
        Arc::new(Self {
            used: AtomicUsize::new(0),
            cap,
        })
    }

    /// Attempts to charge `bytes`. Returns `true` and reserves the bytes when they
    /// fit within the remaining budget, `false` (reserving nothing) on overflow. A
    /// CAS loop keeps the reservation exact under concurrent charges from several
    /// inboxes — no transient over-charge is ever observable.
    fn try_charge(&self, bytes: usize) -> bool {
        let mut current = self.used.load(Ordering::Acquire);
        loop {
            let Some(projected) = current.checked_add(bytes) else {
                return false;
            };
            if projected > self.cap {
                return false;
            }
            match self.used.compare_exchange_weak(
                current,
                projected,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => return true,
                Err(observed) => current = observed,
            }
        }
    }

    /// Releases `bytes` previously charged (at dequeue). Saturating so a double
    /// release can never wrap the counter below zero.
    fn release(&self, bytes: usize) {
        let mut current = self.used.load(Ordering::Acquire);
        loop {
            let next = current.saturating_sub(bytes);
            match self.used.compare_exchange_weak(
                current,
                next,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => return,
                Err(observed) => current = observed,
            }
        }
    }

    /// Bytes currently reserved across the connection's inboxes.
    #[cfg(test)]
    pub(crate) fn used(&self) -> usize {
        self.used.load(Ordering::Acquire)
    }
}

/// Everything a server connection installs onto a subscription's inbox.
///
/// Carries the shared §5 byte budget, the per-inbox fairness cap, and the R3
/// wake notifier. Passed INTO the subscribe call so the installation happens at
/// inbox construction — strictly BEFORE the registration is published to the
/// channel actor — closing the pre-install window in which envelopes could be
/// admitted uncharged or without a wake.
pub struct InboxInstall {
    /// Shared per-connection byte budget (§5).
    pub budget: Arc<ConnectionInboxBudget>,
    /// Per-inbox envelope-count fairness trip (§5).
    pub depth_cap: usize,
    /// R3 wake notifier fired on every envelope admitted to the inbox. `None`
    /// when the caller has no waker (scheduler-free unit tests).
    pub notifier: Option<InboxNotifier>,
    /// A1 §2: the consumer's declared capacity — the in-flight window the
    /// `Subscribe` frame's `max_in_flight` carries, paired with the bus-policy
    /// buffer band. `None` keeps
    /// [`crate::channel::admission::default_capacity`], so an inbox is bounded
    /// whether or not the caller declared anything: there is no opt-out.
    ///
    /// Declared HERE rather than through a separate `subscribe_with_capacity`
    /// because this install is the seam that was deliberately unified to make
    /// installation happen at inbox construction, strictly before the
    /// registration reaches the actor. A second subscribe entry point would
    /// fork that ordering guarantee.
    pub capacity: Option<ConsumerCapacity>,
}

impl std::fmt::Debug for InboxInstall {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("InboxInstall")
            .field("depth_cap", &self.depth_cap)
            .field("has_notifier", &self.notifier.is_some())
            .field("capacity", &self.capacity)
            .finish_non_exhaustive()
    }
}

/// Mutable inbox state guarded by one lock: the queued envelopes (each carrying
/// the exact bytes CHARGED for it, so release is symmetric with charge), the
/// installed shared budget and per-inbox fairness cap, the wake notifier, and
/// the closed marker.
struct InboxState {
    /// Each entry is `(envelope, charged_bytes)` — the amount actually charged to
    /// the shared budget at enqueue (0 when no budget was installed at admit
    /// time), released verbatim at dequeue/close. Storing the CHARGE, not the
    /// size, makes release byte-identical to charge on every entry even across a
    /// budget install, so the budget can never under- or over-release.
    queue: VecDeque<(Envelope, usize, Option<u64>)>,
    /// Shared per-connection byte budget (§5). `None` = unbounded (standalone
    /// library use / tests), preserving the pre-bounding behaviour exactly.
    budget: Option<Arc<ConnectionInboxBudget>>,
    /// Per-inbox envelope-count secondary fairness trip (§5). `usize::MAX` = off;
    /// stops one subscription starving its siblings inside the shared byte budget.
    depth_cap: usize,
    /// Wake callback (R3). `None` until the connection installs one.
    notifier: Option<InboxNotifier>,
    /// Terminal marker set by [`SubscriptionInbox::close`]: admissions are refused
    /// WITHOUT charging, and all queued charges have been released. Closing is the
    /// release-by-construction seam — every teardown path (explicit unsubscribe,
    /// overflow shed, connection teardown, and the `Drop` backstop) funnels
    /// through it, so queued bytes can never be stranded on the connection-lifetime
    /// budget.
    closed: bool,
    /// A1 §2: the consumer's declared capacity. The two bands are DERIVED from
    /// `queue.len()` against it on every admission — never stored, never
    /// mutated — so `queue.len() == in_flight_band + buffered_band` holds by
    /// construction and no counter can drift or underflow.
    capacity: ConsumerCapacity,
    /// A1 §4: this durable subscriber missed a live push and is converging via
    /// replay. While set, EVERY live push is shed, so the host-side refill owns
    /// the queue and in-order delivery is preserved. Cleared only when a refill
    /// read reaches the log head with no shed racing it.
    ///
    /// Never set on an ephemeral channel: there is no log to catch up from, so
    /// an ephemeral overflow is a plain per-message Reject and nothing more.
    lagging: bool,
    /// Bumped on every live push shed while `lagging`. The refill loop reads it
    /// before a head read and clears `lagging` only if it is unchanged
    /// afterwards, so a push shed during the read window cannot be lost to a
    /// "caught up" verdict taken before it arrived.
    shed_generation: u64,
    /// A1 §4: the next durable sequence this subscriber has not been offered.
    /// Advanced by every live push that was queued (or predicate-filtered) and
    /// by every refilled entry; frozen the moment a shed sets `lagging`, which
    /// is what makes the frozen value the exact start of the missed range.
    ///
    /// MONOTONE (round 2). Advancing is `max(current, position + 1)`, never a
    /// bare assignment. Under the ordered fan-out this is the same number every
    /// time; if that ordering is ever broken again, monotonicity is what stops a
    /// low-positioned push dragging the cursor back over messages this
    /// subscriber already holds and sending the next refill to re-offer them.
    next_replay_seq: u64,
    /// A1 §4 exactly-once: one past the highest durable sequence the REPLAY
    /// DOOR has offered to (or filtered past for) this subscriber. The
    /// comparand of the live-push suppression guard, and advanced by nothing
    /// else — not by a live push, not by the join seed.
    ///
    /// Split from `next_replay_seq` in round 2. The cursor answers "where does
    /// the missed range start", which every door moves; the guard asks "did the
    /// refill already hand this subscriber this exact position", which only the
    /// refill can answer. Collapsing the two made a live push that arrived
    /// behind a higher-positioned sibling look like one that had already been
    /// delivered, and it was silently dropped.
    replay_offered_seq: u64,
}

/// The shared subscription inbox (R3 + §5). Replaces the bare
/// `Arc<Mutex<VecDeque<Envelope>>>`: it fires a wake notifier on every admitted
/// envelope and enforces the connection-scoped byte budget plus the per-inbox
/// fairness trip, shedding the offending subscription on overflow.
pub(crate) struct SubscriptionInbox {
    state: Mutex<InboxState>,
    /// Sticky overflow marker: set when an admission is refused by the byte budget
    /// or the fairness trip. The server-side delivery pump observes it and sheds
    /// this subscription with a typed error frame, mirroring the outbound overflow
    /// policy (a slow consumer sheds its own subscription; it cannot grow server
    /// memory without bound). Sticky (never cleared) because a shed is terminal.
    overflowed: AtomicBool,
}

impl std::fmt::Debug for SubscriptionInbox {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("SubscriptionInbox")
            .field("overflowed", &self.overflowed.load(Ordering::Acquire))
            .finish_non_exhaustive()
    }
}

/// What an inbox did with an offered envelope.
///
/// Two independent regimes meet here and MUST NOT be collapsed:
///
/// * **A1 pacing** ([`Admitted`](Self::Admitted) / [`Deferred`](Self::Deferred)
///   / [`Rejected`](Self::Rejected)) — per-message, non-terminal, and
///   de-escalating. A resuming consumer walks Rejected → Deferred → Admitted as
///   its queue drains, and NONE of these three touch the sticky overflow
///   marker. A1's Reject sheds one message for one subscriber; it does not end
///   the subscription.
/// * **§5 memory safety** ([`BudgetExceeded`](Self::BudgetExceeded) /
///   [`FairnessTripped`](Self::FairnessTripped)) — terminal. These set the
///   sticky overflow marker and the server pump sheds the whole subscription
///   with a typed error frame. They stay exactly as they were: the byte budget
///   is the memory backstop, A1's counts are the pacing bound. With the
///   defaults (128 + 1024 = 1152 envelopes against a 4096 depth cap) A1 bites
///   first, so a §5 shed remains what it has always been — a genuine budget
///   violation, not a slow consumer.
///
/// A closed inbox refuses without charging and without marking either way.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum InboxAdmission {
    /// A1 Accept band: the envelope was queued with in-flight credit to spare,
    /// and the wake notifier fired.
    Admitted(PressureSignal),
    /// A1 Defer band: the in-flight window is exhausted but the buffer band has
    /// room, so the envelope WAS queued and the wake notifier fired. Defer is
    /// purely a pacing signal to the producer — the bus already holds the
    /// message and the consumer's next pop is its "redelivery".
    Deferred(PressureSignal),
    /// A1 Reject band: the buffer band is full, so the envelope is shed for
    /// this subscriber and NOT queued. Non-terminal — the sticky overflow
    /// marker is untouched.
    Rejected(PressureSignal),
    /// A1 §4 exactly-once: this durable envelope was ALREADY offered to this
    /// subscriber — its sequence is below the REPLAY WATERMARK, so the
    /// host-side refill read it out of the log and offered it before this live
    /// push arrived. Suppressed here, not queued again.
    ///
    /// The watermark is the refill's own record, moved by no other door. A push
    /// that merely arrives behind a higher-positioned sibling is NOT this: it
    /// was offered by nobody, and it is queued.
    ///
    /// Counts as a delivery ([`Self::is_queued`]) because it genuinely was
    /// one: the envelope entered this inbox, exactly once, by the other door.
    /// Its signal never says Reject — a message the bus took custody of was
    /// not shed.
    AlreadyOffered(PressureSignal),
    /// The shared connection byte budget (§5) had no room; the subscription is
    /// shed. Carries the bands the envelope was shed AT, in the A1 vocabulary
    /// — see [`Self::shed_signal`] for why that is not [`Self::signal`].
    BudgetExceeded(PressureSignal),
    /// The per-inbox fairness trip (§5) is full; the subscription is shed.
    /// Carries the bands it was shed at, as [`Self::BudgetExceeded`] does.
    FairnessTripped(PressureSignal),
    /// The inbox was closed (unsubscribe/shed/teardown): the envelope is dropped
    /// without charging the budget — a closed inbox can never re-accumulate cost.
    Closed,
}

impl InboxAdmission {
    /// Whether the envelope reached this subscriber's queue. True for the two
    /// A1 bands that admit (Accept and Defer) and for
    /// [`AlreadyOffered`](Self::AlreadyOffered) — where it reached the queue on
    /// an earlier, replay-side offer of the SAME durable position — and false
    /// for every shed and refusal. This is what the delivery-ack count means:
    /// "the envelope entered the inbox", counted once per envelope.
    pub(crate) const fn is_queued(&self) -> bool {
        matches!(
            *self,
            Self::Admitted(_) | Self::Deferred(_) | Self::AlreadyOffered(_)
        )
    }

    /// The A1 pressure signal this admission carries, when it took one. The
    /// §5 refusals and a closed inbox took no pressure decision and so carry
    /// none — they are not pacing outcomes and must never be aggregated as if
    /// they were.
    pub(crate) const fn signal(&self) -> Option<&PressureSignal> {
        match *self {
            Self::Admitted(ref signal)
            | Self::Deferred(ref signal)
            | Self::Rejected(ref signal)
            | Self::AlreadyOffered(ref signal) => Some(signal),
            Self::BudgetExceeded(_) | Self::FairnessTripped(_) | Self::Closed => None,
        }
    }

    /// The bands a §5 refusal shed this envelope at, always a
    /// [`PressureSignal::Reject`].
    ///
    /// Deliberately NOT folded into [`Self::signal`]: a §5 refusal is a
    /// memory-safety verdict about bytes, not a pacing verdict about a slow
    /// consumer, and it must never be reported as one. But it IS a
    /// delivered-to-nobody outcome for its subscriber, and the producer's
    /// aggregate answers exactly one question — did the bus take custody. With
    /// these outcomes invisible to the aggregate, a publish EVERY subscriber
    /// dropped resolved to the zero-subscriber `Accept` sentinel and
    /// `is_admitted()` lied; `PressureAggregate::record_dropped` is this
    /// accessor's only caller and the one place the two regimes meet.
    ///
    /// `Closed` carries none: a closed inbox belongs to a subscription that is
    /// already gone, and counting it would turn a publish landing during a
    /// teardown into a producer-visible Reject.
    pub(crate) const fn shed_signal(&self) -> Option<&PressureSignal> {
        match *self {
            Self::BudgetExceeded(ref signal) | Self::FairnessTripped(ref signal) => Some(signal),
            Self::Admitted(_)
            | Self::Deferred(_)
            | Self::Rejected(_)
            | Self::AlreadyOffered(_)
            | Self::Closed => None,
        }
    }
}

/// The A1 decision for an inbox holding `queued` envelopes under `capacity`.
///
/// Counts are DERIVED from the authoritative queue length (§0.1/§2) and the
/// verdict is [`CapacityTracker::pressure_signal`] — the complete, already-
/// tested decision model in `pressure/capacity.rs`, reused rather than
/// reimplemented. The tracker's `record_*` mutators are never called here, so
/// the underflow class they can produce cannot arise on this path.
const fn derived_signal(capacity: &ConsumerCapacity, queued: usize) -> PressureSignal {
    CapacityTracker::derived(*capacity, queued).pressure_signal()
}

/// A §5 refusal, expressed in the A1 vocabulary: the verdict is Reject (the
/// envelope was shed for this subscriber and reached nobody) and the bands are
/// the REAL occupancy at the moment of the refusal, split by the same
/// [`CapacityTracker::derived`] authority the pacing decision reads. Nothing is
/// invented and no second count is kept.
const fn shed_signal(capacity: &ConsumerCapacity, queued: usize) -> PressureSignal {
    let tracker = CapacityTracker::derived(*capacity, queued);
    PressureSignal::reject(
        tracker.current_in_flight(),
        capacity.max_in_flight,
        tracker.current_buffer_depth(),
        capacity.max_buffer_depth,
    )
}

impl SubscriptionInbox {
    /// Creates an unbounded, notifier-less inbox — the standalone/default shape,
    /// byte-identical to the pre-bounding behaviour. A server connection passes an
    /// [`InboxInstall`] through subscribe so budget/cap/notifier are installed at
    /// construction instead.
    /// The byte budget and fairness cap are off (the standalone/default shape,
    /// byte-identical to the pre-bounding behaviour); the A1 capacity is NOT.
    /// Every inbox is bounded by the A1 bands from construction, with no
    /// opt-out (§2), which is precisely what closes the unbounded-inbox hole
    /// for a plain library `subscribe()`.
    pub(crate) fn new() -> Arc<Self> {
        Arc::new(Self {
            state: Mutex::new(InboxState {
                queue: VecDeque::new(),
                budget: None,
                depth_cap: usize::MAX,
                notifier: None,
                closed: false,
                capacity: default_capacity(),
                lagging: false,
                shed_generation: 0,
                next_replay_seq: 0,
                replay_offered_seq: 0,
            }),
            overflowed: AtomicBool::new(false),
        })
    }

    /// Installs the connection's shared byte budget and per-inbox fairness cap
    /// (§5). Runs at inbox construction (via [`InboxInstall`]) — before the
    /// registration is published to the channel actor — so no envelope can be
    /// admitted uncharged.
    pub(crate) fn install_budget(&self, budget: Arc<ConnectionInboxBudget>, depth_cap: usize) {
        if let Ok(mut state) = self.state.lock() {
            state.budget = Some(budget);
            state.depth_cap = depth_cap;
        }
    }

    /// Installs the consumer's declared A1 capacity (§2), replacing the
    /// defaults. Runs at inbox construction on the same pre-registration
    /// ordering guarantee as the budget install, so the first envelope the
    /// actor can possibly deliver is already decided against the declared
    /// window.
    ///
    /// # Errors
    ///
    /// Returns [`CapacityError::InvalidCapacity`] when a band is zero. A zero
    /// window would make every publish a Reject, which is a configuration
    /// fault, not a pressure decision — so it is REFUSED, aloud.
    ///
    /// It used to return early and say nothing, which is worse than either
    /// answer: the inbox kept the library defaults (128 + 1024) while the
    /// caller had every reason to believe its declared window was in force. A
    /// declared window must install or refuse; a subscription running under a
    /// bound nobody agreed to is invisible, and a failed subscribe is not.
    pub(crate) fn install_capacity(&self, capacity: ConsumerCapacity) -> Result<(), CapacityError> {
        capacity.validate()?;
        if let Ok(mut state) = self.state.lock() {
            state.capacity = capacity;
        }
        Ok(())
    }

    /// Seeds the replay cursor at the durable log head the subscription joined
    /// at (§4). A gap this subscriber never had cannot be a gap it has to
    /// catch up on, so a fresh subscription's missed range starts where it
    /// started listening — not at sequence zero, which would replay the whole
    /// history on the first shed.
    ///
    /// It seeds the REPLAY ORIGIN only, never the suppression watermark. A row
    /// appended just below the head can still be fanned out after this
    /// registration lands, and that live push is DELIVERED: the fan-out list is
    /// the authority on who is a live subscriber, this subscription was on it,
    /// and no door ever offered that position. Suppressing it would drop a
    /// message the producer is told was delivered, to buy nothing — the refill
    /// starts at the seed, so it can never re-offer the position and there is
    /// no duplicate to prevent.
    pub(crate) fn seed_replay_cursor(&self, head: u64) {
        if let Ok(mut state) = self.state.lock() {
            state.next_replay_seq = head;
        }
    }

    /// Installs the wake notifier (R3), fired on every admitted envelope,
    /// capturing the connection scheduler's enqueue handle (§1.2(2)).
    ///
    /// Defensive invariant: the install RECHECKS non-emptiness under the lock and
    /// fires the notifier (outside the lock) when envelopes are already queued —
    /// those envelopes were admitted while there was no notifier to fire, so
    /// without this recheck their wake would be lost to install ordering. On the
    /// normal construction path the queue is empty and this is a no-op.
    pub(crate) fn install_notifier(&self, notifier: InboxNotifier) {
        let fire = {
            let Ok(mut state) = self.state.lock() else {
                return;
            };
            let pending = !state.queue.is_empty();
            let handle = notifier.clone();
            state.notifier = Some(notifier);
            pending.then_some(handle)
        };
        if let Some(notifier) = fire {
            notifier();
        }
    }

    /// Admits `envelope` under the byte budget and fairness trip, charging the
    /// serialized bytes as admitted and firing the wake notifier for EVERY
    /// admitted envelope (level-triggered — see the fire site below for why the
    /// edge-triggered form starved a subscriber that fell more than one delivery
    /// slice behind). On budget/fairness refusal the sticky overflow marker is set
    /// and the envelope dropped (memory never grows past the bound); a closed
    /// inbox refuses without charging or marking.
    ///
    /// The notifier fires OUTSIDE the state lock so the publishing actor's slice
    /// never holds the inbox lock across the scheduler enqueue.
    pub(crate) fn admit(&self, envelope: Envelope) -> InboxAdmission {
        self.admit_at(envelope, None)
    }

    /// [`Self::admit`] carrying the envelope's durable sequence, when the
    /// publishing channel has one (A1 §4).
    ///
    /// The position is what makes auto-catch-up possible: a queued envelope
    /// advances the replay cursor past itself, and the first shed FREEZES the
    /// cursor at the message it lost, so the missed range is exactly
    /// `[next_replay_seq, head)` with no bookkeeping that could disagree with
    /// the queue.
    ///
    /// That sentence assumes durable positions arrive here in ASCENDING ORDER,
    /// and they do: the fan-out command is enqueued under the append lock that
    /// assigned the position and the actor drains its queue in push order (see
    /// `ChannelHandle::persist_and_enqueue`). The cursor still advances by
    /// `max` rather than assignment, so a future reordering costs a duplicate
    /// at worst and never a message.
    ///
    /// The A1 band decision is taken FIRST — ahead of the §5 depth cap and byte
    /// budget — and never touches the sticky overflow marker, so an A1 Reject
    /// paces one message instead of killing the subscription. It reads
    /// `queue.len()` under the SAME lock that then pushes, so decide+push is
    /// atomic against a concurrent [`Self::pop`] and concurrent publishers
    /// serialise on it (the publish-time TOCTOU §2 closes).
    pub(crate) fn admit_at(
        &self,
        envelope: Envelope,
        durable_position: Option<u64>,
    ) -> InboxAdmission {
        self.admit_inner(envelope, durable_position, false)
    }

    /// The replay door (A1 §4): admits one entry the host-side refill read out
    /// of the durable log, through the SAME bounded admission a live push takes.
    ///
    /// Returns whether the entry was queued; `false` stops the refill loop and
    /// leaves the subscriber lagging, so the next pop retries. This is the ONLY
    /// writer allowed to move the replay cursor while the gap is open — a live
    /// push is shed instead, which is what preserves in-order delivery.
    ///
    /// It is also the only writer of `replay_offered_seq`, alongside
    /// [`Self::note_replayed_filtered`]: this door is the one that can offer a
    /// position ahead of that position's own live push, so this door's record
    /// is the only honest basis for suppressing that push when it arrives.
    pub(crate) fn admit_replayed(&self, envelope: Envelope, sequence: u64) -> bool {
        self.admit_inner(envelope, Some(sequence), true).is_queued()
    }

    /// Steps the replay cursor past an entry the refill read but this
    /// subscriber's predicate filtered out. Unlike [`Self::note_filtered`] this
    /// advances WHILE lagging, because the refill owns the cursor then.
    ///
    /// It advances the replay watermark too: the refill has DEALT with this
    /// position — a live push for it would be the second offer of a message
    /// this subscription has already decided it does not want, and the guard
    /// must suppress it exactly as it suppresses a re-offered queued one.
    pub(crate) fn note_replayed_filtered(&self, sequence: u64) {
        if let Ok(mut state) = self.state.lock()
            && !state.closed
        {
            let next = sequence.saturating_add(1);
            state.next_replay_seq = state.next_replay_seq.max(next);
            state.replay_offered_seq = state.replay_offered_seq.max(next);
        }
    }

    /// The one admission body. `replay` distinguishes the host-side refill from
    /// a live push: only a live push is shed while lagging, and only a live
    /// push can OPEN a gap.
    fn admit_inner(
        &self,
        envelope: Envelope,
        durable_position: Option<u64>,
        replay: bool,
    ) -> InboxAdmission {
        // Serialize once, before the lock: the admitted byte count is the wire
        // size (§5 denomination). The entry stores the amount actually CHARGED
        // (0 when no budget is installed), so dequeue/close releases exactly
        // what enqueue charged.
        let bytes = encode_envelope(&envelope).len();
        let signal;
        let notifier = {
            let Ok(mut state) = self.state.lock() else {
                // A poisoned inbox lock is terminal for this subscription; treat it
                // as a shed rather than silently dropping into a dead inbox.
                //
                // Zero bands, matching `capacity_bound`'s poisoned-lock reading:
                // an inbox whose lock is poisoned admits nothing and holds no
                // live buffer, so there is no occupancy to report. The verdict
                // — shed, reached nobody — is the part that is certain.
                self.overflowed.store(true, Ordering::Release);
                return InboxAdmission::BudgetExceeded(PressureSignal::reject(0, 0, 0, 0));
            };
            if state.closed {
                return InboxAdmission::Closed;
            }
            // A1 §4 EXACTLY ONCE, and the seam that makes it true.
            //
            // The durable publish path appends and THEN fans out
            // (`ChannelHandle::publish_with_delivery`), so a row is readable
            // from the log strictly before its live push reaches any inbox. The
            // host-side refill reads the log. It can therefore read, offer, and
            // step past a row whose live push has not happened yet — and then
            // clear `lagging` legitimately, because nothing was SHED and the
            // shed generation is unchanged. The live push then arrives at a
            // caught-up subscriber and would be queued a SECOND time.
            //
            // The comparand is `replay_offered_seq`, and it is deliberately NOT
            // the replay cursor (round 2). Only the replay door can offer a
            // position ahead of that position's own live push, so only the
            // replay door can create the duplicate this guard exists to
            // suppress — and `replay_offered_seq` is advanced by the replay
            // door alone. A live push below it was genuinely handed to this
            // subscriber already, out of the log, and its second offer is
            // suppressed here.
            //
            // WHY NOT THE CURSOR. `next_replay_seq` is advanced by every door,
            // a live push included, and a live push moves it PAST ITSELF. Test
            // a live push against it and any push that arrives behind a
            // higher-positioned sibling reads as already-offered although
            // nothing ever offered it — silent loss, counted to the producer as
            // a delivery (`is_queued`), with nothing shed to make it
            // recoverable. That is strictly worse than the duplicate, and it is
            // what round 1 shipped.
            //
            // The publish path is now ordered — the fan-out command is enqueued
            // under the append lock that assigned the position
            // (`ChannelHandle::persist_and_enqueue`) — so an inversion cannot
            // reach this line at all. This guard does not RELY on that: it is
            // the second, independent reason the loss cannot happen, and the
            // one that keeps the failure direction safe (a duplicate, never a
            // loss) if the publish path is ever reordered again.
            //
            // Checked BEFORE the lagging shed on purpose: an already-offered
            // envelope must not bump the shed generation, or a refill would
            // loop chasing a gap that a delivered message opened.
            if !replay
                && let Some(position) = durable_position
                && position < state.replay_offered_seq
            {
                // Never a Reject: the bus took custody of this envelope for
                // this subscriber. The bands still report the real occupancy,
                // so the producer's pacing hint stays honest.
                let signal = defer_after_append(derived_signal(&state.capacity, state.queue.len()));
                return InboxAdmission::AlreadyOffered(signal);
            }
            // A1 §4: a lagging durable subscriber sheds EVERY live push so the
            // host-side refill can deliver the missed range in order. Its live
            // band is closed by policy, which is reported as a full buffer band
            // — true of the live path, and it keeps the aggregate's
            // `Reject => delivered to nobody` exact.
            if state.lagging && !replay {
                state.shed_generation = state.shed_generation.wrapping_add(1);
                let capacity = state.capacity;
                return InboxAdmission::Rejected(PressureSignal::reject(
                    capacity.max_in_flight,
                    capacity.max_in_flight,
                    capacity.max_buffer_depth,
                    capacity.max_buffer_depth,
                ));
            }
            signal = derived_signal(&state.capacity, state.queue.len());
            if matches!(signal, PressureSignal::Reject { .. }) {
                // Shed for THIS message only: nothing is queued, nothing is
                // charged, and `overflowed` stays exactly as it was. The next
                // pop moves this subscriber back into the Defer band.
                //
                // On a durable channel the shed OPENS a gap: the cursor is
                // already at this sequence and is now frozen there, so the
                // missed range starts exactly at the message that was lost. A
                // refill Reject means the free band was already full and is not
                // a new gap — the subscriber is lagging already.
                if durable_position.is_some() && !replay {
                    state.lagging = true;
                    state.shed_generation = state.shed_generation.wrapping_add(1);
                }
                return InboxAdmission::Rejected(signal);
            }
            if state.queue.len() >= state.depth_cap {
                self.overflowed.store(true, Ordering::Release);
                let shed = shed_signal(&state.capacity, state.queue.len());
                let notifier = state.notifier.clone();
                drop(state);
                if let Some(notifier) = notifier {
                    notifier();
                }
                return InboxAdmission::FairnessTripped(shed);
            }
            let charged = match state.budget.as_ref() {
                Some(budget) => {
                    if !budget.try_charge(bytes) {
                        self.overflowed.store(true, Ordering::Release);
                        let shed = shed_signal(&state.capacity, state.queue.len());
                        let notifier = state.notifier.clone();
                        drop(state);
                        if let Some(notifier) = notifier {
                            notifier();
                        }
                        return InboxAdmission::BudgetExceeded(shed);
                    }
                    bytes
                }
                None => 0,
            };
            state.queue.push_back((envelope, charged, durable_position));
            // A1 §4: this envelope is now this subscriber's, so the replay
            // cursor steps past it. Done under the SAME lock as the push, so a
            // concurrent shed cannot interleave and leave the cursor pointing
            // at a message that was in fact delivered.
            //
            // `max`, not assignment: the cursor names where the missed range
            // STARTS, and it must never be dragged back below a position this
            // subscriber already holds — a refill from there would re-read and
            // re-offer delivered messages.
            //
            // The replay watermark moves only on the replay door's own offers.
            // That is the whole point of it being a separate number: it is the
            // record of what the REFILL handed over, and a live push is not
            // evidence about that.
            if let Some(position) = durable_position {
                let next = position.saturating_add(1);
                state.next_replay_seq = state.next_replay_seq.max(next);
                if replay {
                    state.replay_offered_seq = state.replay_offered_seq.max(next);
                }
            }
            // LEVEL-TRIGGERED, not edge-triggered: EVERY successful enqueue fires.
            //
            // The consumer drains a BOUNDED slice (the server's delivery pump: 32
            // envelopes per connection slice), so an inbox more than a slice deep
            // does NOT empty when it is serviced. Under the edge rule a subscriber
            // in exactly that state — the normal state of anyone who has fallen
            // behind — earned one wake for an entire burst and none afterwards,
            // and every later envelope arrived with no wake attached to it at all.
            // That is a lost-wake hazard on its face, and it is removed here.
            //
            // HONEST SCOPE, measured — this is NOT what starves a subscriber at
            // today's bytes, and it must not be cited as if it were. A/B over 120
            // fresh-boot iterations per arm (gate-logs/p0-55/) found the edge and
            // level forms indistinguishable: 51.7% vs 53.3% of boots lost a
            // subscriber to the depth-cap shed. The reason is R6 coalescing
            // itself. N fires collapse into one mailbox drain, so turning one wake
            // into N cannot buy the connection a single extra SLICE, and slices —
            // not wakes — are what drain the queue. The variable that does move it
            // is the pump's per-slice budget (32 -> 256 took the same harness to
            // 0/120), which is a cross-connection fairness knob and not this
            // file's to turn. See the report accompanying this lane.
            //
            // What firing every time costs: one non-blocking `enqueue_atom_message`
            // per admitted envelope, which R6 coalescing collapses to one slice.
            // An idle inbox admits nothing and so still fires nothing — the
            // zero-cost-at-rest property is unchanged.
            state.notifier.clone()
        };
        if let Some(notifier) = notifier {
            notifier();
        }
        if matches!(signal, PressureSignal::Defer { .. }) {
            InboxAdmission::Deferred(signal)
        } else {
            InboxAdmission::Admitted(signal)
        }
    }

    /// Steps the replay cursor past an envelope this subscriber's predicate
    /// filtered out (A1 §2 + §4).
    ///
    /// A non-matching envelope contributes no backpressure and is not a gap:
    /// leaving the cursor behind it would make every later refill re-read and
    /// re-filter it forever. A lagging subscriber's cursor stays frozen — the
    /// refill owns it until the gap closes.
    pub(crate) fn note_filtered(&self, durable_position: u64) {
        if let Ok(mut state) = self.state.lock()
            && !state.lagging
            && !state.closed
        {
            state.next_replay_seq = state
                .next_replay_seq
                .max(durable_position.saturating_add(1));
        }
    }

    /// Removes and returns the next envelope, releasing its CHARGED bytes back to
    /// the shared budget (exact charge/release symmetry).
    #[cfg(test)]
    pub(crate) fn pop(&self) -> Option<Envelope> {
        self.pop_positioned()
            .ok()
            .flatten()
            .map(|(envelope, _)| envelope)
    }

    fn pop_positioned(&self) -> Result<Option<(Envelope, Option<u64>)>, LiminalError> {
        let (envelope, charged, position, budget) = {
            let mut state = self.state.lock().map_err(subscription_error)?;
            let Some((envelope, charged, position)) = state.queue.pop_front() else {
                return Ok(None);
            };
            (envelope, charged, position, state.budget.clone())
        };
        // Release the charged bytes AFTER dropping the state lock so the shared
        // budget's atomic is never touched while the inbox lock is held.
        if let Some(budget) = budget {
            budget.release(charged);
        }
        Ok(Some((envelope, position)))
    }

    /// Non-consuming race-barrier query used after a connection arms readiness.
    pub(crate) fn has_pending(&self) -> bool {
        self.state.lock().is_ok_and(|state| !state.queue.is_empty())
    }

    /// Atomically closes the inbox, releasing every queued charge back to the
    /// shared budget: under the lock it marks the inbox closed, drains all
    /// entries, and detaches the notifier and budget; the summed release happens
    /// outside the lock. Idempotent. Admissions after close are refused without
    /// charging ([`InboxAdmission::Closed`]).
    ///
    /// This is the release-by-construction seam: explicit unsubscribe, overflow
    /// shed, and connection teardown ALL reach it through the subscription
    /// handle's drop (see [`SubscriptionInner::drop`]), and the inbox's own `Drop`
    /// is the final backstop — no teardown path can strand queued bytes on the
    /// connection-lifetime budget.
    pub(crate) fn close(&self) {
        let (released, budget) = {
            let Ok(mut state) = self.state.lock() else {
                return;
            };
            if state.closed {
                return;
            }
            state.closed = true;
            let released: usize = state
                .queue
                .drain(..)
                .map(|(_envelope, charged, _position)| charged)
                .sum();
            state.notifier = None;
            (released, state.budget.take())
        };
        if let Some(budget) = budget {
            budget.release(released);
        }
    }

    /// Whether this subscription has been marked for shedding by an overflow.
    pub(crate) fn is_overflowed(&self) -> bool {
        self.overflowed.load(Ordering::Acquire)
    }

    /// Number of queued envelopes.
    pub(crate) fn queued_len(&self) -> usize {
        self.state.lock().map_or(0, |state| state.queue.len())
    }

    /// Number of queued envelopes (test observability alias).
    #[cfg(test)]
    pub(crate) fn len(&self) -> usize {
        self.queued_len()
    }

    /// This inbox's total A1 bound: `max_in_flight + max_buffer_depth`.
    ///
    /// A poisoned lock reports `0`, which contributes nothing to the durable
    /// channel-aggregate watermark — the honest reading, since an inbox whose
    /// lock is poisoned admits nothing and so holds no live buffer.
    pub(crate) fn capacity_bound(&self) -> usize {
        self.state.lock().map_or(0, |state| {
            state
                .capacity
                .max_in_flight
                .saturating_add(state.capacity.max_buffer_depth)
        })
    }

    /// Whether this subscriber is converging on the durable log via replay
    /// (A1 §4).
    pub(crate) fn is_lagging(&self) -> bool {
        self.state.lock().is_ok_and(|state| state.lagging)
    }

    /// The next refill batch to read, or `None` when none is due (A1 §4).
    ///
    /// `Some((generation, cursor, budget))` only when this subscriber is
    /// lagging AND its queue has drained below the low watermark
    /// (`max_in_flight / 2`) AND the free buffer band is non-empty. That last
    /// conjunct is load-bearing: a zero-length read would return an empty batch
    /// and be indistinguishable from "caught up", which would clear `lagging`
    /// while the gap was still open.
    ///
    /// `budget` is the free band, so a replay batch can never blow the bound it
    /// exists to serve (§4 "replay batches size themselves to the free buffer
    /// band").
    pub(crate) fn refill_plan(&self) -> Option<(u64, u64, usize)> {
        let state = self.state.lock().ok()?;
        if !state.lagging || state.closed {
            return None;
        }
        let queued = state.queue.len();
        let low_watermark = state.capacity.max_in_flight / 2;
        if queued > low_watermark {
            return None;
        }
        let bound = state
            .capacity
            .max_in_flight
            .saturating_add(state.capacity.max_buffer_depth);
        let budget = bound.saturating_sub(queued);
        if budget == 0 {
            return None;
        }
        Some((state.shed_generation, state.next_replay_seq, budget))
    }

    /// Declares the gap closed, unless a live push was shed since `generation`
    /// was taken (A1 §4).
    ///
    /// Returns `true` when `lagging` was cleared. The generation check is what
    /// makes the loop-until-caught-up test correct under concurrent appends: a
    /// push shed while the head read was in flight bumps the generation, so the
    /// caller loops instead of declaring victory over a message it never
    /// delivered.
    pub(crate) fn clear_lagging_if_unchanged(&self, generation: u64) -> bool {
        let Ok(mut state) = self.state.lock() else {
            return false;
        };
        if state.shed_generation != generation {
            return false;
        }
        state.lagging = false;
        true
    }
}

impl Drop for SubscriptionInbox {
    fn drop(&mut self) {
        // Backstop: if no teardown path ever called `close`, release the queued
        // charges here so the last Arc dropping can never strand budget bytes.
        // Idempotent against an earlier close (the closed marker short-circuits).
        self.close();
    }
}

/// A delivery predicate evaluated by the channel actor against each published
/// envelope. `None` (no predicate) means deliver everything.
pub(crate) type SubscriptionPredicate = Arc<dyn Fn(&Envelope) -> bool + Send + Sync>;

/// Real beamr native process backing one subscription.
///
/// For LOCAL delivery it is an idle handler (mirroring
/// `aion::worker::link::IdleWorkerProcess`): local envelopes travel through the
/// shared [`SubscriberInbox`] the channel actor writes and
/// [`SubscriptionHandle::try_next`] reads. Its other job is to BE a first-class
/// linkable, killable process whose lifetime equals the subscription's, so the
/// channel actor detects the subscription dying via a real EXIT signal rather
/// than by polling a weak pointer.
///
/// For CROSS-NODE delivery (SRV-005) it is also the landing point for a remote
/// publish: a remote node sends a published envelope, encoded by
/// [`crate::channel::wire::encode_envelope`], as a single beamr binary directly
/// to this process's pid (the pid the cluster registered in the channel's
/// distributed process group). The binary lands in this process's mailbox; the
/// handler decodes it back into an [`Envelope`] and pushes it onto the SAME
/// inbox a local publish would, so a subscriber observes local and remote
/// messages identically. Non-binary wakeups (trapped `{EXIT, _, _}` signals) are
/// drained and ignored.
struct SubscriberProcess {
    inbox: SubscriberInbox,
}

impl NativeHandler for SubscriberProcess {
    fn handle(&mut self, ctx: &mut NativeContext<'_>) -> NativeOutcome {
        // Trapping is set authoritatively at spawn (see `SubscriptionHandle::spawn`)
        // so it holds before the actor ever links — re-assert it here defensively
        // for any future restart of this handler.
        ctx.set_trap_exit(true);
        // Drain every queued wakeup. A binary message is a remote envelope frame
        // (SRV-005) to decode and enqueue; everything else (e.g. a trapped
        // `{EXIT, _, _}` tuple from a crashed actor this subscriber outlives) is
        // ignored. Death is driven only by an explicit `terminate_process` on
        // unsubscribe/handle drop.
        while let Some(message) = ctx.recv() {
            if BinaryRef::new(message).is_some() {
                // beamr 0.20.0 ties byte access to a `HeapBorrow` witness, and
                // `NativeContext` exposes no borrow of the process heap, so the
                // frame is deep-copied into ETS-owned storage and read under
                // the copy's own witness — the one borrow-correct path a
                // native handler has (beamr#36 tracks a direct accessor). A
                // frame whose copy fails is dropped, same as one that fails to
                // decode: a corrupt cross-node payload must never crash the
                // subscriber or stall delivery of well-formed messages.
                if let Ok(owned) = beamr::ets::copy_term_to_ets(message)
                    && let Some(binary) = BinaryRef::new(owned.root())
                {
                    self.accept_remote_frame(binary.as_bytes(owned.borrow_terms()));
                }
            }
        }
        NativeOutcome::Wait
    }
}

impl SubscriberProcess {
    /// Decode a remote envelope frame and push it onto the inbox. A frame that
    /// fails to decode is dropped: a corrupt cross-node payload must never crash
    /// the subscriber or stall delivery of well-formed messages.
    fn accept_remote_frame(&self, bytes: &[u8]) {
        let Ok(envelope) = decode_envelope(bytes) else {
            return;
        };
        // R3: the remote-delivery leg fires the same wake notifier and obeys the
        // same §5 byte budget as the local leg. An overflow marks the subscription
        // for shedding (inside `admit`); the frame is dropped rather than growing
        // server memory.
        self.inbox.admit(envelope);
    }
}

/// The actor-side record of one subscriber: the inbox to deliver into and the
/// optional predicate to gate delivery. Held by the channel actor INSIDE its
/// process, keyed by the subscriber process's pid.
pub(crate) struct SubscriberRegistration {
    pid: u64,
    inbox: SubscriberInbox,
    predicate: Option<SubscriptionPredicate>,
}

impl SubscriberRegistration {
    pub(crate) const fn pid(&self) -> u64 {
        self.pid
    }

    /// Offers `envelope` to this subscriber, returning the admission outcome —
    /// or `None` when the subscriber's predicate filtered it out.
    ///
    /// `None` is not a refusal and is deliberately distinct from every
    /// [`InboxAdmission`]: A1 §2 requires the decision to run **after the
    /// predicate**, so a non-matching subscriber contributes no backpressure at
    /// all. Folding it in as a Reject would produce false Defer signals for
    /// producers on a heavily filtered channel.
    ///
    /// R3 + §5 are unchanged underneath: admission charges the connection byte
    /// budget, fires the wake notifier for every queued envelope, and on a
    /// §5 overflow marks the subscription for shedding. Only
    /// [`InboxAdmission::is_queued`] counts as a genuine delivery.
    pub(crate) fn deliver(
        &self,
        envelope: &Envelope,
        durable_position: Option<u64>,
    ) -> Option<InboxAdmission> {
        if let Some(predicate) = self.predicate.as_ref() {
            if !predicate(envelope) {
                // Not this subscriber's message, so not this subscriber's gap:
                // step the replay cursor past it (A1 §4).
                if let Some(position) = durable_position {
                    self.inbox.note_filtered(position);
                }
                return None;
            }
        }
        Some(self.inbox.admit_at(envelope.clone(), durable_position))
    }

    /// This subscriber's live-buffer occupancy and its bound, for the durable
    /// pre-append watermark (A1 §4). Read HOST-SIDE, off the actor's command
    /// slice, and deliberately coarse: no predicate is evaluated.
    pub(crate) fn occupancy(&self) -> (usize, usize) {
        (self.inbox.queued_len(), self.inbox.capacity_bound())
    }
}

impl std::fmt::Debug for SubscriberRegistration {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("SubscriberRegistration")
            .field("pid", &self.pid)
            .field("has_predicate", &self.predicate.is_some())
            .finish_non_exhaustive()
    }
}

/// Handle returned by channel subscriptions for receiving validated envelopes.
///
/// Owns the subscriber's beamr pid, the shared inbox, and a clone of the
/// scheduler so the process can be terminated when the subscription ends. The
/// handle is the subscription's lifetime: dropping the last clone terminates the
/// subscriber process, whose EXIT prunes the channel actor's fan-out list.
#[derive(Clone)]
pub struct SubscriptionHandle {
    inner: Arc<SubscriptionInner>,
}

struct SubscriptionInner {
    pid: u64,
    inbox: SubscriberInbox,
    scheduler: Arc<Scheduler>,
    /// A1 §4: everything the host-side catch-up needs, attached by the durable
    /// subscribe path. `OnceLock` because it is written exactly once, before
    /// the handle is handed to the caller, and read on every `try_next`.
    /// Absent on an ephemeral channel — there is nothing to replay from.
    refill: OnceLock<DurableRefill>,
    /// Serialises concurrent refills of ONE subscription. Contended only when
    /// two threads drain the same handle; the loser skips (the winner is
    /// already filling the same queue) rather than queueing behind a store
    /// read.
    refilling: Mutex<()>,
    consumer: OnceLock<Mutex<SubscriptionConsumer>>,
}

struct SubscriptionConsumer {
    cursor: ConsumerCursor,
    pending: Option<u64>,
}

fn subscription_error(error: impl std::fmt::Display) -> LiminalError {
    LiminalError::SubscriptionFailed {
        message: error.to_string(),
    }
}

/// The host-side auto-catch-up source for a lagging durable subscriber
/// (A1 §4, graft §0.4).
///
/// Deliberately held by the SUBSCRIPTION HANDLE, not by the channel actor: the
/// implementability judge relocated catch-up off the actor's command slice
/// precisely so a blocking store read can never stall the actor for every
/// other subscriber on the channel. The refill runs on whichever thread is
/// draining this one subscription.
struct DurableRefill {
    store: Arc<dyn DurableStore>,
    /// The channel's durable partition stream key.
    stream_key: String,
    /// The schema id stamped on refilled envelopes: the channel's schema at
    /// subscribe time. The durable record stores validated payload bytes, not
    /// the schema that validated them, so this is the only id available and
    /// pretending otherwise would be an invention.
    schema_id: SchemaId,
    /// The subscription's predicate, re-applied to refilled envelopes so a
    /// filtered subscription catches up on exactly the messages it would have
    /// received live.
    predicate: Option<SubscriptionPredicate>,
}

impl std::fmt::Debug for DurableRefill {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("DurableRefill")
            .field("stream_key", &self.stream_key)
            .field("has_predicate", &self.predicate.is_some())
            .finish_non_exhaustive()
    }
}

impl SubscriptionHandle {
    /// Spawns a real subscriber process on `scheduler` and returns the handle
    /// plus its actor-side registration record (carrying any predicate).
    ///
    /// # Errors
    /// Returns [`LiminalError::SubscriptionFailed`] when the scheduler cannot
    /// spawn the subscriber process.
    pub(crate) fn spawn(
        scheduler: &Arc<Scheduler>,
        predicate: Option<SubscriptionPredicate>,
        install: Option<InboxInstall>,
    ) -> Result<(Self, SubscriberRegistration), LiminalError> {
        let inbox: SubscriberInbox = SubscriptionInbox::new();
        // Install the §5 budget/fairness cap and the R3 wake notifier AT
        // CONSTRUCTION — strictly before the registration is handed to the
        // channel actor — so there is no window in which a publish can be
        // admitted uncharged, past the depth cap, or without a wake.
        if let Some(install) = install {
            inbox.install_budget(install.budget, install.depth_cap);
            if let Some(capacity) = install.capacity {
                // REFUSES ALOUD. A window the bus cannot honour fails the
                // subscribe rather than leaving the caller on defaults it never
                // asked for and cannot observe.
                inbox.install_capacity(capacity).map_err(|error| {
                    LiminalError::SubscriptionFailed {
                        message: format!("declared consumer capacity is invalid: {error}"),
                    }
                })?;
            }
            if let Some(notifier) = install.notifier {
                inbox.install_notifier(notifier);
            }
        }
        let process_inbox = Arc::clone(&inbox);
        let factory = Box::new(move || {
            Box::new(SubscriberProcess {
                inbox: Arc::clone(&process_inbox),
            }) as Box<dyn NativeHandler>
        });
        // trap_exit is set on the process BEFORE it is published as runnable
        // (0.16.1 `spawn_native_trap_exit` — pre-runnable by construction), so
        // an abnormal channel-actor crash is trapped (delivered as a message
        // the subscriber drains) instead of cascading across the link and
        // killing the subscriber. This makes the subscriber outlive a
        // channel-actor crash so the restarted actor can re-link to it on boot
        // (R2/R4): the flag is in place before the process's first slice, with
        // no post-spawn window for `set_trap_exit` to race (or return
        // `NoCaller` against a mid-first-slice process).
        let pid = scheduler.spawn_native_trap_exit(factory).map_err(|error| {
            LiminalError::SubscriptionFailed {
                message: format!("failed to spawn subscriber process: {error:?}"),
            }
        })?;
        let handle = Self {
            inner: Arc::new(SubscriptionInner {
                pid,
                inbox: Arc::clone(&inbox),
                scheduler: Arc::clone(scheduler),
                refill: OnceLock::new(),
                refilling: Mutex::new(()),
                consumer: OnceLock::new(),
            }),
        };
        let registration = SubscriberRegistration {
            pid,
            inbox,
            predicate,
        };
        Ok((handle, registration))
    }

    /// Returns the beamr pid of the subscriber process this handle owns.
    #[must_use]
    pub(crate) fn pid(&self) -> u64 {
        self.inner.pid
    }

    /// Attempts to receive the next delivered envelope without blocking.
    ///
    /// # Errors
    ///
    /// Returns [`LiminalError::SubscriptionFailed`] when the subscription inbox cannot be read.
    pub fn try_next(&self) -> Result<Option<Envelope>, LiminalError> {
        self.try_next_positioned()
            .map(|next| next.map(|(envelope, _)| envelope))
    }

    /// Receives one envelope with its store position, retaining it until acknowledged.
    ///
    /// # Errors
    /// A cursor still awaits its previous acknowledgement, or the inbox/store failed.
    pub fn try_next_positioned(&self) -> Result<Option<(Envelope, Option<u64>)>, LiminalError> {
        let mut consumer = self
            .inner
            .consumer
            .get()
            .map(|consumer| consumer.lock())
            .transpose()
            .map_err(subscription_error)?;
        if consumer
            .as_ref()
            .is_some_and(|consumer| consumer.pending.is_some())
        {
            return Err(subscription_error(
                "a delivered position still awaits acknowledgement",
            ));
        }
        self.refill_if_lagging()?;
        let next = self.inner.inbox.pop_positioned()?;
        if let (Some(consumer), Some((_, position))) = (consumer.as_mut(), next.as_ref()) {
            consumer.pending = *position;
        }
        Ok(next)
    }

    /// Persists the next position only after the consumer has settled this delivery.
    ///
    /// # Errors
    /// The position was not delivered, or checkpoint/flush failed.
    pub fn acknowledge_position(&self, position: u64) -> Result<(), LiminalError> {
        let consumer = self
            .inner
            .consumer
            .get()
            .ok_or_else(|| subscription_error("subscription has no durable consumer"))?;
        let refill = self
            .inner
            .refill
            .get()
            .ok_or_else(|| subscription_error("subscription has no durable stream"))?;
        let mut consumer = consumer.lock().map_err(subscription_error)?;
        if consumer.pending != Some(position) {
            return Err(subscription_error(
                "acknowledgement does not name the delivered position",
            ));
        }
        let next = position
            .checked_add(1)
            .ok_or_else(|| subscription_error("subscription position overflow"))?;
        block_on(consumer.cursor.checkpoint(refill.store.as_ref(), next))
            .map_err(subscription_error)?
            .map_err(subscription_error)?;
        block_on(refill.store.flush())
            .map_err(subscription_error)?
            .map_err(subscription_error)?;
        consumer.pending = None;
        drop(consumer);
        Ok(())
    }

    /// The persisted consumer position, absent on an ordinary live subscription.
    ///
    /// # Errors
    /// The consumer lock failed.
    pub fn consumer_position(&self) -> Result<Option<u64>, LiminalError> {
        self.inner
            .consumer
            .get()
            .map(|consumer| {
                consumer
                    .lock()
                    .map(|consumer| consumer.cursor.current_offset())
                    .map_err(subscription_error)
            })
            .transpose()
    }

    pub(crate) fn resume_consumer(&self, id: &str) -> Result<(), LiminalError> {
        if id.is_empty() {
            return Err(subscription_error("durable consumer id is empty"));
        }
        let refill =
            self.inner.refill.get().ok_or_else(|| {
                subscription_error("a durable consumer requires a durable channel")
            })?;
        let cursor = block_on(ConsumerCursor::resume(
            id,
            &refill.stream_key,
            refill.store.as_ref(),
        ))
        .map_err(subscription_error)?
        .map_err(subscription_error)?;
        let offset = cursor.current_offset();
        self.inner
            .consumer
            .set(Mutex::new(SubscriptionConsumer {
                cursor,
                pending: None,
            }))
            .map_err(|_| subscription_error("durable consumer is already installed"))?;
        let mut state = self.inner.inbox.state.lock().map_err(subscription_error)?;
        state.next_replay_seq = offset;
        state.replay_offered_seq = offset;
        state.lagging = true;
        drop(state);
        Ok(())
    }

    /// Seeds this subscription's replay cursor at the durable log head it is
    /// joining at (A1 §4). Called by the durable subscribe path BEFORE the
    /// registration reaches the channel actor, so the first publish this
    /// subscription can possibly see is already measured against the right
    /// origin.
    pub(crate) fn seed_replay_cursor(&self, head: u64) {
        self.inner.inbox.seed_replay_cursor(head);
    }

    /// Attaches the durable catch-up source (A1 §4). Called once by the durable
    /// subscribe path before the handle is returned; a second call is a
    /// mis-wiring of that path and trips a debug assertion rather than being
    /// silently ignored.
    pub(crate) fn attach_durable_refill(
        &self,
        store: Arc<dyn DurableStore>,
        stream_key: String,
        schema_id: SchemaId,
        predicate: Option<SubscriptionPredicate>,
    ) {
        let attached = self.inner.refill.set(DurableRefill {
            store,
            stream_key,
            schema_id,
            predicate,
        });
        debug_assert!(
            attached.is_ok(),
            "the refill OnceLock is set exactly once per attach: attach_durable_refill \
             was called twice on one subscription, which only the durable subscribe \
             path may call and only before the handle is returned"
        );
    }

    /// Replays the missed range into the bounded inbox until this subscriber is
    /// caught up or its free band is full (A1 §4, graft §0.4).
    ///
    /// **Loop-until-caught-up.** The log head can move under concurrent
    /// appends, so "caught up" is not a snapshot taken once: each iteration
    /// re-reads from the cursor, and the flag clears only when a read comes
    /// back empty AND no live push was shed while that read was in flight. The
    /// chase is bounded because the pre-append watermark throttles producers
    /// while any subscriber lags.
    ///
    /// **In-order, exactly once.** Live pushes stay shed while `lagging` is
    /// set, so nothing can overtake the replay; the cursor advances only past
    /// entries this call actually offered, so nothing is offered twice.
    ///
    /// Recovery runs before dequeue so a store failure leaves the delivery
    /// in the inbox and reaches the consumer as an error.
    pub(crate) fn refill_if_lagging(&self) -> Result<(), LiminalError> {
        let Some(refill) = self.inner.refill.get() else {
            return Ok(());
        };
        if !self.inner.inbox.is_lagging() {
            return Ok(());
        }
        // Another thread draining this same subscription is already refilling
        // it; a second concurrent replay would read the same range twice.
        let _guard = match self.inner.refilling.try_lock() {
            Ok(guard) => guard,
            Err(std::sync::TryLockError::WouldBlock) => return Ok(()),
            Err(std::sync::TryLockError::Poisoned(error)) => return Err(subscription_error(error)),
        };
        while let Some((generation, cursor, budget)) = self.inner.inbox.refill_plan() {
            let batch = block_on(replay_range(
                refill.store.as_ref(),
                &refill.stream_key,
                cursor,
                budget,
            ))
            .map_err(subscription_error)?
            .map_err(subscription_error)?;
            if batch.is_empty() {
                // The read reached the head. Clear the gap ONLY if no live push
                // was shed since the generation was taken; otherwise loop and
                // read again for the message that shed.
                if self.inner.inbox.clear_lagging_if_unchanged(generation) {
                    return Ok(());
                }
                continue;
            }
            for (sequence, stored) in batch {
                let envelope = refilled_envelope(&stored, refill.schema_id);
                let matched = refill
                    .predicate
                    .as_ref()
                    .is_none_or(|predicate| predicate(&envelope));
                if matched {
                    // Through the SAME bounded admission as a live push. The
                    // lagging flag is still set, so `admit_at` would shed this
                    // as a live push — the refill therefore goes in through the
                    // replay door below, which is the only writer allowed to
                    // move the cursor while the gap is open.
                    if !self.inner.inbox.admit_replayed(envelope, sequence) {
                        return Ok(());
                    }
                } else {
                    self.inner.inbox.note_replayed_filtered(sequence);
                }
            }
        }
        Ok(())
    }

    /// The subscriber process's pid: the id
    /// [`ChannelHandle::publish_recording`](crate::channel::ChannelHandle::publish_recording)
    /// records for this subscription, and the one a
    /// [`ClusterObserver`](crate::channel::ClusterObserver) hears it by.
    #[must_use]
    pub fn subscriber_pid(&self) -> u64 {
        self.inner.pid
    }

    /// Whether an envelope is available without consuming it.
    #[must_use]
    pub fn has_pending(&self) -> bool {
        self.inner.inbox.has_pending()
    }

    /// Whether an overflow has marked this subscription for shedding (§5).
    #[must_use]
    pub fn is_overflowed(&self) -> bool {
        self.inner.inbox.is_overflowed()
    }
}

impl std::fmt::Debug for SubscriptionHandle {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("SubscriptionHandle")
            .field("pid", &self.inner.pid)
            .finish_non_exhaustive()
    }
}

impl Drop for SubscriptionInner {
    fn drop(&mut self) {
        // Close the inbox FIRST: atomically mark it closed, drain queued entries,
        // and release every charged byte back to the shared connection budget
        // (§5 — queued bytes must never be stranded on the connection-lifetime
        // budget by unsubscribe, shed, or teardown; all of them funnel through
        // this drop). Post-close deliveries from the channel actor (whose EXIT
        // prune below is asynchronous) are refused without charging.
        self.inbox.close();
        // Terminating the subscriber process fires the bidirectional link to the
        // channel actor, which traps the EXIT and removes this subscriber from
        // its fan-out list. This is the real-beamr unsubscribe-on-drop path.
        self.scheduler
            .terminate_process(self.pid, ExitReason::Normal);
    }
}

/// Rebuilds a bus [`Envelope`] from a durable log entry, for the host-side
/// refill (A1 §4).
///
/// Two fields are deliberately NOT reconstructed, because the log does not hold
/// them and inventing them would make a refilled envelope claim more than the
/// store can support:
///
/// * **causal context** — the durable publish path persists `None` for it
///   (`ChannelHandle::persist_and_enqueue`), so there is nothing to restore;
/// * **message id** — the durable record does not store the live envelope's id,
///   so a fresh one is minted. A consumer that needs to correlate a replayed
///   message with a live one uses its durable position, which is exact.
fn refilled_envelope(stored: &MessageEnvelope, schema_id: SchemaId) -> Envelope {
    let millis = i64::try_from(stored.timestamp).unwrap_or(i64::MAX);
    let timestamp = chrono::TimeZone::timestamp_millis_opt(&chrono::Utc, millis)
        .single()
        .unwrap_or_else(chrono::Utc::now);
    Envelope::with_timestamp(
        stored.payload.clone(),
        None,
        schema_id,
        PublisherId::new(stored.publisher_id.clone()),
        timestamp,
    )
}

/// WR-9b: the REAL [`SubscriberProcess`] running on beamr's cooperative
/// (single-threaded / wasm) [`beamr::scheduler::WasmScheduler`].
///
/// This proves the production subscriber handler — the same `NativeHandler` the
/// threaded [`SubscriptionHandle::spawn`] spawns — runs unchanged on the
/// cooperative scheduler that a browser host drives. There is no toy stand-in:
/// the test spawns the genuine [`SubscriberProcess`], delivers a genuine
/// [`crate::channel::wire::encode_envelope`] frame as a real beamr binary, pumps
/// cooperative `run_until_idle` turns, and asserts the envelope is decoded by the
/// handler's own `accept_remote_frame` path and lands in the shared inbox a
/// [`SubscriptionHandle::try_next`] would read.
///
/// The handler runs cooperatively AS-IS: its `handle` only touches
/// platform-neutral [`NativeContext`] capabilities (`set_trap_exit`, `recv`),
/// [`BinaryRef`], and [`decode_envelope`] — none of which reach for threads,
/// tokio, sockets, or a `SharedState`. The only wiring the smoke supplies is the
/// cooperative driver (spawn + owned-binary delivery + turn pump), exactly the
/// host-side seam the threaded `SubscriptionHandle`/channel-actor provide on
/// native.
#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
mod cooperative_smoke {
    use std::cell::RefCell;
    use std::rc::Rc;
    use std::sync::Arc;

    use beamr::atom::AtomTable;
    use beamr::ets::copy_term_to_ets;
    use beamr::module::ModuleRegistry;
    use beamr::native::BifRegistryImpl;
    use beamr::process::heap::Heap;
    use beamr::scheduler::WasmScheduler;
    use beamr::term::shared_binary::{SharedBinary, write_proc_bin};

    use super::{SubscriberInbox, SubscriberProcess, SubscriptionInbox};
    use crate::channel::SchemaId;
    use crate::channel::wire::encode_envelope;
    use crate::envelope::{Envelope, PublisherId};

    /// Build a cooperative scheduler the way a wasm host holds it (single
    /// `Rc<RefCell<…>>` on one thread).
    fn cooperative_scheduler() -> Rc<RefCell<WasmScheduler>> {
        let atom_table = Arc::new(AtomTable::with_common_atoms());
        let modules = Arc::new(ModuleRegistry::new());
        let bifs = Arc::new(BifRegistryImpl::new());
        Rc::new(RefCell::new(WasmScheduler::new(atom_table, modules, bifs)))
    }

    /// Encode `envelope` into the production wire frame and wrap it as a
    /// heap-independent beamr binary term ready for `send_owned`, mirroring how a
    /// remote node hands a published frame to a subscriber pid (SRV-005).
    fn frame_as_owned_binary(envelope: &Envelope) -> beamr::ets::OwnedTerm {
        let bytes = encode_envelope(envelope);
        let shared = SharedBinary::new(bytes);
        // A ProcBin reference needs three heap words; copy it into ETS-owned
        // memory so the scratch heap can be dropped before delivery.
        let mut scratch = Heap::new(8);
        let words = scratch
            .alloc_slice(3)
            .expect("scratch heap holds a proc-bin reference");
        let term = write_proc_bin(words, &shared).expect("proc-bin term writes");
        copy_term_to_ets(term).expect("frame copies into an owned binary")
    }

    fn sample_envelope() -> Envelope {
        // A whole-millisecond timestamp so the round-trip through the wire codec
        // (which carries millisecond resolution, see `channel::wire`) is exact;
        // `Utc::now()` sub-millisecond precision would otherwise be truncated on
        // decode and is irrelevant to what this smoke proves.
        let timestamp = chrono::TimeZone::timestamp_millis_opt(&chrono::Utc, 1_700_000_000_123)
            .single()
            .expect("valid fixed millisecond timestamp");
        Envelope::with_timestamp(
            b"{\"value\":42}".to_vec(),
            None,
            SchemaId::new(),
            PublisherId::from("publisher-cooperative"),
            timestamp,
        )
    }

    #[test]
    fn real_subscriber_process_delivers_a_published_envelope_cooperatively() {
        let scheduler = cooperative_scheduler();

        // The shared inbox the subscriber pushes decoded envelopes onto — the
        // exact channel the threaded `SubscriptionHandle::try_next` reads.
        let inbox: SubscriberInbox = SubscriptionInbox::new();
        let process_inbox = Arc::clone(&inbox);

        // Spawn the GENUINE production subscriber handler as a first-class native
        // process on the cooperative scheduler.
        let pid = scheduler.borrow_mut().spawn_native_root(Box::new(move || {
            Box::new(SubscriberProcess {
                inbox: Arc::clone(&process_inbox),
            }) as Box<dyn beamr::native::native_process::NativeHandler>
        }));

        // First turn: the handler runs once, asserts trap_exit, finds an empty
        // mailbox, and parks (`Wait`). No envelope has been delivered yet.
        scheduler.borrow_mut().run_until_idle();
        assert_eq!(
            inbox.len(),
            0,
            "no envelope is delivered before one is published"
        );

        // Publish: deliver a real encoded frame as a beamr binary straight to the
        // subscriber pid, exactly as a remote publish lands (SRV-005). This wakes
        // the parked process.
        let published = sample_envelope();
        let frame = frame_as_owned_binary(&published);
        scheduler
            .borrow_mut()
            .send_owned(pid, &frame)
            .expect("frame is delivered to the live subscriber pid");

        // Pump turns: the woken handler drains the binary, decodes it through its
        // own `accept_remote_frame` path, and pushes the envelope onto the inbox.
        let mut delivered = None;
        for _ in 0..8 {
            scheduler.borrow_mut().run_until_idle();
            let next = inbox.pop();
            if let Some(envelope) = next {
                delivered = Some(envelope);
                break;
            }
        }

        assert_eq!(
            delivered.as_ref(),
            Some(&published),
            "the real subscriber decoded and delivered the published envelope"
        );
    }
}

/// R3 (§1.2(2)) + §5 inbox-bounding library core: the notifier fires on every
/// admitted envelope; the shared byte budget is spent across ALL a
/// connection's inboxes; overflow sheds the offending subscription; the per-inbox
/// fairness trip stops one inbox starving its siblings; and charge/release is
/// exact. These exercise [`SubscriptionInbox`]/[`ConnectionInboxBudget`] directly,
/// with no scheduler — the server-side wake wiring and shed are tested there.
#[cfg(test)]
#[allow(clippy::expect_used)]
mod inbox_bounding {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use super::{ConnectionInboxBudget, InboxAdmission, SubscriptionInbox};
    use crate::channel::SchemaId;
    use crate::channel::wire::encode_envelope;
    use crate::envelope::{Envelope, PublisherId};
    use crate::pressure::ConsumerCapacity;

    fn envelope(payload: &[u8]) -> Envelope {
        Envelope::new(
            payload.to_vec(),
            None,
            SchemaId::new(),
            PublisherId::from("inbox-bounding-test"),
        )
    }

    fn admitted_bytes(env: &Envelope) -> usize {
        encode_envelope(env).len()
    }

    /// The level-triggered wake contract (P0 #55). This test previously asserted
    /// the edge-triggered form — that a second admit into a non-empty inbox does
    /// NOT re-fire — which is precisely the starvation the fix removes: the
    /// consumer drains a bounded slice, so a non-empty inbox is the normal state
    /// of a subscriber that has fallen behind, and withholding its wake is what
    /// ratchets it to the depth cap and a permanent shed.
    #[test]
    fn notifier_fires_for_every_admitted_envelope() {
        let inbox = SubscriptionInbox::new();
        let fires = Arc::new(AtomicUsize::new(0));
        let counter = Arc::clone(&fires);
        inbox.install_notifier(Arc::new(move || {
            counter.fetch_add(1, Ordering::Relaxed);
        }));

        // First admit into an empty inbox fires.
        assert!(inbox.admit(envelope(b"a")).is_queued());
        assert_eq!(fires.load(Ordering::Relaxed), 1, "the first admit fires");

        // A second admit into a STILL-NON-EMPTY inbox fires again: the consumer
        // may not have reached this envelope's slice, and R6 coalescing means a
        // redundant marker costs one mailbox atom, never a second slice of work.
        assert!(inbox.admit(envelope(b"b")).is_queued());
        assert_eq!(
            fires.load(Ordering::Relaxed),
            2,
            "an admit into a non-empty inbox still fires"
        );

        // Drain to empty, then admit again: still exactly one fire per admit.
        assert!(inbox.pop().is_some());
        assert!(inbox.pop().is_some());
        assert!(inbox.admit(envelope(b"c")).is_queued());
        assert_eq!(
            fires.load(Ordering::Relaxed),
            3,
            "one fire per admitted envelope, whatever the queue depth was"
        );
    }

    #[test]
    fn shared_budget_is_spent_across_all_a_connections_inboxes() {
        let one = envelope(b"payload-one");
        let two = envelope(b"payload-two");
        // A budget large enough for exactly ONE of the two envelopes.
        let cap = admitted_bytes(&one);
        let budget = ConnectionInboxBudget::new(cap);

        let inbox_a = SubscriptionInbox::new();
        let inbox_b = SubscriptionInbox::new();
        inbox_a.install_budget(Arc::clone(&budget), usize::MAX);
        inbox_b.install_budget(Arc::clone(&budget), usize::MAX);

        // Inbox A admits its envelope, consuming the whole shared budget.
        assert!(inbox_a.admit(one).is_queued());
        assert_eq!(budget.used(), cap, "the shared budget is now fully spent");

        // Inbox B — a SIBLING subscription — is refused: the budget is connection
        // scoped, not per-inbox, so A's fill denies B.
        assert!(matches!(
            inbox_b.admit(two),
            InboxAdmission::BudgetExceeded(_)
        ));
        assert!(
            inbox_b.is_overflowed(),
            "the sibling that overflowed the shared budget is shed"
        );
        assert!(!inbox_a.is_overflowed(), "the inbox that fit is not shed");

        // Draining A releases its bytes back to the SHARED budget, so B could then
        // admit (charge/release symmetry across siblings).
        assert!(inbox_a.pop().is_some());
        assert_eq!(
            budget.used(),
            0,
            "release returns bytes to the shared budget"
        );
    }

    #[test]
    fn overflow_sheds_and_does_not_grow_memory() {
        let env = envelope(b"x");
        let budget = ConnectionInboxBudget::new(admitted_bytes(&env)); // room for one
        let inbox = SubscriptionInbox::new();
        inbox.install_budget(budget, usize::MAX);

        assert!(inbox.admit(env.clone()).is_queued());
        // The next admit overflows: refused, marked for shedding, and NOT queued —
        // the queue length does not grow past the bound.
        assert!(matches!(
            inbox.admit(env),
            InboxAdmission::BudgetExceeded(_)
        ));
        assert!(inbox.is_overflowed());
        assert_eq!(
            inbox.len(),
            1,
            "the overflowed envelope is dropped, not queued"
        );
    }

    #[test]
    fn per_inbox_fairness_trip_stops_one_inbox_starving_siblings() {
        // A huge byte budget so the FAIRNESS count — not the budget — is the trip.
        let budget = ConnectionInboxBudget::new(usize::MAX);
        let inbox = SubscriptionInbox::new();
        inbox.install_budget(budget, 2); // depth cap of 2 envelopes

        assert!(inbox.admit(envelope(b"1")).is_queued());
        assert!(inbox.admit(envelope(b"2")).is_queued());
        // The third trips the fairness cap even though bytes are available.
        assert!(matches!(
            inbox.admit(envelope(b"3")),
            InboxAdmission::FairnessTripped(_)
        ));
        assert!(inbox.is_overflowed());
        assert_eq!(
            inbox.len(),
            2,
            "the fairness trip holds the inbox at its cap"
        );
    }

    #[test]
    fn charge_and_release_are_exact() {
        let budget = ConnectionInboxBudget::new(1024 * 1024);
        let inbox = SubscriptionInbox::new();
        inbox.install_budget(Arc::clone(&budget), usize::MAX);

        let a = envelope(b"first-envelope");
        let b = envelope(b"second-longer-envelope-payload");
        let charge = admitted_bytes(&a) + admitted_bytes(&b);
        assert!(inbox.admit(a).is_queued());
        assert!(inbox.admit(b).is_queued());
        assert_eq!(budget.used(), charge, "used == sum of admitted bytes");

        assert!(inbox.pop().is_some());
        assert!(inbox.pop().is_some());
        assert_eq!(
            budget.used(),
            0,
            "every admitted byte is released on dequeue — exact symmetry"
        );
    }

    /// Review round 1 item 4: closing an inbox with a QUEUED backlog (the shed
    /// shape — an overflowed inbox is near-full by construction) releases every
    /// charged byte back to the shared budget, so a sibling subscription can
    /// admit again. Without the close-release, one shed strands its whole share
    /// of the 4 MiB budget forever.
    #[test]
    fn close_releases_queued_charges_so_siblings_recover() {
        // Budget sized so inbox A can queue a 256-envelope backlog (the §5
        // fairness-cap depth) and exhaust the shared budget doing it.
        let one = envelope(b"backlog-envelope-payload");
        let unit = admitted_bytes(&one);
        let budget = ConnectionInboxBudget::new(unit * 256);

        let inbox_a = SubscriptionInbox::new();
        let inbox_b = SubscriptionInbox::new();
        inbox_a.install_budget(Arc::clone(&budget), usize::MAX);
        inbox_b.install_budget(Arc::clone(&budget), usize::MAX);

        // Queue the full 256-envelope backlog on A, consuming the whole budget.
        for _ in 0..256 {
            assert!(inbox_a.admit(one.clone()).is_queued());
        }
        assert_eq!(budget.used(), unit * 256, "the backlog holds the budget");
        // The sibling is starved (the shed trigger condition).
        assert!(matches!(
            inbox_b.admit(one.clone()),
            InboxAdmission::BudgetExceeded(_)
        ));

        // Shed/unsubscribe/teardown all funnel through close: EVERY queued charge
        // returns to the shared budget in one atomic close.
        inbox_a.close();
        assert_eq!(
            budget.used(),
            0,
            "close releases the entire queued backlog back to the shared budget"
        );
        // The sibling recovers: it can admit again.
        assert!(
            inbox_b.admit(one).is_queued(),
            "a sibling admits again after the other inbox is shed"
        );
    }

    /// Review round 1 item 4: a closed inbox refuses admissions WITHOUT charging
    /// the budget, so a shed subscription can never re-accumulate cost while the
    /// channel actor's asynchronous EXIT prune is still in flight.
    #[test]
    fn closed_inbox_refuses_without_charging() {
        let env = envelope(b"post-close");
        let budget = ConnectionInboxBudget::new(1024 * 1024);
        let inbox = SubscriptionInbox::new();
        inbox.install_budget(Arc::clone(&budget), usize::MAX);

        inbox.close();
        assert_eq!(inbox.admit(env), InboxAdmission::Closed);
        assert_eq!(budget.used(), 0, "a closed inbox never charges the budget");
        assert_eq!(inbox.len(), 0, "a closed inbox never queues");
    }

    /// Review round 1 item 4: the `Drop` backstop — if no teardown path ever
    /// called `close`, the last handle dropping still releases the queued charges
    /// (release-by-construction: a release that cannot be omitted).
    #[test]
    fn drop_backstop_releases_queued_charges() {
        let env = envelope(b"dropped-while-queued");
        let unit = admitted_bytes(&env);
        let budget = ConnectionInboxBudget::new(1024 * 1024);
        {
            let inbox = SubscriptionInbox::new();
            inbox.install_budget(Arc::clone(&budget), usize::MAX);
            assert!(inbox.admit(env.clone()).is_queued());
            assert!(inbox.admit(env).is_queued());
            assert_eq!(budget.used(), unit * 2);
            // No close() call: the Arc drops here.
        }
        assert_eq!(
            budget.used(),
            0,
            "dropping the last inbox handle releases every queued charge"
        );
    }

    /// Review round 1 item 4: close is idempotent, and pop-after-close finds
    /// nothing (the queue was drained into the release).
    #[test]
    fn close_is_idempotent_and_drains_the_queue() {
        let env = envelope(b"x");
        let budget = ConnectionInboxBudget::new(1024 * 1024);
        let inbox = SubscriptionInbox::new();
        inbox.install_budget(Arc::clone(&budget), usize::MAX);
        assert!(inbox.admit(env).is_queued());

        inbox.close();
        inbox.close(); // second close is a no-op, not a double release
        assert_eq!(budget.used(), 0);
        assert!(inbox.pop().is_none(), "a closed inbox holds nothing");
    }

    /// Review round 1 item 5 (charge ownership): an envelope admitted BEFORE the
    /// budget was installed carries a charge of 0 — its dequeue releases exactly
    /// 0 against the later-installed budget, never bytes it did not charge. The
    /// production subscribe path installs the budget at inbox construction so
    /// this window is structurally closed; this pins the defensive invariant
    /// that makes release byte-identical to charge on EVERY entry regardless.
    #[test]
    fn per_entry_charge_ownership_survives_budget_install() {
        let uncharged = envelope(b"admitted-before-budget-install");
        let charged = envelope(b"admitted-after-budget-install");
        let inbox = SubscriptionInbox::new();

        // Admitted with no budget installed: charge ownership 0.
        assert!(inbox.admit(uncharged).is_queued());

        let budget = ConnectionInboxBudget::new(1024 * 1024);
        inbox.install_budget(Arc::clone(&budget), usize::MAX);
        let unit = admitted_bytes(&charged);
        assert!(inbox.admit(charged).is_queued());
        assert_eq!(budget.used(), unit, "only the post-install entry charged");

        // Popping the uncharged entry releases exactly 0 — the budget cannot
        // under-count (over-admitting past the signed 4 MiB) by releasing bytes
        // that were never charged.
        assert!(inbox.pop().is_some());
        assert_eq!(budget.used(), unit, "the uncharged entry released nothing");
        assert!(inbox.pop().is_some());
        assert_eq!(
            budget.used(),
            0,
            "the charged entry released its exact charge"
        );
    }

    /// Review round 1 item 5 (install recheck): installing a notifier onto an
    /// ALREADY-NON-EMPTY inbox fires it exactly once — those envelopes were
    /// admitted while there was no notifier to fire, so the install regenerates
    /// their wake and one can never be lost to install ordering. (The production
    /// subscribe path installs at construction, when the queue is guaranteed
    /// empty; this pins the defensive invariant.)
    #[test]
    fn notifier_install_onto_non_empty_inbox_fires_once() {
        let inbox = SubscriptionInbox::new();
        assert!(inbox.admit(envelope(b"pre-install")).is_queued());

        let fires = Arc::new(AtomicUsize::new(0));
        let counter = Arc::clone(&fires);
        inbox.install_notifier(Arc::new(move || {
            counter.fetch_add(1, Ordering::Relaxed);
        }));
        assert_eq!(
            fires.load(Ordering::Relaxed),
            1,
            "install onto a non-empty inbox regenerates exactly one wake"
        );

        // The install is a ONE-OFF regeneration, not an extra fire per admit: a
        // subsequent admit adds exactly its own one fire.
        assert!(inbox.admit(envelope(b"second")).is_queued());
        assert_eq!(fires.load(Ordering::Relaxed), 2);
    }

    /// **PIN (round 2) — the exactly-once guard suppresses what the REPLAY DOOR
    /// offered, and nothing else.**
    ///
    /// A live push that arrives behind a higher-positioned sibling has been
    /// offered by nobody. Suppressing it is silent loss, and worse than the
    /// duplicate the guard exists to prevent: nothing is shed, so `lagging` is
    /// never set and auto-catch-up can never recover it, while
    /// [`InboxAdmission::is_queued`] still reports it to the producer as
    /// delivered.
    ///
    /// The publish path is ordered so that this cannot happen in production
    /// (the fan-out command is enqueued under the append lock). This pin is the
    /// inbox's OWN control on the same claim: whatever the publish path does,
    /// the guard's comparand may only be advanced by a door that genuinely made
    /// an offer, so the worst an inversion could ever cost is a duplicate —
    /// never a lost message.
    #[test]
    fn a_live_push_is_not_suppressed_by_a_higher_positioned_sibling() {
        let inbox = SubscriptionInbox::new();

        assert!(
            inbox.admit_at(envelope(b"position-5"), Some(5)).is_queued(),
            "the higher-positioned push is queued"
        );
        let behind = inbox.admit_at(envelope(b"position-3"), Some(3));
        assert!(
            !matches!(behind, InboxAdmission::AlreadyOffered(_)),
            "position 3 was offered by NOBODY — its only sin is arriving behind \
             position 5, and suppressing it loses it silently; got {behind:?}"
        );
        assert!(behind.is_queued(), "position 3 must reach the queue");
        assert_eq!(
            inbox.queued_len(),
            2,
            "both live pushes are in the queue, in arrival order"
        );
    }

    /// **PIN (round 2), the anti-vacuity partner** — the guard still fires for
    /// the position the replay door actually offered. Without this, the pin
    /// above passes on a guard that was simply deleted.
    #[test]
    fn a_live_push_the_replay_door_already_offered_is_suppressed() {
        let inbox = SubscriptionInbox::new();

        assert!(
            inbox.admit_replayed(envelope(b"replayed-5"), 5),
            "the replay door offers position 5"
        );
        let live = inbox.admit_at(envelope(b"live-5"), Some(5));
        assert!(
            matches!(live, InboxAdmission::AlreadyOffered(_)),
            "position 5 was already offered by the refill; got {live:?}"
        );
        assert_eq!(
            inbox.queued_len(),
            1,
            "the suppressed push did not enter the queue a second time"
        );
    }

    /// **PIN (round 2) — the refill cursor never backsteps onto a position that
    /// was already delivered.**
    ///
    /// The cursor names where the missed range STARTS. Assigning it
    /// `position + 1` unconditionally lets a live push that arrives behind a
    /// higher-positioned sibling drag it backwards over messages this
    /// subscriber already has, and the next refill then re-reads and re-offers
    /// them — the duplicate this whole seam exists to prevent, arriving through
    /// the other door.
    #[test]
    fn the_refill_cursor_never_backsteps_below_a_delivered_position() {
        let inbox = SubscriptionInbox::new();
        inbox
            .install_capacity(ConsumerCapacity::new(1, 1).expect("1/1 is a legal capacity"))
            .expect("a legal capacity installs");

        assert!(inbox.admit_at(envelope(b"position-5"), Some(5)).is_queued());
        assert!(inbox.admit_at(envelope(b"position-3"), Some(3)).is_queued());
        // The bound (1 + 1) is now full, so this one is shed and OPENS the gap.
        assert!(matches!(
            inbox.admit_at(envelope(b"position-6"), Some(6)),
            InboxAdmission::Rejected(_)
        ));
        assert!(inbox.is_lagging(), "the shed opened a gap");

        assert!(inbox.pop().is_some());
        assert!(inbox.pop().is_some());
        let (_generation, cursor, _budget) = inbox
            .refill_plan()
            .expect("a drained lagging inbox has a refill due");
        assert_eq!(
            cursor, 6,
            "the missed range starts above every delivered position; a cursor of \
             4 would send the refill back over positions 4 and 5, and position 5 \
             is already in this subscriber's hands"
        );
    }
}