feat(net): response-body bounds — stall timeout + buffered size cap (ADR-0034 Am#13)

CHANGELOG.md

@@ -91,6 +91,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `http_circuit_breaker_transitions_total{to="open"}` metric gains a `reason` label
   (`rate`/`throttle`/`probe_failed`/`abandoned`). Prior art: tower-resilience (not
   adopted — OATH keeps its RPITIT `Service`).
+- **Breaking (pre-release) — net timing + config surface.** `Timer` gains an associated
+  `type Sleep: Future<Output=()> + Send` (was `fn sleep(&self) -> impl Future`), so body
+  wrappers store the sleep future inline without boxing. `HttpConfig` gains
+  `body_stall_timeout: Option<Duration>` and `ConnConfig` gains
+  `max_response_bytes: Option<usize>` (both new required fields); `HttpError`/`ErrorKind`
+  gain a `BodyTooLarge` variant.
 
 ### Added
 
@@ -259,6 +265,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `retry_after_fallback` default). `429` is still never retried. An `HTTP-date`,
   float, overflowing, or absent value falls back to existing behavior. A new
   site-labelled `http_retry_after_honored_total` metric. (ADR-0031 Amendment #2)
+- **net-http response-body bounds.** A streaming **stall timeout** (`StallTimeoutLayer`
+  / `TimeoutBody`, `HttpConfig.body_stall_timeout`) bounds a mid-transfer body
+  inactivity gap so a slow venue body can no longer wedge a concurrency permit; a
+  **buffered size cap** (`LimitedBody`, `ConnConfig::max_response_bytes`) rejects an
+  oversized `BufferMode::Buffer` body with the new non-retryable `HttpError::BodyTooLarge`
+  (`error_kind="body_too_large"`) before OOM. Both are `Option` (disable-able).
+  Implements ADR-0034 Amendment #13 (un-defers Am#6's `TimeoutBody`; wires §2's size guard).
 
 ### Fixed
 

crates/adapter/net/api/src/error_kind.rs

@@ -39,6 +39,11 @@ pub enum ErrorKind {
     /// cooldown elapses. A deliberate local decision, not a transport outcome;
     /// non-retryable.
     CircuitOpen,
+
+    /// A response body exceeded the configured maximum size and was rejected
+    /// before being fully buffered. A deliberate local decision, not a transport
+    /// outcome; non-retryable.
+    BodyTooLarge,
 }
 
 /// Implemented by error types that can be classified as an [`ErrorKind`].

crates/adapter/net/api/src/timer.rs

@@ -10,8 +10,13 @@ use std::time::{Duration, Instant};
 /// it deterministically in tests while production passes a runtime-backed impl.
 /// A trait — not a runtime — so the kernel stays std-only.
 pub trait Timer: Clone + Send + Sync {
+    /// The concrete future returned by [`sleep`](Timer::sleep). Named (not
+    /// `impl Future`) so body wrappers can store it inline in a `#[pin]` field
+    /// without boxing.
+    type Sleep: Future<Output = ()> + Send;
+
     /// Complete after `dur` has elapsed.
-    fn sleep(&self, dur: Duration) -> impl Future<Output = ()> + Send;
+    fn sleep(&self, dur: Duration) -> Self::Sleep;
 
     /// The current instant — for elapsed-time reads (token-bucket refill,
     /// circuit cooldown).
@@ -21,14 +26,14 @@ pub trait Timer: Clone + Send + Sync {
 #[cfg(test)]
 mod tests {
     use super::Timer;
-    use std::future::Future;
     use std::time::{Duration, Instant};
 
     #[derive(Clone)]
     struct FixedTimer(Instant);
 
     impl Timer for FixedTimer {
-        fn sleep(&self, _dur: Duration) -> impl Future<Output = ()> + Send {
+        type Sleep = std::future::Ready<()>;
+        fn sleep(&self, _dur: Duration) -> std::future::Ready<()> {
             std::future::ready(())
         }
         fn now(&self) -> Instant {

crates/adapter/net/http/api/src/body.rs

@@ -211,13 +211,105 @@ where
     }
 }
 
+pin_project_lite::pin_project! {
+    /// Wraps a response body, failing with [`HttpError::BodyTooLarge`] once the
+    /// cumulative DATA-frame bytes exceed `remaining`. A **typed** alternative to
+    /// `http_body_util::Limited` (which boxes its error): the whole HTTP stack
+    /// keeps one concrete `HttpError` for service *and* body. Forwards
+    /// `is_end_stream`; clamps `size_hint` to `remaining` (ADR-0034 §2), so a
+    /// downstream collector stays bounded.
+    pub struct LimitedBody<B> {
+        #[pin]
+        inner: B,
+        remaining: u64,
+    }
+}
+
+impl<B> LimitedBody<B> {
+    /// Wrap `inner`, rejecting once cumulative DATA bytes exceed `max_bytes`.
+    ///
+    /// # Example
+    /// ```
+    /// use oath_adapter_net_http_api::LimitedBody;
+    /// use http_body_util::Empty;
+    /// use bytes::Bytes;
+    ///
+    /// let _body = LimitedBody::new(Empty::<Bytes>::new(), 16 * 1024 * 1024);
+    /// ```
+    #[must_use]
+    pub const fn new(inner: B, max_bytes: u64) -> Self {
+        Self {
+            inner,
+            remaining: max_bytes,
+        }
+    }
+}
+
+impl<B: fmt::Debug> fmt::Debug for LimitedBody<B> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("LimitedBody")
+            .field("inner", &self.inner)
+            .field("remaining", &self.remaining)
+            .finish()
+    }
+}
+
+impl<B> Body for LimitedBody<B>
+where
+    B: Body<Data = Bytes, Error = HttpError>,
+{
+    type Data = Bytes;
+    type Error = HttpError;
+
+    fn poll_frame(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Bytes>, HttpError>>> {
+        let this = self.project();
+        match ready!(this.inner.poll_frame(cx)) {
+            Some(Ok(frame)) => {
+                if let Some(data) = frame.data_ref() {
+                    let len = data.len() as u64;
+                    if len > *this.remaining {
+                        *this.remaining = 0;
+                        return Poll::Ready(Some(Err(HttpError::BodyTooLarge)));
+                    }
+                    *this.remaining -= len;
+                }
+                Poll::Ready(Some(Ok(frame)))
+            },
+            // Terminal None or an inner error: pass through unchanged.
+            other => Poll::Ready(other),
+        }
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.inner.is_end_stream()
+    }
+
+    fn size_hint(&self) -> SizeHint {
+        // Mirror http_body_util::Limited's clamp so lower never exceeds upper.
+        let mut hint = self.inner.size_hint();
+        let n = self.remaining;
+        if hint.lower() >= n {
+            hint.set_exact(n);
+        } else if let Some(max) = hint.upper() {
+            hint.set_upper(n.min(max));
+        } else {
+            hint.set_upper(n);
+        }
+        hint
+    }
+}
+
 #[cfg(test)]
 mod tests {
-    use super::{BufferMode, Guarded, ResponseBody};
+    use super::{BufferMode, Guarded, LimitedBody, ResponseBody};
     use crate::HttpError;
     use async_lock::Semaphore;
     use bytes::Bytes;
     use http_body::{Body, Frame, SizeHint};
+    use http_body_util::BodyExt;
     use std::collections::VecDeque;
     use std::pin::{Pin, pin};
     use std::sync::Arc;
@@ -461,4 +553,136 @@ mod tests {
             "permit released on the error frame"
         );
     }
+
+    #[tokio::test]
+    async fn limited_body_passes_under_cap() {
+        let body = LimitedBody::new(Frames::new([b"ab", b"cde"]), 10);
+        let bytes = body.collect().await.unwrap().to_bytes();
+        assert_eq!(bytes, Bytes::from_static(b"abcde"));
+    }
+
+    #[tokio::test]
+    async fn limited_body_passes_at_exact_cap() {
+        // 2 + 3 == 5; each frame's len is not > remaining, so the boundary passes.
+        let body = LimitedBody::new(Frames::new([b"ab", b"cde"]), 5);
+        let bytes = body.collect().await.unwrap().to_bytes();
+        assert_eq!(bytes, Bytes::from_static(b"abcde"));
+    }
+
+    #[tokio::test]
+    async fn limited_body_errors_over_cap() {
+        // cap 4: "abc" (3) ok → remaining 1; "def" (3) > 1 → BodyTooLarge.
+        let body = LimitedBody::new(Frames::new([b"abc", b"def"]), 4);
+        let err = body.collect().await.expect_err("must overflow");
+        assert!(matches!(err, HttpError::BodyTooLarge));
+    }
+
+    #[test]
+    fn limited_body_clamps_size_hint_and_forwards_is_end_stream() {
+        // inner exact = 100, cap 10 → clamp to exact 10 (lower >= remaining path).
+        let wrapped = LimitedBody::new(Stub { remaining: 100 }, 10);
+        assert_eq!(wrapped.size_hint().exact(), Some(10));
+        assert!(!wrapped.is_end_stream());
+        let ended = LimitedBody::new(Stub { remaining: 0 }, 10);
+        assert!(ended.is_end_stream()); // forwarded
+    }
+
+    /// Inner body reporting a RANGE size hint (`lower != upper`), unlike
+    /// `Stub`'s exact hint — needed to exercise `LimitedBody::size_hint`'s
+    /// `lower < remaining` branches, which an exact hint can never reach.
+    struct RangeHint {
+        lower: u64,
+        upper: Option<u64>,
+    }
+
+    impl Body for RangeHint {
+        type Data = Bytes;
+        type Error = HttpError;
+        fn poll_frame(
+            self: Pin<&mut Self>,
+            _: &mut Context<'_>,
+        ) -> Poll<Option<Result<Frame<Bytes>, HttpError>>> {
+            Poll::Ready(None)
+        }
+        fn size_hint(&self) -> SizeHint {
+            let mut hint = SizeHint::default();
+            hint.set_lower(self.lower);
+            if let Some(upper) = self.upper {
+                hint.set_upper(upper);
+            }
+            hint
+        }
+    }
+
+    #[test]
+    fn limited_body_clamps_size_hint_upper_when_bounded_above_cap() {
+        // inner lower(2) < cap(10) and inner upper(50) > cap: the
+        // `lower < remaining` branch with `Some(max)` must clamp upper to
+        // `remaining.min(max)` == 10, not leave it at the inner's 50.
+        let wrapped = LimitedBody::new(
+            RangeHint {
+                lower: 2,
+                upper: Some(50),
+            },
+            10,
+        );
+        let hint = wrapped.size_hint();
+        assert_eq!(hint.upper(), Some(10));
+        assert!(hint.lower() <= 10);
+    }
+
+    #[test]
+    fn limited_body_clamps_size_hint_upper_when_inner_unbounded() {
+        // inner lower(2) < cap(10) and inner upper is `None` (unbounded): the
+        // final `else` branch must set upper to `remaining` == 10, not leave
+        // it unbounded.
+        let wrapped = LimitedBody::new(
+            RangeHint {
+                lower: 2,
+                upper: None,
+            },
+            10,
+        );
+        assert_eq!(wrapped.size_hint().upper(), Some(10));
+    }
+
+    /// Yields a trailers frame (no `data_ref()`), then ends. Used to prove
+    /// `LimitedBody` only counts DATA-frame bytes toward `remaining` — a
+    /// trailers frame must pass through without consuming budget or erroring.
+    struct TrailersOnly {
+        sent: bool,
+    }
+
+    impl Body for TrailersOnly {
+        type Data = Bytes;
+        type Error = HttpError;
+        fn poll_frame(
+            self: Pin<&mut Self>,
+            _: &mut Context<'_>,
+        ) -> Poll<Option<Result<Frame<Bytes>, HttpError>>> {
+            let this = self.get_mut();
+            if this.sent {
+                Poll::Ready(None)
+            } else {
+                this.sent = true;
+                Poll::Ready(Some(Ok(Frame::trailers(http::HeaderMap::new()))))
+            }
+        }
+        fn is_end_stream(&self) -> bool {
+            self.sent
+        }
+        fn size_hint(&self) -> SizeHint {
+            SizeHint::default()
+        }
+    }
+
+    #[tokio::test]
+    async fn limited_body_does_not_count_trailers_toward_cap() {
+        // cap 0: if the trailers' bytes were (wrongly) counted, any nonzero
+        // count would overflow a zero cap and this would error. A tiny cap
+        // with a non-DATA-only frame is the sharpest regression trap.
+        let body = LimitedBody::new(TrailersOnly { sent: false }, 0);
+        let collected = body.collect().await.expect("trailers must not error");
+        assert!(collected.trailers().is_some());
+    }
 }

crates/adapter/net/http/api/src/circuit_breaker.rs

@@ -636,6 +636,18 @@ mod classify_tests {
         assert_eq!(classify(&ok(429)), Class::TripNow);
     }
 
+    #[test]
+    fn body_too_large_is_breaker_neutral_not_a_failure() {
+        // `BodyTooLarge` is a purely LOCAL size-cap rejection — the request reached
+        // the host and a response was flowing, but the cap is a local policy
+        // decision, not a host-health signal — so it must be Ignored, never
+        // Failure, even if the classify catch-all above is ever narrowed.
+        assert_eq!(
+            classify::<()>(&Err(HttpError::BodyTooLarge)),
+            Class::Ignored
+        );
+    }
+
     #[test]
     fn client_errors_auth_and_unknown_are_ignored() {
         assert_eq!(classify(&ok(400)), Class::Ignored);

crates/adapter/net/http/api/src/error.rs

@@ -34,6 +34,9 @@ pub enum HttpError {
     /// The circuit breaker is open — the request was rejected without being sent.
     #[error("circuit open: request rejected without being sent")]
     CircuitOpen,
+    /// A response body exceeded the configured maximum buffered size.
+    #[error("response body exceeded the configured maximum")]
+    BodyTooLarge,
 }
 
 impl HttpError {
@@ -65,6 +68,7 @@ impl HasErrorKind for HttpError {
             Self::Auth(_) => ErrorKind::Auth,
             Self::Other(_) => ErrorKind::Unknown,
             Self::CircuitOpen => ErrorKind::CircuitOpen,
+            Self::BodyTooLarge => ErrorKind::BodyTooLarge,
         }
     }
 }
@@ -82,6 +86,7 @@ mod tests {
         assert_eq!(HttpError::auth("expired").kind(), ErrorKind::Auth);
         assert_eq!(HttpError::other("boom").kind(), ErrorKind::Unknown);
         assert_eq!(HttpError::CircuitOpen.kind(), ErrorKind::CircuitOpen);
+        assert_eq!(HttpError::BodyTooLarge.kind(), ErrorKind::BodyTooLarge);
     }
 
     #[test]

crates/adapter/net/http/api/src/lib.rs

@@ -45,11 +45,12 @@ pub mod retry;
 mod retry_after;
 pub mod service;
 pub mod stack;
+pub mod stall;
 pub mod timeout;
 pub mod trace;
 
 pub use auth::{Auth, AuthSource, NoAuth, SetHeaders};
-pub use body::{BufferMode, Guarded, ResponseBody};
+pub use body::{BufferMode, Guarded, LimitedBody, ResponseBody};
 pub use circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitBreakerLayer};
 pub use client::HttpClient;
 pub use error::{BoxError, HttpError};
@@ -62,5 +63,6 @@ pub use rate_limit::{RateLimit, RateLimitLayer, RateScope};
 pub use retry::{Retry, RetryConfig, RetryLayer, Retryable};
 pub use service::Service;
 pub use stack::{HttpConfig, stack};
+pub use stall::{StallTimeout, StallTimeoutLayer, TimeoutBody};
 pub use timeout::{RequestTimeout, Timeout, TimeoutLayer};
 pub use trace::{Tracing, TracingLayer};