Implement hard-cap memory-budget protection (8.3.4)

docs/adr-002-streaming-requests-and-shared-message-assembly.md

@@ -331,6 +331,25 @@ Precedence is:
   rather than an indefinite read stop, so in-flight assemblies can continue to
   make progress.
 
+#### Implementation decisions (2026-02-26)
+
+- Roadmap item `8.3.4` implements hard-cap connection abort as a
+  defence-in-depth safety net in the inbound read loop. When total buffered
+  bytes strictly exceed the aggregate cap
+  (`min(bytes_per_connection, bytes_in_flight)`), the connection is terminated
+  immediately with `InvalidData`, bypassing the `DeserFailureTracker` counter.
+- The hard-cap check is combined with the soft-limit check into a single
+  `evaluate_memory_pressure()` function in
+  `src/app/frame_handling/backpressure.rs`, returning a `MemoryPressureAction`
+  enum (`Continue`, `Pause`, or `Abort`). This keeps the inbound loop
+  (`src/app/inbound_handler.rs`) within the 400-line file limit.
+- The hard cap uses `>` (strictly exceeds) for the comparison, matching the
+  `check_aggregate_budgets` convention in `budget.rs`. The limit value itself
+  is permitted.
+- Under normal operation, per-frame budget enforcement (`8.3.2`) prevents
+  total buffered bytes from exceeding the limit. The hard cap catches the edge
+  case where this invariant is violated.
+
 #### Budget enforcement
 
 - Budgets MUST cover: bytes buffered per message, bytes buffered per

docs/generic-message-fragmentation-and-re-assembly-design.md

@@ -481,11 +481,13 @@ reads and assembly work. When a hard cap is exceeded, Wireframe aborts early,
 releases partial state, and surfaces `std::io::ErrorKind::InvalidData` from the
 inbound processing path.
 
-The current soft-limit implementation paces reads by inserting a short pause
-before polling the next inbound frame once buffered bytes reach 80% of the
-smaller aggregate budget (`bytes_per_connection` and `bytes_in_flight`). This
-keeps pressure visible without permanently stalling assemblies that still need
-additional frames to complete.
+The soft-limit implementation paces reads by inserting a short pause before
+polling the next inbound frame once buffered bytes reach 80% of the smaller
+aggregate budget (`bytes_per_connection` and `bytes_in_flight`). This keeps
+pressure visible without permanently stalling assemblies that still need
+additional frames to complete. If total buffered bytes strictly exceed the
+aggregate cap (100%), the connection is terminated immediately with
+`InvalidData` as a defence-in-depth safety net.
 
 If both transport fragmentation and `MessageAssembler` are enabled, the
 effective message cap is whichever guard triggers first. Operators should set

docs/roadmap.md

@@ -295,7 +295,7 @@ and standardized per-connection memory budgets.
 - [x] 8.3.2. Implement budget enforcement covering bytes per message, bytes
   per connection, and bytes across in-flight assemblies.
 - [x] 8.3.3. Implement soft limit (back-pressure by pausing reads) behaviour.
-- [ ] 8.3.4. Implement hard cap (abort early, release partial state, surface
+- [x] 8.3.4. Implement hard cap (abort early, release partial state, surface
   `InvalidData`) behaviour.
 - [ ] 8.3.5. Define derived defaults based on `buffer_capacity` when budgets
   are not set explicitly.

docs/users-guide.md

@@ -609,12 +609,22 @@ is the minimum of the fragmentation `max_message_size` and the configured
 `bytes_per_message`. Single-frame messages that complete immediately are never
 counted against aggregate budgets, since they do not buffer.
 
-Wireframe also applies soft-limit read pacing before hard-cap rejection. When
-buffered assembly bytes reach the soft-pressure threshold (80% of the smaller
-aggregate cap from `bytes_per_connection` and `bytes_in_flight`), the inbound
-connection loop briefly pauses socket reads before polling the next frame. This
-propagates back-pressure to senders while still allowing progress on in-flight
-assemblies.
+Wireframe provides a three-tier protection model for inbound memory budgets:
+
+1. **Per-frame enforcement** — frames that would cause total buffered bytes to
+   exceed the per-connection or in-flight budget are rejected, the offending
+   partial assembly is freed, and the failure is surfaced through the existing
+   deserialization-failure policy (`InvalidData`).
+2. **Soft-limit read pacing** — when buffered assembly bytes reach the
+   soft-pressure threshold (80% of the smaller aggregate cap from
+   `bytes_per_connection` and `bytes_in_flight`), the inbound connection loop
+   briefly pauses socket reads before polling the next frame. This propagates
+   back-pressure to senders while allowing progress on in-flight assemblies.
+3. **Hard-cap connection abort** — if total buffered bytes strictly exceed the
+   aggregate cap (100% of the smaller of `bytes_per_connection` and
+   `bytes_in_flight`), the connection is terminated immediately with
+   `InvalidData`. This is a defence-in-depth safety net; under normal
+   operation, per-frame enforcement prevents this state from being reached.
 
 #### Message key multiplexing (8.2.3)
 

src/app/frame_handling/backpressure.rs

@@ -1,11 +1,17 @@
-//! Soft-limit back-pressure helpers for inbound read pacing.
+//! Memory budget pressure helpers for inbound read pacing.
 //!
-//! These helpers detect when buffered assembly bytes approach configured
-//! aggregate memory budgets and provide a small pause duration that throttles
-//! subsequent socket reads.
+//! These helpers detect when buffered assembly bytes approach or exceed
+//! configured aggregate memory budgets. The module provides two tiers of
+//! protection:
+//!
+//! - **Soft limit** (80% of aggregate cap): paces reads with a short pause.
+//! - **Hard cap** (100% of aggregate cap): signals immediate connection abort.
 
 use std::time::Duration;
 
+use log::{debug, warn};
+use tokio::{io, time::sleep};
+
 use crate::{app::MemoryBudgets, message_assembler::MessageAssemblyState};
 
 /// Soft-pressure threshold numerator (4/5 == 80%).
@@ -15,9 +21,90 @@ const SOFT_LIMIT_DENOMINATOR: u128 = 5;
 /// Read-pacing delay applied while under soft budget pressure.
 const SOFT_LIMIT_PAUSE_DURATION: Duration = Duration::from_millis(5);
 
+/// Action to take based on current memory budget pressure.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum MemoryPressureAction {
+    /// No pressure; proceed normally.
+    Continue,
+    /// Soft pressure; pause reads briefly before continuing.
+    Pause(Duration),
+    /// Hard cap breached; abort the connection immediately.
+    Abort,
+}
+
+/// Evaluate memory budget pressure and return the appropriate action.
+///
+/// Checks the hard cap first (connection abort at 100% of aggregate limit),
+/// then the soft limit (read pacing at 80%). Returns `Continue` when no
+/// budgets are configured or buffered bytes are below both thresholds.
+#[must_use]
+pub(crate) fn evaluate_memory_pressure(
+    state: Option<&MessageAssemblyState>,
+    budgets: Option<MemoryBudgets>,
+) -> MemoryPressureAction {
+    if has_hard_cap_been_breached(state, budgets) {
+        return MemoryPressureAction::Abort;
+    }
+    if should_pause_inbound_reads(state, budgets) {
+        return MemoryPressureAction::Pause(SOFT_LIMIT_PAUSE_DURATION);
+    }
+    MemoryPressureAction::Continue
+}
+
+/// Act on the result of [`evaluate_memory_pressure`].
+///
+/// - `Abort`: logs a warning and returns `Err(InvalidData)`.
+/// - `Pause(d)`: logs at debug level, sleeps for `d`, then purges expired assemblies via the
+///   caller-supplied closure.
+/// - `Continue`: no-op.
+///
+/// # Errors
+///
+/// Returns an [`io::Error`] with kind `InvalidData` when the hard cap is
+/// breached.
+pub(crate) async fn apply_memory_pressure(
+    action: MemoryPressureAction,
+    mut purge: impl FnMut(),
+) -> io::Result<()> {
+    match action {
+        MemoryPressureAction::Abort => {
+            warn!("memory budget hard cap exceeded; aborting connection");
+            Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "per-connection memory budget hard cap exceeded",
+            ))
+        }
+        MemoryPressureAction::Pause(duration) => {
+            debug!("soft memory budget pressure; pausing inbound reads");
+            sleep(duration).await;
+            purge();
+            Ok(())
+        }
+        MemoryPressureAction::Continue => Ok(()),
+    }
+}
+
+/// Return `true` when buffered assembly bytes strictly exceed the aggregate
+/// budget cap, indicating the connection must be aborted immediately.
+///
+/// This is a defence-in-depth safety net. Under normal operation, per-frame
+/// budget enforcement (8.3.2) prevents the total from exceeding the limit.
+#[must_use]
+pub(super) fn has_hard_cap_been_breached(
+    state: Option<&MessageAssemblyState>,
+    budgets: Option<MemoryBudgets>,
+) -> bool {
+    let (Some(state), Some(budgets)) = (state, budgets) else {
+        return false;
+    };
+    let buffered_bytes = state.total_buffered_bytes();
+    let aggregate_limit = active_aggregate_limit_bytes(budgets);
+    buffered_bytes > aggregate_limit
+}
+
 /// Return `true` when inbound reads should be paced due to soft budget pressure.
 #[must_use]
-pub(crate) fn should_pause_inbound_reads(
+pub(super) fn should_pause_inbound_reads(
     state: Option<&MessageAssemblyState>,
     budgets: Option<MemoryBudgets>,
 ) -> bool {
@@ -30,10 +117,6 @@ pub(crate) fn should_pause_inbound_reads(
     is_at_or_above_soft_limit(buffered_bytes, aggregate_limit)
 }
 
-/// Duration to pause between inbound reads while soft pressure is active.
-#[must_use]
-pub(crate) const fn soft_limit_pause_duration() -> Duration { SOFT_LIMIT_PAUSE_DURATION }
-
 fn active_aggregate_limit_bytes(budgets: MemoryBudgets) -> usize {
     budgets
         .bytes_per_connection()