liminal/pressure/

capacity.rs

pub use state::{CapacityError, CapacityTracker, ConsumerCapacity};

mod state {
    use crate::pressure::signal::PressureSignal;

    /// Consumer-declared capacity limits for pressure-aware delivery.
    ///
    /// `Copy` because it is two `usize`s: it is passed by value through the
    /// admission path and the depth-cap clamp, and cloning a pair of integers
    /// to read them is noise. Additive.
    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    pub struct ConsumerCapacity {
        /// Maximum messages this consumer can process concurrently.
        pub max_in_flight: usize,
        /// Maximum messages that may wait for this consumer's capacity to free.
        pub max_buffer_depth: usize,
    }

    impl ConsumerCapacity {
        /// Creates a capacity declaration after verifying both limits are positive.
        ///
        /// # Errors
        ///
        /// Returns [`CapacityError::InvalidCapacity`] when either declared limit is zero.
        pub const fn new(
            max_in_flight: usize,
            max_buffer_depth: usize,
        ) -> Result<Self, CapacityError> {
            if max_in_flight == 0 || max_buffer_depth == 0 {
                Err(CapacityError::InvalidCapacity {
                    max_in_flight,
                    max_buffer_depth,
                })
            } else {
                Ok(Self {
                    max_in_flight,
                    max_buffer_depth,
                })
            }
        }

        /// Verifies that the declared capacity contains positive limits.
        ///
        /// # Errors
        ///
        /// Returns [`CapacityError::InvalidCapacity`] when either declared limit is zero.
        pub const fn validate(&self) -> Result<(), CapacityError> {
            if self.max_in_flight == 0 || self.max_buffer_depth == 0 {
                Err(CapacityError::InvalidCapacity {
                    max_in_flight: self.max_in_flight,
                    max_buffer_depth: self.max_buffer_depth,
                })
            } else {
                Ok(())
            }
        }
    }

    /// Capacity tracking failures that keep counters from entering invalid states.
    #[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
    pub enum CapacityError {
        /// A capacity declaration used zero for at least one required positive limit.
        #[error("consumer capacity limits must be positive")]
        InvalidCapacity {
            /// Declared maximum in-flight messages.
            max_in_flight: usize,
            /// Declared maximum buffered messages.
            max_buffer_depth: usize,
        },
        /// Processing completion was recorded while no message was in flight.
        #[error("cannot decrement in-flight count below zero")]
        InFlightUnderflow,
        /// Buffer removal was recorded while no message was buffered.
        #[error("cannot decrement buffer depth below zero")]
        BufferUnderflow,
    }

    /// Per-consumer tracker for current in-flight and buffered message counts.
    #[derive(Clone, Debug, PartialEq, Eq)]
    pub struct CapacityTracker {
        capacity: ConsumerCapacity,
        current_in_flight: usize,
        current_buffer_depth: usize,
    }

    impl CapacityTracker {
        /// Creates a tracker for an explicitly declared consumer capacity.
        #[must_use]
        pub const fn new(capacity: ConsumerCapacity) -> Self {
            Self {
                capacity,
                current_in_flight: 0,
                current_buffer_depth: 0,
            }
        }

        /// Builds a tracker whose two bands are DERIVED from `queued` — the
        /// authoritative depth of the queue the messages actually sit in —
        /// rather than accumulated by independent `record_*` mutations
        /// (A1 §0.1/§2, `docs/design/A1-DEFER-SEMANTICS.md`).
        ///
        /// The in-flight band is `min(queued, max_in_flight)` and the buffered
        /// band is `queued.saturating_sub(max_in_flight)`, so the occupancy
        /// invariant `queued == current_in_flight + current_buffer_depth`
        /// holds for EVERY `queued`, and neither band can be decremented below
        /// zero because neither band is ever decremented at all. That is what
        /// retires the [`CapacityError::InFlightUnderflow`] /
        /// [`CapacityError::BufferUnderflow`] drift class by construction: the
        /// hot path calls this constructor and reads
        /// [`Self::pressure_signal`], and never touches a `record_*` mutator.
        ///
        /// The mutators stay for the explicit-credit (v2) accounting the
        /// design specifies but does not ship, and for the standalone unit
        /// tests already written against them.
        #[must_use]
        pub const fn derived(capacity: ConsumerCapacity, queued: usize) -> Self {
            let max_in_flight = capacity.max_in_flight;
            let current_in_flight = if queued < max_in_flight {
                queued
            } else {
                max_in_flight
            };
            Self {
                capacity,
                current_in_flight,
                current_buffer_depth: queued.saturating_sub(max_in_flight),
            }
        }

        /// Returns the consumer capacity declaration this tracker follows.
        #[must_use]
        pub const fn capacity(&self) -> &ConsumerCapacity {
            &self.capacity
        }

        /// Returns the number of messages currently being processed by the consumer.
        #[must_use]
        pub const fn current_in_flight(&self) -> usize {
            self.current_in_flight
        }

        /// Returns the number of messages currently buffered for the consumer.
        #[must_use]
        pub const fn current_buffer_depth(&self) -> usize {
            self.current_buffer_depth
        }

        /// Records that a message was delivered and processing began.
        pub const fn record_delivery(&mut self) {
            if self.current_in_flight < usize::MAX {
                self.current_in_flight += 1;
            }
        }

        /// Records that processing completed for one in-flight message.
        ///
        /// # Errors
        ///
        /// Returns [`CapacityError::InFlightUnderflow`] if no message is currently in flight.
        pub const fn record_completion(&mut self) -> Result<(), CapacityError> {
            if self.current_in_flight == 0 {
                Err(CapacityError::InFlightUnderflow)
            } else {
                self.current_in_flight -= 1;
                Ok(())
            }
        }

        /// Records that a message was buffered pending consumer capacity.
        pub const fn record_buffered(&mut self) {
            if self.current_buffer_depth < usize::MAX {
                self.current_buffer_depth += 1;
            }
        }

        /// Records that one buffered message left the buffer.
        ///
        /// # Errors
        ///
        /// Returns [`CapacityError::BufferUnderflow`] if no message is currently buffered.
        pub const fn record_buffer_drained(&mut self) -> Result<(), CapacityError> {
            if self.current_buffer_depth == 0 {
                Err(CapacityError::BufferUnderflow)
            } else {
                self.current_buffer_depth -= 1;
                Ok(())
            }
        }

        /// Determines the pressure signal for the next message without mutating counters.
        #[must_use]
        pub const fn pressure_signal(&self) -> PressureSignal {
            if self.current_in_flight < self.capacity.max_in_flight {
                PressureSignal::accept(self.current_in_flight, self.capacity.max_in_flight)
            } else if self.current_buffer_depth < self.capacity.max_buffer_depth {
                PressureSignal::defer(
                    self.current_in_flight,
                    self.capacity.max_in_flight,
                    self.current_buffer_depth,
                    self.capacity.max_buffer_depth,
                )
            } else {
                PressureSignal::reject(
                    self.current_in_flight,
                    self.capacity.max_in_flight,
                    self.current_buffer_depth,
                    self.capacity.max_buffer_depth,
                )
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{CapacityError, CapacityTracker, ConsumerCapacity};
    use crate::pressure::PressureSignal;

    const fn capacity(max_in_flight: usize, max_buffer_depth: usize) -> ConsumerCapacity {
        ConsumerCapacity {
            max_in_flight,
            max_buffer_depth,
        }
    }

    /// A1 §0.1/§2 PIN — **derived counters cannot underflow by construction.**
    ///
    /// The occupancy invariant `queued == in_flight + buffered` is asserted
    /// across every band boundary, and the whole point is that it is a
    /// *property of the constructor*, not of a call sequence: there is no
    /// ordering of pops and pushes that can make it false, because neither band
    /// is ever decremented. The two underflow errors this retires
    /// ([`CapacityError::InFlightUnderflow`] / [`CapacityError::BufferUnderflow`])
    /// are reachable ONLY through the `record_*` mutators, which the admission
    /// hot path never calls.
    #[test]
    fn derived_counters_satisfy_the_occupancy_invariant_at_every_depth() {
        let declared = capacity(4, 8);
        let bound = declared.max_in_flight + declared.max_buffer_depth;
        // Past the bound too: a queue can legitimately hold more than the bound
        // for one observation (a fairness/budget-free inbox admitted before a
        // capacity install), and the bands must still add up rather than wrap.
        for queued in 0..=(bound + 5) {
            let tracker = CapacityTracker::derived(declared, queued);
            assert_eq!(
                tracker.current_in_flight() + tracker.current_buffer_depth(),
                queued,
                "occupancy invariant must hold at depth {queued}"
            );
            assert!(
                tracker.current_in_flight() <= declared.max_in_flight,
                "the in-flight band never exceeds the declared window at depth {queued}"
            );
        }
    }

    /// The band boundaries the A1 decision turns on, read through the SAME
    /// `pressure_signal()` rule the unwired decision model already had.
    #[test]
    fn derived_counters_reproduce_the_accept_defer_reject_bands() {
        let declared = capacity(4, 8);
        assert_eq!(
            CapacityTracker::derived(declared, 0).pressure_signal(),
            PressureSignal::accept(0, 4)
        );
        assert_eq!(
            CapacityTracker::derived(declared, 3).pressure_signal(),
            PressureSignal::accept(3, 4),
            "the last slot in the in-flight window still Accepts"
        );
        assert_eq!(
            CapacityTracker::derived(declared, 4).pressure_signal(),
            PressureSignal::defer(4, 4, 0, 8),
            "a full window with an empty buffer band Defers"
        );
        assert_eq!(
            CapacityTracker::derived(declared, 11).pressure_signal(),
            PressureSignal::defer(4, 4, 7, 8),
            "the last slot in the buffer band still Defers"
        );
        assert_eq!(
            CapacityTracker::derived(declared, 12).pressure_signal(),
            PressureSignal::reject(4, 4, 8, 8),
            "a full buffer band Rejects"
        );
        // De-escalation is the same function read backwards: nothing is stored,
        // so a drained queue reports the band it is actually in.
        assert_eq!(
            CapacityTracker::derived(declared, 4).pressure_signal(),
            PressureSignal::defer(4, 4, 0, 8),
            "a resuming consumer de-escalates through Defer with no state to reset"
        );
    }

    #[test]
    fn consumer_capacity_constructs_with_public_fields_and_validates_positive_limits() {
        let declaration = ConsumerCapacity {
            max_in_flight: 10,
            max_buffer_depth: 50,
        };

        assert_eq!(declaration.max_in_flight, 10);
        assert_eq!(declaration.max_buffer_depth, 50);
        assert_eq!(declaration.validate(), Ok(()));
        assert_eq!(ConsumerCapacity::new(10, 50), Ok(declaration));
        assert_eq!(
            ConsumerCapacity::new(0, 50),
            Err(CapacityError::InvalidCapacity {
                max_in_flight: 0,
                max_buffer_depth: 50,
            })
        );
    }

    #[test]
    fn capacity_tracker_starts_empty_and_records_counts() {
        let mut tracker = CapacityTracker::new(capacity(10, 50));

        assert_eq!(tracker.current_in_flight(), 0);
        assert_eq!(tracker.current_buffer_depth(), 0);
        assert_eq!(tracker.capacity(), &capacity(10, 50));

        tracker.record_delivery();
        assert_eq!(tracker.current_in_flight(), 1);

        assert_eq!(tracker.record_completion(), Ok(()));
        assert_eq!(tracker.current_in_flight(), 0);

        tracker.record_buffered();
        assert_eq!(tracker.current_buffer_depth(), 1);

        assert_eq!(tracker.record_buffer_drained(), Ok(()));
        assert_eq!(tracker.current_buffer_depth(), 0);
    }

    #[test]
    fn capacity_tracker_reports_underflow_errors_without_negative_counts() {
        let mut tracker = CapacityTracker::new(capacity(10, 50));

        assert_eq!(
            tracker.record_completion(),
            Err(CapacityError::InFlightUnderflow)
        );
        assert_eq!(tracker.current_in_flight(), 0);

        assert_eq!(
            tracker.record_buffer_drained(),
            Err(CapacityError::BufferUnderflow)
        );
        assert_eq!(tracker.current_buffer_depth(), 0);
    }

    #[test]
    fn pressure_signal_accepts_when_in_flight_capacity_is_available() {
        let mut tracker = CapacityTracker::new(capacity(2, 5));
        tracker.record_delivery();

        assert_eq!(tracker.pressure_signal(), PressureSignal::accept(1, 2));
        assert_eq!(tracker.current_in_flight(), 1);
        assert_eq!(tracker.current_buffer_depth(), 0);
    }

    #[test]
    fn pressure_signal_defers_when_in_flight_full_and_buffer_has_capacity() {
        let mut tracker = CapacityTracker::new(capacity(2, 5));
        tracker.record_delivery();
        tracker.record_delivery();
        tracker.record_buffered();
        tracker.record_buffered();
        tracker.record_buffered();

        assert_eq!(tracker.pressure_signal(), PressureSignal::defer(2, 2, 3, 5));
        assert_eq!(tracker.current_in_flight(), 2);
        assert_eq!(tracker.current_buffer_depth(), 3);
    }

    #[test]
    fn pressure_signal_rejects_when_in_flight_and_buffer_limits_are_reached() {
        let mut tracker = CapacityTracker::new(capacity(2, 5));
        tracker.record_delivery();
        tracker.record_delivery();
        tracker.record_buffered();
        tracker.record_buffered();
        tracker.record_buffered();
        tracker.record_buffered();
        tracker.record_buffered();

        assert_eq!(
            tracker.pressure_signal(),
            PressureSignal::reject(2, 2, 5, 5)
        );
        assert_eq!(tracker.current_in_flight(), 2);
        assert_eq!(tracker.current_buffer_depth(), 5);
    }

    #[test]
    fn pressure_signal_accepts_available_in_flight_regardless_of_buffer_state() {
        let mut tracker = CapacityTracker::new(capacity(1, 1));
        tracker.record_buffered();

        assert_eq!(tracker.pressure_signal(), PressureSignal::accept(0, 1));
        assert_eq!(tracker.current_in_flight(), 0);
        assert_eq!(tracker.current_buffer_depth(), 1);
    }

    #[test]
    fn pressure_root_re_exports_capacity_types() {
        use crate::pressure::{
            CapacityError as RootCapacityError, CapacityTracker as RootCapacityTracker,
            ConsumerCapacity as RootConsumerCapacity,
        };

        let mut tracker = RootCapacityTracker::new(RootConsumerCapacity {
            max_in_flight: 1,
            max_buffer_depth: 1,
        });

        assert_eq!(
            tracker.record_completion(),
            Err(RootCapacityError::InFlightUnderflow)
        );
    }
}