From 03fc3c8ec04a2ea16c9aa2c6fb9490d824e15da2 Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 08:10:01 +0000 Subject: [PATCH 01/10] docs(net): response-body bounds spec + implementation plan --- ...026-07-10-net-http-response-body-bounds.md | 1494 +++++++++++++++++ ...10-net-http-response-body-bounds-design.md | 349 ++++ 2 files changed, 1843 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-10-net-http-response-body-bounds.md create mode 100644 docs/superpowers/specs/2026-07-10-net-http-response-body-bounds-design.md diff --git a/docs/superpowers/plans/2026-07-10-net-http-response-body-bounds.md b/docs/superpowers/plans/2026-07-10-net-http-response-body-bounds.md new file mode 100644 index 0000000..1d3349b --- /dev/null +++ b/docs/superpowers/plans/2026-07-10-net-http-response-body-bounds.md @@ -0,0 +1,1494 @@ +# net-http response-body bounds — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Bound a misbehaving venue response body on both axes — a per-frame *stall* timeout for streaming bodies (frees a wedged concurrency permit) and a *max-bytes* cap for buffered bodies (prevents OOM) — landing ADR-0034 Amendment #13. + +**Architecture:** A new `TimeoutBody` + `StallTimeoutLayer` in `net-http-api` wraps streaming response bodies with a `Timer`-driven inactivity deadline, placed innermost in `stack()` so `Guarded` wraps it and a stall releases the permit. A new typed `LimitedBody` (also `net-http-api`) caps the hyper leaf's buffered `collect()`. To store the sleep future inline (no `Box`/`dyn`), the `Timer` trait gains `type Sleep`. Overflow is a new non-retryable `HttpError::BodyTooLarge`. + +**Tech Stack:** Rust 2024, `http-body` 1.x, `http-body-util`, `pin-project-lite`, `bytes`, the crate's runtime-neutral `Timer` seam + `MockTimer`, `tokio` (dev/test only). + +**Design source:** [docs/superpowers/specs/2026-07-10-net-http-response-body-bounds-design.md](../specs/2026-07-10-net-http-response-body-bounds-design.md). + +## Global Constraints + +- Edition **2024**, MSRV **1.90** (`just msrv`). +- **No `unsafe`** (`unsafe_code = "deny"`); use `pin_project_lite` for pin projection. +- **No `unwrap`/`expect`/indexing in non-test code** — return `Result`, model errors with `thiserror`. Test code is exempt. +- **Document every public item** (`missing_docs` warned) — rustdoc + a doctest on each new public item. +- **Respect dependency direction:** `oath-adapter-net-api` ← `oath-adapter-net-http-api` ← `oath-adapter-net-http-hyper`. Never introduce a cycle. +- **Clippy `all` is deny-level.** No new warnings. +- **Conventional Commits** (commit-msg hook). Breaking pre-release changes use `!`. +- **Definition of done:** `just ci` green (fmt, lint, test, **doc**, deny, typos). Work happens on a worktree branch under `.claude/worktrees/` off `main`; one PR `Closes #N`. +- **CHANGELOG.md `[Unreleased]`** updated (Task 7). + +--- + +## Task 1: `Timer::Sleep` associated type + +Behaviour-preserving refactor that makes each `Timer` impl's sleep future *nameable*, so `TimeoutBody` (Task 3) can store it inline. No new behaviour — the safety net is that every existing timing suite must stay green and Task 3 must compile. + +**Files:** +- Modify: `crates/adapter/net/api/src/timer.rs:12-19` (trait) + its `FixedTimer` test double (`:30-37`) +- Modify: `crates/adapter/net/http/hyper/src/timer.rs:15-18` (`TokioTimer`) +- Modify: `crates/adapter/net/mock/src/timer.rs:99-114` (`MockTimer`) + +**Interfaces:** +- Produces: `trait Timer { type Sleep: Future + Send; fn sleep(&self, dur: Duration) -> Self::Sleep; fn now(&self) -> Instant; }` + +- [ ] **Step 1: Add the associated type to the trait** + +In `crates/adapter/net/api/src/timer.rs`, change the trait: + +```rust +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 + Send; + + /// Complete after `dur` has elapsed. + fn sleep(&self, dur: Duration) -> Self::Sleep; + + /// The current instant — for elapsed-time reads (token-bucket refill, + /// circuit cooldown). + fn now(&self) -> Instant; +} +``` + +And update the `FixedTimer` test double in the same file's `mod tests`: + +```rust +impl Timer for FixedTimer { + type Sleep = std::future::Ready<()>; + fn sleep(&self, _dur: Duration) -> std::future::Ready<()> { + std::future::ready(()) + } + fn now(&self) -> Instant { + self.0 + } +} +``` + +- [ ] **Step 2: Update `TokioTimer`** + +In `crates/adapter/net/http/hyper/src/timer.rs`: + +```rust +impl Timer for TokioTimer { + type Sleep = tokio::time::Sleep; + fn sleep(&self, dur: Duration) -> tokio::time::Sleep { + tokio::time::sleep(dur) + } + + fn now(&self) -> Instant { + tokio::time::Instant::now().into_std() + } +} +``` + +Remove the now-unused `use std::future::Future;` if clippy flags it (it may still be referenced elsewhere in the file — only remove if `just lint` warns). + +- [ ] **Step 3: Update `MockTimer`** + +In `crates/adapter/net/mock/src/timer.rs`, change only the impl signature (the body already builds the named `Sleep` struct): + +```rust +impl Timer for MockTimer { + type Sleep = Sleep; + fn sleep(&self, dur: Duration) -> Sleep { + let deadline = { + let state = lock(&self.state); + state.now + dur + }; + Sleep { + state: Arc::clone(&self.state), + deadline, + } + } + + fn now(&self) -> Instant { + lock(&self.state).now + } +} +``` + +- [ ] **Step 4: Verify every existing timing suite still passes** + +Run: `cargo test -p oath-adapter-net-api -p oath-adapter-net-mock -p oath-adapter-net-http-api -p oath-adapter-net-http-hyper` +Expected: PASS — the trait change is behaviour-preserving; every `self.timer.sleep(dur).await` compiles and runs identically. + +- [ ] **Step 5: Lint + doc** + +Run: `just lint && just doc` +Expected: no warnings; rustdoc builds (the new `type Sleep` is documented). + +- [ ] **Step 6: Commit** + +```bash +git add crates/adapter/net/api/src/timer.rs crates/adapter/net/http/hyper/src/timer.rs crates/adapter/net/mock/src/timer.rs +git commit -m "refactor(net)!: add Timer::Sleep associated type for inline sleep storage" +``` + +--- + +## Task 2: `BodyTooLarge` error surface + +Add the error kind + variant + telemetry label, and pin its non-retryability. + +**Files:** +- Modify: `crates/adapter/net/api/src/error_kind.rs:14-42` (add `ErrorKind::BodyTooLarge`) +- Modify: `crates/adapter/net/http/api/src/error.rs:18-37,60-69` (variant + `HasErrorKind` arm) +- Modify: `crates/adapter/net/http/api/src/trace.rs:31-42` (label arm) + its test module +- Modify: `crates/adapter/net/http/api/src/retry.rs:410-418` (`err_of` arm) + a new test + +**Interfaces:** +- Produces: `ErrorKind::BodyTooLarge`; `HttpError::BodyTooLarge` (unit); `kind_label(ErrorKind::BodyTooLarge) == "body_too_large"`. + +- [ ] **Step 1: Write the failing label test** + +In `crates/adapter/net/http/api/src/trace.rs`, inside `mod tests` add: + +```rust +#[test] +fn body_too_large_has_a_stable_label() { + assert_eq!( + super::kind_label(oath_adapter_net_api::ErrorKind::BodyTooLarge), + "body_too_large" + ); +} +``` + +- [ ] **Step 2: Run it — expect a compile failure** + +Run: `cargo test -p oath-adapter-net-http-api trace::tests::body_too_large_has_a_stable_label` +Expected: FAIL to compile — `no variant named BodyTooLarge found for enum ErrorKind`. + +- [ ] **Step 3: Add `ErrorKind::BodyTooLarge`** + +In `crates/adapter/net/api/src/error_kind.rs`, add before the closing brace of the enum (after `CircuitOpen`): + +```rust + /// 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, +``` + +- [ ] **Step 4: Add `HttpError::BodyTooLarge` + its classification** + +In `crates/adapter/net/http/api/src/error.rs`, add the variant to the enum (after `CircuitOpen`): + +```rust + /// A response body exceeded the configured maximum buffered size. + #[error("response body exceeded the configured maximum")] + BodyTooLarge, +``` + +and add the arm to `HasErrorKind::kind`: + +```rust + Self::BodyTooLarge => ErrorKind::BodyTooLarge, +``` + +Extend the existing `kind_maps_each_variant` test in `error.rs`'s `mod tests`: + +```rust + assert_eq!(HttpError::BodyTooLarge.kind(), ErrorKind::BodyTooLarge); +``` + +- [ ] **Step 5: Add the `kind_label` arm** + +In `crates/adapter/net/http/api/src/trace.rs::kind_label`, add before the `_` arm: + +```rust + ErrorKind::BodyTooLarge => "body_too_large", +``` + +- [ ] **Step 6: Run the label + error tests — expect PASS** + +Run: `cargo test -p oath-adapter-net-http-api trace::tests::body_too_large_has_a_stable_label error::tests::kind_maps_each_variant` +Expected: PASS. + +- [ ] **Step 7: Write the failing non-retry test** + +In `crates/adapter/net/http/api/src/retry.rs`, inside `mod tests` add: + +```rust +#[tokio::test] +async fn body_too_large_is_not_retried() { + // BodyTooLarge is not a transient kind, so even an eligible request sends once. + let leaf = ScriptLeaf::new(vec![Step::Err(ErrorKind::BodyTooLarge)]); + let svc = RetryLayer::new( + cfg(3, Duration::from_millis(1), Duration::from_millis(1)), + MockTimer::new(), + ) + .layer(leaf.clone()); + let err = svc.call(req(true)).await.unwrap_err(); + assert!(matches!(err, HttpError::BodyTooLarge)); + assert_eq!(leaf.calls(), 1, "BodyTooLarge is non-transient → never retried"); +} +``` + +- [ ] **Step 8: Run it — expect a wrong-variant failure** + +Run: `cargo test -p oath-adapter-net-http-api retry::tests::body_too_large_is_not_retried` +Expected: FAIL — `err_of` maps the kind through its `_ => HttpError::other("boom")` fallback, so `matches!(err, HttpError::BodyTooLarge)` is false. + +- [ ] **Step 9: Map the kind in `err_of`** + +In `crates/adapter/net/http/api/src/retry.rs::err_of`, add before the `_` arm: + +```rust + ErrorKind::BodyTooLarge => HttpError::BodyTooLarge, +``` + +- [ ] **Step 10: Run it — expect PASS** + +Run: `cargo test -p oath-adapter-net-http-api retry::tests::body_too_large_is_not_retried` +Expected: PASS — `is_transient(BodyTooLarge)` is false (only `Timeout`/`Connection` are), so the leaf is hit exactly once. + +- [ ] **Step 11: Lint + doc + commit** + +Run: `just lint && just doc` + +```bash +git add crates/adapter/net/api/src/error_kind.rs crates/adapter/net/http/api/src/error.rs crates/adapter/net/http/api/src/trace.rs crates/adapter/net/http/api/src/retry.rs +git commit -m "feat(net): add BodyTooLarge error kind, variant, and label" +``` + +--- + +## Task 3: `TimeoutBody` + `StallTimeoutLayer` + +The stall body wrapper and its layer, plus the permit-release proof (`Guarded` outside `TimeoutBody`). + +**Files:** +- Create: `crates/adapter/net/http/api/src/stall.rs` +- Modify: `crates/adapter/net/http/api/src/lib.rs:48,65` (module decl + re-export) + +**Interfaces:** +- Consumes: `Timer::Sleep` (Task 1); `HttpError::Timeout` (existing); `Guarded` (existing, from `body.rs`). +- Produces: `TimeoutBody::::new(inner: B, timeout: Option, timer: T)`; `StallTimeoutLayer::::new(timeout: Option, timer: T)`; the layer's `Wrapped = StallTimeout` mapping `http::Response` → `http::Response>`. + +- [ ] **Step 1: Register the module + create the file skeleton with the failing test** + +In `crates/adapter/net/http/api/src/lib.rs`, add `pub mod stall;` (after `pub mod service;`) and `pub use stall::{StallTimeout, StallTimeoutLayer, TimeoutBody};` (after the `stack` re-export). + +Create `crates/adapter/net/http/api/src/stall.rs` with the module doc, imports, and **only** the first test (so we can watch it fail): + +```rust +//! The `StallTimeout` body-inactivity layer (ADR-0034 Amendment #13). +//! +//! Wraps a *streaming* response body with a per-frame **inactivity** deadline via +//! the runtime-neutral [`Timer`] seam: if no frame arrives within the configured +//! duration the body yields [`HttpError::Timeout`], so a stalled transfer can no +//! longer wedge a `Guarded` concurrency permit. Placed innermost in `stack()` so +//! `Guarded` wraps [`TimeoutBody`] and the stall error releases the permit. +//! Inert on buffered bodies (one ready frame) and when the deadline is `None`. + +use crate::{HttpError, Service}; +use bytes::Bytes; +use http_body::{Body, Frame, SizeHint}; +use oath_adapter_net_api::{Layer, Timer}; +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll, ready}; +use std::time::Duration; + +#[cfg(test)] +mod tests { + use super::TimeoutBody; + use crate::HttpError; + use bytes::Bytes; + use http_body::{Body, Frame, SizeHint}; + use oath_adapter_net_mock::MockTimer; + use std::collections::VecDeque; + use std::pin::{Pin, pin}; + use std::task::{Context, Poll, Waker}; + use std::time::Duration; + + // A body that never yields a frame — models a wedged/stalled transfer. + struct PendingBody; + impl Body for PendingBody { + type Data = Bytes; + type Error = HttpError; + fn poll_frame( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll, HttpError>>> { + Poll::Pending + } + fn is_end_stream(&self) -> bool { + false + } + fn size_hint(&self) -> SizeHint { + SizeHint::default() + } + } + + #[tokio::test] + async fn stall_fires_when_no_frame_arrives() { + let timer = MockTimer::new(); + let body = TimeoutBody::new(PendingBody, Some(Duration::from_secs(1)), timer.clone()); + let waiter = tokio::spawn(async move { + let mut body = pin!(body); + std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await + }); + tokio::task::yield_now().await; // arms the deadline, inner pends + timer.advance(Duration::from_secs(1)); // fire the stall deadline + let frame = waiter.await.unwrap(); + assert!( + matches!(frame, Some(Err(HttpError::Timeout))), + "a stalled body must yield Timeout" + ); + } +} +``` + +- [ ] **Step 2: Run it — expect a compile failure** + +Run: `cargo test -p oath-adapter-net-http-api stall::tests::stall_fires_when_no_frame_arrives` +Expected: FAIL to compile — `TimeoutBody` is not defined. + +- [ ] **Step 3: Implement `TimeoutBody`** + +Add to `stall.rs` (above the `#[cfg(test)]` module): + +```rust +pin_project_lite::pin_project! { + /// A response body wrapper enforcing a per-frame **inactivity** timeout. + /// + /// On each poll it (re-)arms a [`Timer::sleep`] deadline; if the deadline + /// fires before the inner body produces a frame, it yields + /// [`HttpError::Timeout`]. Each frame resets the deadline (lazily: `sleep` is + /// cleared and re-armed on the next poll). A `None` timeout is fully inert — + /// the wrapper forwards to `inner` unchanged. Forwards `is_end_stream` and + /// `size_hint` (ADR-0034 §2 transparency). + pub struct TimeoutBody { + #[pin] + inner: B, + #[pin] + sleep: Option, + timeout: Option, + timer: T, + } +} + +impl TimeoutBody { + /// Wrap `inner` with a stall timeout. `None` disables it (pass-through). + /// + /// # Example + /// ``` + /// use oath_adapter_net_http_api::TimeoutBody; + /// use oath_adapter_net_mock::MockTimer; + /// use http_body_util::Empty; + /// use bytes::Bytes; + /// use std::time::Duration; + /// + /// let _body = TimeoutBody::new( + /// Empty::::new(), + /// Some(Duration::from_secs(30)), + /// MockTimer::new(), + /// ); + /// ``` + #[must_use] + pub const fn new(inner: B, timeout: Option, timer: T) -> Self { + Self { + inner, + sleep: None, + timeout, + timer, + } + } +} + +impl fmt::Debug for TimeoutBody { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TimeoutBody") + .field("inner", &self.inner) + .field("timeout", &self.timeout) + .finish_non_exhaustive() + } +} + +impl Body for TimeoutBody +where + B: Body, + T: Timer, +{ + type Data = Bytes; + type Error = HttpError; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, HttpError>>> { + let mut this = self.project(); + let Some(dur) = *this.timeout else { + // Disabled: fully transparent. + return this.inner.poll_frame(cx); + }; + // Arm the deadline if unset. + if this.sleep.is_none() { + let nap = this.timer.sleep(dur); + this.sleep.set(Some(nap)); + } + // Poll the timer FIRST so its waker stays registered while the body pends. + if let Some(sleep) = this.sleep.as_mut().as_pin_mut() { + if sleep.poll(cx).is_ready() { + return Poll::Ready(Some(Err(HttpError::Timeout))); + } + } + let frame = ready!(this.inner.poll_frame(cx)); + this.sleep.set(None); // lazy per-frame reset (inactivity semantics) + Poll::Ready(frame) + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> SizeHint { + self.inner.size_hint() + } +} +``` + +- [ ] **Step 4: Run the stall test — expect PASS** + +Run: `cargo test -p oath-adapter-net-http-api stall::tests::stall_fires_when_no_frame_arrives` +Expected: PASS. + +- [ ] **Step 5: Add the reset / inert / transparency tests** + +Add to `stall.rs`'s `mod tests` (the `Frames`/`Stub` doubles mirror `body.rs`): + +```rust + // A multi-frame body; every frame is immediately ready. + struct Frames { + frames: VecDeque, + } + impl Frames { + fn new(frames: [&'static [u8]; N]) -> Self { + Self { + frames: frames.iter().copied().map(Bytes::from_static).collect(), + } + } + } + impl Body for Frames { + type Data = Bytes; + type Error = HttpError; + fn poll_frame( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll, HttpError>>> { + let this = self.get_mut(); + Poll::Ready(this.frames.pop_front().map(|d| Ok(Frame::data(d)))) + } + fn is_end_stream(&self) -> bool { + self.frames.is_empty() + } + fn size_hint(&self) -> SizeHint { + SizeHint::with_exact(self.frames.iter().map(|f| f.len() as u64).sum()) + } + } + + #[test] + fn each_frame_resets_the_deadline() { + // timeout = 1s; two 500ms gaps (1s total) must NOT trip, because each + // per-frame gap is < 1s. A no-reset impl (deadline armed once) WOULD trip + // once cumulative time reaches 1s. + let timer = MockTimer::new(); + let body = TimeoutBody::new(Frames::new([b"a", b"b"]), Some(Duration::from_secs(1)), timer.clone()); + let mut body = pin!(body); + let mut cx = Context::from_waker(Waker::noop()); + assert!(matches!(body.as_mut().poll_frame(&mut cx), Poll::Ready(Some(Ok(_))))); // frame a @ t=0 + timer.advance(Duration::from_millis(500)); + assert!(matches!(body.as_mut().poll_frame(&mut cx), Poll::Ready(Some(Ok(_))))); // frame b @ t=0.5s + timer.advance(Duration::from_millis(500)); // t=1.0s total + assert!(matches!(body.as_mut().poll_frame(&mut cx), Poll::Ready(None))); // end, NOT a stall + } + + #[tokio::test] + async fn none_timeout_is_inert() { + // A None timeout never arms a deadline: a pending body just stays pending + // even after the clock advances arbitrarily. + let timer = MockTimer::new(); + let body = TimeoutBody::new(PendingBody, None, timer.clone()); + let mut body = pin!(body); + let mut cx = Context::from_waker(Waker::noop()); + assert!(body.as_mut().poll_frame(&mut cx).is_pending()); + timer.advance(Duration::from_secs(3600)); + assert!( + body.as_mut().poll_frame(&mut cx).is_pending(), + "None disables the stall timeout entirely" + ); + } + + #[test] + fn forwards_is_end_stream_and_size_hint() { + let inner = Frames::new([b"ab", b"cde"]); + let ref_hint = inner.size_hint().exact(); + let wrapped = + TimeoutBody::new(Frames::new([b"ab", b"cde"]), Some(Duration::from_secs(1)), MockTimer::new()); + assert_eq!(wrapped.size_hint().exact(), ref_hint); // NOT silently unbounded + assert!(!wrapped.is_end_stream()); + let ended = + TimeoutBody::new(Frames::new([]), Some(Duration::from_secs(1)), MockTimer::new()); + assert!(ended.is_end_stream()); // forwarded, not the `false` default + } +``` + +- [ ] **Step 6: Run the new body tests — expect PASS** + +Run: `cargo test -p oath-adapter-net-http-api stall::tests` +Expected: PASS (all four body tests). + +- [ ] **Step 7: Add the permit-release test (the crux: `Guarded` outside `TimeoutBody`)** + +Add to `stall.rs`'s `mod tests` (add `use crate::Guarded;` and `use async_lock::Semaphore;` and `use std::sync::Arc;` to the test imports): + +```rust + // A stall must release the concurrency permit: with Guarded OUTSIDE TimeoutBody, + // the stall error frame is a non-Ok frame that Guarded observes and releases on. + #[expect( + clippy::significant_drop_tightening, + reason = "the test holds the guarded body across polls to prove release at the stall while still alive" + )] + #[test] + fn stall_releases_the_concurrency_permit() { + let sem = Arc::new(Semaphore::new(1)); + let permit = sem.try_acquire_arc().expect("permit free at start"); + let timer = MockTimer::new(); + let inner = TimeoutBody::new(PendingBody, Some(Duration::from_secs(1)), timer.clone()); + let body = Guarded::new(inner, Some(permit)); + let mut body = pin!(body); + let mut cx = Context::from_waker(Waker::noop()); + + assert!(body.as_mut().poll_frame(&mut cx).is_pending()); // arms deadline, permit held + assert!(sem.try_acquire_arc().is_none(), "permit held while streaming"); + + timer.advance(Duration::from_secs(1)); // fire the stall + assert!(matches!( + body.as_mut().poll_frame(&mut cx), + Poll::Ready(Some(Err(HttpError::Timeout))) + )); + assert!( + sem.try_acquire_arc().is_some(), + "the stall error must release the permit" + ); + } +``` + +- [ ] **Step 8: Run it — expect PASS** + +Run: `cargo test -p oath-adapter-net-http-api stall::tests::stall_releases_the_concurrency_permit` +Expected: PASS. + +- [ ] **Step 9: Implement `StallTimeoutLayer` + `StallTimeout` and a transparency test** + +Add to `stall.rs` (below `TimeoutBody`, above the tests): + +```rust +/// The [`StallTimeout`] [`Layer`] factory: holds the (optional) inactivity +/// deadline + clock and wraps any inner service's response body in +/// [`TimeoutBody`]. +pub struct StallTimeoutLayer { + timeout: Option, + timer: T, +} + +impl StallTimeoutLayer { + /// Build the layer. `None` disables the stall timeout (pass-through body). + /// + /// # Example + /// ``` + /// use oath_adapter_net_http_api::StallTimeoutLayer; + /// use oath_adapter_net_mock::MockTimer; + /// use std::time::Duration; + /// + /// let _layer = StallTimeoutLayer::new(Some(Duration::from_secs(30)), MockTimer::new()); + /// ``` + #[must_use] + pub const fn new(timeout: Option, timer: T) -> Self { + Self { timeout, timer } + } +} + +impl Clone for StallTimeoutLayer { + fn clone(&self) -> Self { + Self { + timeout: self.timeout, + timer: self.timer.clone(), + } + } +} + +impl fmt::Debug for StallTimeoutLayer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("StallTimeoutLayer") + .field("timeout", &self.timeout) + .finish_non_exhaustive() + } +} + +impl Layer for StallTimeoutLayer { + type Wrapped = StallTimeout; + fn layer(&self, inner: S) -> StallTimeout { + StallTimeout { + inner, + timeout: self.timeout, + timer: self.timer.clone(), + } + } +} + +/// The `StallTimeout` middleware: wraps the inner service's response body in a +/// [`TimeoutBody`]. The send itself is untouched (that is the `Timeout` layer's +/// job); only the streamed body gains the inactivity deadline. +pub struct StallTimeout { + inner: S, + timeout: Option, + timer: T, +} + +impl Clone for StallTimeout { + fn clone(&self) -> Self { + Self { + inner: self.inner.clone(), + timeout: self.timeout, + timer: self.timer.clone(), + } + } +} + +impl fmt::Debug for StallTimeout { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("StallTimeout") + .field("timeout", &self.timeout) + .finish_non_exhaustive() + } +} + +impl Service> for StallTimeout +where + S: Service, Response = http::Response, Error = HttpError> + Sync, + T: Timer, + B: Body, +{ + type Response = http::Response>; + type Error = HttpError; + + #[allow(clippy::manual_async_fn)] + fn call( + &self, + req: http::Request, + ) -> impl Future> + Send { + async move { + let resp = self.inner.call(req).await?; + let (parts, body) = resp.into_parts(); + let body = TimeoutBody::new(body, self.timeout, self.timer.clone()); + Ok(http::Response::from_parts(parts, body)) + } + } +} +``` + +Add a transparency test to `mod tests` (a fast leaf's body passes through the layer intact — mirror `timeout.rs`'s `FastLeaf`/`StubBody` doubles inline): + +```rust + use crate::Service; + use http_body_util::BodyExt; + use oath_adapter_net_api::Layer; + use std::future::Future; + + #[derive(Debug)] + struct StubBody { + data: Option, + } + impl Body for StubBody { + type Data = Bytes; + type Error = HttpError; + fn poll_frame( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll, HttpError>>> { + Poll::Ready(self.get_mut().data.take().map(|d| Ok(Frame::data(d)))) + } + fn is_end_stream(&self) -> bool { + self.data.is_none() + } + fn size_hint(&self) -> SizeHint { + self.data.as_ref().map_or_else( + || SizeHint::with_exact(0), + |d| SizeHint::with_exact(d.len() as u64), + ) + } + } + #[derive(Clone)] + struct FastLeaf; + impl Service> for FastLeaf { + type Response = http::Response; + type Error = HttpError; + fn call( + &self, + _req: http::Request, + ) -> impl Future> + Send { + async { + Ok(http::Response::new(StubBody { + data: Some(Bytes::from_static(b"ok")), + })) + } + } + } + + #[tokio::test] + async fn layer_wraps_and_body_streams_through() { + use super::StallTimeoutLayer; + let svc = + StallTimeoutLayer::new(Some(Duration::from_secs(30)), MockTimer::new()).layer(FastLeaf); + let resp = svc + .call(http::Request::new(Bytes::new())) + .await + .expect("fast leaf"); + let body = resp.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(body, Bytes::from_static(b"ok")); // wrapped, streamed through untouched + } +``` + +- [ ] **Step 10: Run the full stall suite + lint + doc** + +Run: `cargo test -p oath-adapter-net-http-api stall::tests && just lint && just doc` +Expected: PASS; no warnings; doctests for `TimeoutBody::new`/`StallTimeoutLayer::new` build and run. + +- [ ] **Step 11: Commit** + +```bash +git add crates/adapter/net/http/api/src/stall.rs crates/adapter/net/http/api/src/lib.rs +git commit -m "feat(net): TimeoutBody + StallTimeoutLayer for streaming body stall timeouts" +``` + +--- + +## Task 4: Wire `StallTimeoutLayer` into `stack()` + `body_stall_timeout` config + +**Files:** +- Modify: `crates/adapter/net/http/api/src/stack.rs` — `HttpConfig` field + `Debug` + `validate_config` + composition + doctest + `http_cfg` test helper + a full-stack test +- Modify: `crates/adapter/net/http/hyper/src/build.rs` — doctest `HttpConfig` + `http_cfg` test helper +- Modify: `crates/adapter/net/http/hyper/examples/client_with_directives.rs` — `HttpConfig` literal + +**Interfaces:** +- Consumes: `StallTimeoutLayer` (Task 3). +- Produces: `HttpConfig.body_stall_timeout: Option`; `stack()` composes `.layer(StallTimeoutLayer::new(cfg.body_stall_timeout, timer))` innermost; `BuildError::ZeroDuration("body_stall_timeout")` when `Some(ZERO)`. + +- [ ] **Step 1: Write the failing validation test** + +In `crates/adapter/net/http/api/src/stack.rs`'s `mod tests`, add: + +```rust +#[test] +fn zero_body_stall_timeout_is_rejected_at_build() { + let timer = MockTimer::new(); + let leaf = ScriptLeaf::new(timer.clone(), vec![Step::Status(200)]); + let mut cfg = http_cfg(1, Duration::from_secs(1), Duration::ZERO); + cfg.body_stall_timeout = Some(Duration::ZERO); + let Err(err) = stack(leaf, cfg, timer, NoAuth, rate_cfg()) else { + panic!("Some(ZERO) body_stall_timeout must be a BuildError"); + }; + assert_eq!(err, BuildError::ZeroDuration("body_stall_timeout")); +} +``` + +- [ ] **Step 2: Run it — expect a compile failure** + +Run: `cargo test -p oath-adapter-net-http-api stack::tests::zero_body_stall_timeout_is_rejected_at_build` +Expected: FAIL to compile — `HttpConfig` has no field `body_stall_timeout`. + +- [ ] **Step 3: Add the `HttpConfig` field + `Debug` + validation** + +In `crates/adapter/net/http/api/src/stack.rs`, add to `HttpConfig` (after `rate_limit_max_wait`): + +```rust + /// Per-frame inactivity deadline for a **streaming** response body; `None` + /// disables it. Bounds a mid-transfer stall so a slow body cannot pin a + /// concurrency permit indefinitely (ADR-0034 Amendment #13). Inert on + /// buffered responses. + pub body_stall_timeout: Option, +``` + +Add to the manual `Debug` impl (after the `rate_limit_max_wait` field): + +```rust + .field("body_stall_timeout", &self.body_stall_timeout) +``` + +Add to `validate_config` (after the `retry_after_cap` zero check, before the threshold checks): + +```rust + if cfg.body_stall_timeout == Some(Duration::ZERO) { + return Err(BuildError::ZeroDuration("body_stall_timeout")); + } +``` + +- [ ] **Step 4: Fix the two in-crate `HttpConfig` constructors** + +In `stack.rs`, add `body_stall_timeout: Some(Duration::from_secs(30)),` to the `HttpConfig` in the `stack` doctest (the `let cfg = HttpConfig { … }` block) and to the `http_cfg` test helper's `HttpConfig`. + +- [ ] **Step 5: Run the validation test — expect PASS** + +Run: `cargo test -p oath-adapter-net-http-api stack::tests::zero_body_stall_timeout_is_rejected_at_build` +Expected: PASS. + +- [ ] **Step 6: Write the failing full-stack stall test** + +In `stack.rs`'s `mod tests`, add a streaming stalling leaf + test. First extend the `Step` enum with a `StreamStall` variant and its `ScriptLeaf` arm: + +In the `enum Step` add `StreamStall,` and in `ScriptLeaf::call`'s `match step` add: + +```rust + Step::StreamStall => Ok(http::Response::new(StubBody { data: None }.pending())), +``` + +That requires a pending body; simpler — add a dedicated streaming-pending body double and a `Step::StreamStall` arm returning it. Add this body to `mod tests`: + +```rust + // A streaming body that never yields a frame (models a mid-transfer stall). + #[derive(Debug)] + struct StallingBody; + impl Body for StallingBody { + type Data = Bytes; + type Error = HttpError; + fn poll_frame( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll, HttpError>>> { + Poll::Pending + } + fn is_end_stream(&self) -> bool { + false + } + fn size_hint(&self) -> SizeHint { + SizeHint::default() + } + } +``` + +Because `ScriptLeaf::Response` is `http::Response`, the stall leaf needs its own type. Add a separate minimal leaf rather than overloading `ScriptLeaf`: + +```rust + #[derive(Clone)] + struct StallLeaf; + impl Service> for StallLeaf { + type Response = http::Response; + type Error = HttpError; + fn call( + &self, + _req: http::Request, + ) -> impl Future> + Send { + async { Ok(http::Response::new(StallingBody)) } + } + } + + #[tokio::test] + async fn a_stalled_streaming_body_times_out_through_the_stack() { + let timer = MockTimer::new(); + let mut cfg = http_cfg(1, Duration::from_secs(3600), Duration::ZERO); + cfg.body_stall_timeout = Some(Duration::from_secs(1)); // short body-stall bound + let svc = stack(StallLeaf, cfg, timer.clone(), NoAuth, rate_cfg()).expect("total config"); + + // The send returns at headers immediately; the stall bites while draining. + let resp = svc.call(req(RateScope::Global)).await.expect("headers arrive"); + let waiter = tokio::spawn(async move { + let mut body = std::pin::pin!(resp.into_body()); + std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await + }); + tokio::task::yield_now().await; // body registers the stall deadline + timer.advance(Duration::from_secs(1)); // fire it + let frame = waiter.await.unwrap(); + assert!( + matches!(frame, Some(Err(HttpError::Timeout))), + "a stalled streaming body must surface Timeout when drained" + ); + } +``` + +(Remove the abandoned `Step::StreamStall` sketch — the standalone `StallLeaf` is cleaner. Ensure `mod tests` imports `http_body::{Body, Frame, SizeHint}` and `std::task::{Context, Poll}` and `std::pin::Pin`; they are already imported for the existing `StubBody`.) + +- [ ] **Step 7: Run it — expect a compile failure (StallTimeoutLayer not yet wired)** + +Run: `cargo test -p oath-adapter-net-http-api stack::tests::a_stalled_streaming_body_times_out_through_the_stack` +Expected: FAIL — the stall layer is not composed, so the body is `Guarded` (never times out; the spawned poll hangs and the test would not observe `Timeout`). It may compile but hang/deadlock, or fail the assertion after the advance. Confirm it does **not** pass. + +- [ ] **Step 8: Wire `StallTimeoutLayer` into `stack()`** + +In `stack.rs`, add `StallTimeoutLayer` to the `use crate::{…}` import list, and change the `LayerBuilder` chain so the stall layer is innermost: + +```rust + let svc = LayerBuilder::new() + .layer(TracingLayer::new(timer.clone())) // outermost + .layer(CircuitBreakerLayer::new(cfg.circuit_breaker, timer.clone())) + .layer(RetryLayer::new(cfg.retry, timer.clone())) + .layer(rate) + .layer(TimeoutLayer::new(cfg.timeout, timer.clone())) + .layer(StallTimeoutLayer::new(cfg.body_stall_timeout, timer)) // innermost + .wrap(inner); +``` + +(Note the `TimeoutLayer::new(cfg.timeout, timer.clone())` now clones; the final `timer` moves into `StallTimeoutLayer`.) + +- [ ] **Step 9: Run the full-stack stall test + the whole stack suite — expect PASS** + +Run: `cargo test -p oath-adapter-net-http-api stack::tests` +Expected: PASS — including the existing ordering/permit tests (the new innermost layer is body-transparent to the send path). + +- [ ] **Step 10: Fix the remaining `HttpConfig` construction sites** + +Add `body_stall_timeout: Some(Duration::from_secs(30)),` to: +- `crates/adapter/net/http/hyper/src/build.rs` — the `build` doctest's `HttpConfig` literal and the `http_cfg()` test helper. +- `crates/adapter/net/http/hyper/examples/client_with_directives.rs` — the `HttpConfig` literal. + +- [ ] **Step 11: Full verification + commit** + +Run: `just check && just lint && just test && just doc` +Expected: PASS everywhere (all crates compile with the new required field; doctests updated). + +```bash +git add crates/adapter/net/http/api/src/stack.rs crates/adapter/net/http/hyper/src/build.rs crates/adapter/net/http/hyper/examples/client_with_directives.rs +git commit -m "feat(net)!: wire StallTimeoutLayer into stack() + HttpConfig.body_stall_timeout" +``` + +--- + +## Task 5: `LimitedBody` typed max-bytes wrapper + +**Files:** +- Modify: `crates/adapter/net/http/api/src/body.rs` (add `LimitedBody` + tests) +- Modify: `crates/adapter/net/http/api/src/lib.rs:52` (re-export) + +**Interfaces:** +- Consumes: `HttpError::BodyTooLarge` (Task 2). +- Produces: `LimitedBody::::new(inner: B, max_bytes: u64)`; a `Body` that emits `HttpError::BodyTooLarge` once cumulative DATA bytes exceed `max_bytes`. + +- [ ] **Step 1: Write the failing tests** + +In `crates/adapter/net/http/api/src/body.rs`'s `mod tests`, add (`Frames` and `Stub` doubles already exist in this module; add `use http_body_util::BodyExt;` to the test imports): + +```rust + #[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 + } +``` + +- [ ] **Step 2: Run them — expect a compile failure** + +Run: `cargo test -p oath-adapter-net-http-api body::tests::limited_body_passes_under_cap` +Expected: FAIL to compile — `LimitedBody` is not defined. + +- [ ] **Step 3: Implement `LimitedBody`** + +In `body.rs`, add (near `Guarded`; the file already imports `Body, Frame, SizeHint`, `Bytes`, `ready`, `Pin`, `Context`, `Poll`): + +```rust +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 { + #[pin] + inner: B, + remaining: u64, + } +} + +impl LimitedBody { + /// 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::::new(), 16 * 1024 * 1024); + /// ``` + #[must_use] + pub const fn new(inner: B, max_bytes: u64) -> Self { + Self { + inner, + remaining: max_bytes, + } + } +} + +impl fmt::Debug for LimitedBody { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("LimitedBody") + .field("inner", &self.inner) + .field("remaining", &self.remaining) + .finish() + } +} + +impl Body for LimitedBody +where + B: Body, +{ + type Data = Bytes; + type Error = HttpError; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, 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 + } +} +``` + +Add `LimitedBody` to the `body::` re-export in `lib.rs`: `pub use body::{BufferMode, Guarded, LimitedBody, ResponseBody};`. + +- [ ] **Step 4: Run the tests — expect PASS** + +Run: `cargo test -p oath-adapter-net-http-api body::tests` +Expected: PASS (all four `limited_body_*` tests + the existing body tests). + +- [ ] **Step 5: Lint + doc + commit** + +Run: `just lint && just doc` + +```bash +git add crates/adapter/net/http/api/src/body.rs crates/adapter/net/http/api/src/lib.rs +git commit -m "feat(net): LimitedBody — typed max-bytes response-body wrapper" +``` + +--- + +## Task 6: Cap the leaf's buffered collect (`ConnConfig::max_response_bytes`) + +**Files:** +- Modify: `crates/adapter/net/http/hyper/src/leaf.rs` — `ConnConfig` field, `HyperLeaf` carries the cap, `Buffer` arm, `test_conn`, new server doubles + tests +- Modify: `crates/adapter/net/http/hyper/src/build.rs` — doctest `ConnConfig` + `conn()` test helper +- Modify: `crates/adapter/net/http/hyper/examples/client_with_directives.rs` — `ConnConfig` literal + +**Interfaces:** +- Consumes: `LimitedBody` (Task 5), `HttpError::BodyTooLarge` (Task 2). +- Produces: `ConnConfig.max_response_bytes: Option`; `BufferMode::Buffer` responses are capped. + +- [ ] **Step 1: Write the failing "honest oversized Content-Length" test** + +In `crates/adapter/net/http/hyper/src/leaf.rs`'s `mod tests`, add a fixed-body server + the test (`spawn_echo_server` already exists; add a sized variant): + +```rust + // Serves a body of `n` bytes with an explicit Content-Length (via Full). + async fn spawn_body_server(n: usize) -> String { + let payload = Bytes::from(vec![b'x'; n]); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + loop { + let (stream, _) = listener.accept().await.unwrap(); + let payload = payload.clone(); + let io = hyper_util::rt::TokioIo::new(stream); + tokio::spawn(async move { + let svc = hyper::service::service_fn(move |_r| { + let payload = payload.clone(); + async move { + Ok::<_, Infallible>(hyper::Response::new(http_body_util::Full::new( + payload, + ))) + } + }); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(io, svc) + .await; + }); + } + }); + format!("http://{addr}") + } + + #[tokio::test] + async fn buffer_rejects_an_oversized_content_length_upfront() { + let base = spawn_body_server(64).await; + let conn = ConnConfig { + max_response_bytes: Some(16), + ..test_conn() + }; + let leaf = hyper_leaf(conn); + let mut req = http::Request::get(format!("{base}/big")) + .body(Bytes::new()) + .unwrap(); + req.extensions_mut().insert(BufferMode::Buffer); + let err = leaf.call(req).await.expect_err("64-byte body over a 16-byte cap"); + assert!( + matches!(err, oath_adapter_net_http_api::HttpError::BodyTooLarge), + "expected BodyTooLarge, got {err:?}" + ); + } +``` + +- [ ] **Step 2: Run it — expect a compile failure** + +Run: `cargo test -p oath-adapter-net-http-hyper leaf::tests::buffer_rejects_an_oversized_content_length_upfront` +Expected: FAIL to compile — `ConnConfig` has no field `max_response_bytes`. + +- [ ] **Step 3: Add the `ConnConfig` field + carry it on `HyperLeaf`** + +In `leaf.rs`, add to `ConnConfig` (after `http2_keep_alive_while_idle`): + +```rust + /// Maximum bytes to buffer for a `BufferMode::Buffer` response; `None` = + /// unbounded. Rejects an oversized body with [`HttpError::BodyTooLarge`] + /// (ADR-0034 Amendment #13) — memory-safety for a misbehaving venue. + pub max_response_bytes: Option, +``` + +Add the field to `HyperLeaf` and thread it through `hyper_leaf`: + +```rust +#[derive(Clone)] +pub struct HyperLeaf { + client: Client, Full>, + inflight: Arc, + max_response_bytes: Option, +} +``` + +In `hyper_leaf`, capture the cap before `conn` is moved and set it on the returned leaf: + +```rust +pub fn hyper_leaf(conn: ConnConfig) -> HyperLeaf { + let max_response_bytes = conn.max_response_bytes; + // … existing connector/client construction (unchanged) … + HyperLeaf { + client, + inflight: Arc::new(InFlight::default()), + max_response_bytes, + } +} +``` + +- [ ] **Step 4: Apply the cap in the `Buffer` arm** + +In `HyperLeaf::call`, replace the `BufferMode::Buffer` arm. First bind the cap before `async move` (it borrows `&self`): + +```rust + let guard = InFlightGuard::enter(&self.inflight); + let max_response_bytes = self.max_response_bytes; + async move { + let _guard = guard; + // … unchanged: mode, into_parts, request, map_legacy_err, into_parts … + let body = match mode { + BufferMode::Stream => { + let mapper: fn(hyper::Error) -> HttpError = map_hyper_err; + ResponseBody::streaming(incoming.map_err(mapper)) + }, + BufferMode::Buffer => { + let bytes = match max_response_bytes { + Some(cap) => { + let cap = cap as u64; + if incoming.size_hint().upper().is_some_and(|u| u > cap) { + return Err(HttpError::BodyTooLarge); + } + LimitedBody::new(incoming.map_err(map_hyper_err), cap) + .collect() + .await? + .to_bytes() + }, + None => incoming.collect().await.map_err(map_hyper_err)?.to_bytes(), + }; + ResponseBody::buffered(bytes) + }, + }; + Ok(http::Response::from_parts(parts, body)) + } +``` + +Add imports at the top of `leaf.rs`: `LimitedBody` to the `oath_adapter_net_http_api::{…}` use, and `http_body::Body` for `size_hint()` (add `use http_body::Body;`). `BodyExt::collect` is already available via `http_body_util::BodyExt` (add it to the `use http_body_util::{…}` line). + +- [ ] **Step 5: Add `max_response_bytes` to `test_conn` and run the upfront test — expect PASS** + +Add `max_response_bytes: None,` to `test_conn()` in `leaf.rs` (default unbounded for the existing tests), then set `Some(16)` in the new test's override (already written in Step 1). + +Run: `cargo test -p oath-adapter-net-http-hyper leaf::tests::buffer_rejects_an_oversized_content_length_upfront` +Expected: PASS. + +- [ ] **Step 6: Add the streaming-overflow (unsized/chunked) test** + +Add a chunked server (no Content-Length → the upfront check passes; `LimitedBody` catches it while collecting): + +```rust + // Sends a chunked (no Content-Length) body of two `chunk`-sized pieces. + async fn spawn_chunked_server(chunk: &'static [u8], count: usize) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = Vec::new(); + let mut tmp = [0_u8; 256]; + loop { + let n = stream.read(&mut tmp).await.unwrap(); + assert!(n > 0, "peer closed before a full request head"); + buf.extend_from_slice(&tmp[..n]); + if buf.windows(4).any(|w| w == b"\r\n\r\n") { + break; + } + } + stream + .write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n") + .await + .unwrap(); + for _ in 0..count { + stream + .write_all(format!("{:x}\r\n", chunk.len()).as_bytes()) + .await + .unwrap(); + stream.write_all(chunk).await.unwrap(); + stream.write_all(b"\r\n").await.unwrap(); + } + stream.write_all(b"0\r\n\r\n").await.unwrap(); + stream.flush().await.unwrap(); + }); + format!("http://{addr}") + } + + #[tokio::test] + async fn buffer_caps_an_unsized_streaming_body_while_collecting() { + // Two 8-byte chunks = 16 bytes, no Content-Length, over a 10-byte cap. + let base = spawn_chunked_server(b"12345678", 2).await; + let conn = ConnConfig { + max_response_bytes: Some(10), + ..test_conn() + }; + let leaf = hyper_leaf(conn); + let mut req = http::Request::get(format!("{base}/chunked")) + .body(Bytes::new()) + .unwrap(); + req.extensions_mut().insert(BufferMode::Buffer); + let err = leaf + .call(req) + .await + .expect_err("16 chunked bytes over a 10-byte cap"); + assert!( + matches!(err, oath_adapter_net_http_api::HttpError::BodyTooLarge), + "expected BodyTooLarge from the streaming cap, got {err:?}" + ); + } +``` + +- [ ] **Step 7: Add the under-cap and `None`-unbounded tests** + +```rust + #[tokio::test] + async fn buffer_under_cap_collects_normally() { + let base = spawn_body_server(8).await; + let conn = ConnConfig { + max_response_bytes: Some(1024), + ..test_conn() + }; + let leaf = hyper_leaf(conn); + let mut req = http::Request::get(format!("{base}/small")) + .body(Bytes::new()) + .unwrap(); + req.extensions_mut().insert(BufferMode::Buffer); + let resp = leaf.call(req).await.expect("8-byte body under a 1KiB cap"); + let body = resp.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(body.len(), 8); + } + + #[tokio::test] + async fn buffer_with_no_cap_is_unbounded() { + let base = spawn_body_server(64).await; + let conn = ConnConfig { + max_response_bytes: None, + ..test_conn() + }; + let leaf = hyper_leaf(conn); + let mut req = http::Request::get(format!("{base}/big")) + .body(Bytes::new()) + .unwrap(); + req.extensions_mut().insert(BufferMode::Buffer); + let resp = leaf.call(req).await.expect("no cap → unbounded"); + let body = resp.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(body.len(), 64); + } +``` + +- [ ] **Step 8: Run the leaf cap suite — expect PASS** + +Run: `cargo test -p oath-adapter-net-http-hyper leaf::tests` +Expected: PASS (upfront reject, chunked streaming cap, under-cap, no-cap, plus all existing leaf tests). + +- [ ] **Step 9: Fix the remaining `ConnConfig` construction sites** + +Add `max_response_bytes: Some(16 * 1024 * 1024),` to: +- `build.rs` — the `build` doctest's `ConnConfig` literal and the `conn()` test helper. +- `examples/client_with_directives.rs` — the `ConnConfig` literal. + +- [ ] **Step 10: Full verification + commit** + +Run: `just check && just lint && just test && just doc` +Expected: PASS everywhere. + +```bash +git add crates/adapter/net/http/hyper/src/leaf.rs crates/adapter/net/http/hyper/src/build.rs crates/adapter/net/http/hyper/examples/client_with_directives.rs +git commit -m "feat(net)!: cap buffered response collect via ConnConfig::max_response_bytes" +``` + +--- + +## Task 7: ADR-0034 Amendment #13 + CHANGELOG + +**Files:** +- Modify: `docs/adr/0034-http-construction-surface-auth-guarded-boot-coverage.md` (append Amendment #13) +- Modify: `CHANGELOG.md` (`[Unreleased]`) +- Move: `docs/superpowers/specs/2026-07-10-…` + `docs/superpowers/plans/2026-07-10-…` are committed with this branch (already on disk). + +- [ ] **Step 1: Append ADR-0034 Amendment #13** + +After Amendment #12 in `docs/adr/0034-…md`, add: + +```markdown +13. **Response-body bounds — un-defers Am#6's `TimeoutBody`; wires §2's size guard.** + The streaming mid-stream-stall `TimeoutBody` deferred in Amendment #6 lands as a + `StallTimeoutLayer` (innermost, inside `RateLimit` so `Guarded` wraps it): a + per-frame **inactivity** timeout via the `Timer` seam, `HttpError::Timeout` on + stall, inert on buffered bodies and when disabled. To store the sleep future + inline (no `Box`/`dyn`), `Timer` gains `type Sleep: Future + Send`. + Independently, `BufferMode::Buffer`'s collect is **capped** + (`ConnConfig::max_response_bytes`) via a typed `LimitedBody` wrapper plus a + `size_hint().upper()` fast-fail, completing the max-size guard §2's + wrapper-transparency was built to support (N1); overflow is a new non-retryable + `HttpError::BodyTooLarge` (`ErrorKind::BodyTooLarge`, `error_kind="body_too_large"`). + Both config values are `Option` (disable-able); `HttpConfig.body_stall_timeout` + is `validate_config`-checked non-zero when `Some`. Cross-refs: ADR-0031 §1 + (`Timeout`), ADR-0030 §4 (buffering). +``` + +- [ ] **Step 2: Add the CHANGELOG entry** + +In `CHANGELOG.md`, under `## [Unreleased]` → `### Added` (create the subsection if absent) and `### Changed`: + +```markdown +### Added + +- **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). +``` + +Add to `### Changed` (a breaking pre-release note): + +```markdown +- **Breaking (pre-release) — net timing + config surface.** `Timer` gains an associated + `type Sleep: Future + Send` (was `fn sleep(&self) -> impl Future`), so body + wrappers store the sleep future inline without boxing. `HttpConfig` gains + `body_stall_timeout: Option` and `ConnConfig` gains + `max_response_bytes: Option` (both new required fields); `HttpError`/`ErrorKind` + gain a `BodyTooLarge` variant. +``` + +- [ ] **Step 3: Full CI gate** + +Run: `just ci` +Expected: PASS — fmt, lint, test, **doc**, deny, typos all green. + +- [ ] **Step 4: Commit** + +```bash +git add docs/adr/0034-http-construction-surface-auth-guarded-boot-coverage.md CHANGELOG.md docs/superpowers/specs/2026-07-10-net-http-response-body-bounds-design.md docs/superpowers/plans/2026-07-10-net-http-response-body-bounds.md +git commit -m "docs(net): ADR-0034 Am#13 + CHANGELOG for response-body bounds" +``` + +- [ ] **Step 5: Open the PR** + +```bash +gh pr create --title "feat(net): response-body bounds — stall timeout + buffered size cap (ADR-0034 Am#13)" --body "Closes #. Implements the Tier-2 'Streaming stall TimeoutBody' item (#102) + the untracked N1 buffer cap. See docs/superpowers/specs/2026-07-10-net-http-response-body-bounds-design.md." +``` + +Then tick the **Streaming stall TimeoutBody** box in issue #102. + +--- + +## Self-Review + +**Spec coverage** — every spec section maps to a task: +- §4 `Timer::Sleep` → Task 1. §7 error surface → Task 2. §5 `TimeoutBody`/`StallTimeoutLayer` → Task 3. §9 stack placement + §8 `body_stall_timeout` → Task 4. §6 `LimitedBody` → Task 5. §8 `max_response_bytes` + leaf cap → Task 6. §10 ADR + §12 CHANGELOG → Task 7. §11 tests distributed across Tasks 2–6. +- Permit-release (§9 crux) → Task 3 Step 7 (unit) + Task 4 Step 6 (full-stack stall). + +**Placeholder scan** — no "TBD"/"handle errors"/"similar to". One spot to watch: Task 4 Step 6 first sketches a `Step::StreamStall` then discards it for a standalone `StallLeaf`; the executor uses `StallLeaf` (the sketch is explicitly retracted in-step). + +**Type consistency** — `TimeoutBody::new(inner, Option, timer)`, `StallTimeoutLayer::new(Option, timer)`, `LimitedBody::new(inner, u64)`, `HttpConfig.body_stall_timeout: Option`, `ConnConfig.max_response_bytes: Option`, `BuildError::ZeroDuration("body_stall_timeout")`, `HttpError::BodyTooLarge`, `ErrorKind::BodyTooLarge`, label `"body_too_large"` — used consistently across tasks. `LimitedBody` stores `u64`; the leaf converts `cap as usize→u64` at the call site. + +**Note on defaults:** `30s` / `16 MiB` are the spec's proposals — swap in the review if IBKR data differs (config values only; no code change). diff --git a/docs/superpowers/specs/2026-07-10-net-http-response-body-bounds-design.md b/docs/superpowers/specs/2026-07-10-net-http-response-body-bounds-design.md new file mode 100644 index 0000000..55e414e --- /dev/null +++ b/docs/superpowers/specs/2026-07-10-net-http-response-body-bounds-design.md @@ -0,0 +1,349 @@ +# net-http response-body bounds — stall `TimeoutBody` + buffered size cap (2026-07-10) + +Design for the next **Tier-2** net-http hardening item (tracking issue #102): bound a +misbehaving venue **response body** on its two independent axes — **time** and **memory** — +so a slow or oversized body cannot wedge a scarce concurrency permit or OOM the process. + +- **Status:** design — awaiting review, then implementation plan. +- **Scope:** one issue, one PR, one ADR amendment (**ADR-0034 Amendment #13**). +- **Companion audits:** [Fable defect audit](2026-07-05-net-http-audit-findings.md) (N1); + [deep review](2026-07-06-net-http-deep-review.md) (§8 "no stall timeout on streaming + permit / buffered body"). Both were **anticipated by ADR-0034** — see Context. + +## 1. Context & motivation + +Two confirmed findings, on orthogonal axes, both already foreseen in ADR-0034: + +- **Stall (time).** In the default `BufferMode::Stream`, the leaf's `call` returns at + response **headers**; the body then drains *outside* every resilience layer — only + `Guarded` rides it ([body.rs:140](../../../crates/adapter/net/http/api/src/body.rs#L140)). + A body that stalls mid-transfer therefore holds one of the venue's concurrency permits + (IBKR `/history` has only 5) **indefinitely** — nothing bounds it. ADR-0034 **Amendment + #6** named the fix — a `TimeoutBody` bounding "a *streaming* transfer's mid-stream stall" + — and **deferred** it ("lands additively when a streaming venue first needs it"). This + un-defers it. + +- **Memory (N1).** `BufferMode::Buffer` collects the whole body in the leaf + ([leaf.rs:178-182](../../../crates/adapter/net/http/hyper/src/leaf.rs#L178-L182)) with + **no size cap** — a misbehaving venue can allocate unbounded memory. ADR-0034 **§2** + built the `ResponseBody`/`Guarded` `size_hint` **transparency** specifically so "any + `size_hint().upper()` max-size guard" would not "fail open" — i.e. the plumbing for this + guard was laid deliberately. This wires it. + +The two are genuinely complementary: `Buffer` mode is already **time**-bounded (its +`collect()` runs *inside* `call`, so the existing send `Timeout` covers it) but not +**memory**-bounded; `Stream` mode is **memory**-safe (the caller drains frame-by-frame, so +nothing accrues in our code) but not **time**-bounded once `call` returns at headers. So +the stall guard targets **Stream** mode and the size cap targets **Buffer** mode. + +## 2. Goals / non-goals + +**Goals** + +1. A streaming response body that goes idle for longer than a configured duration fails + with `HttpError::Timeout` and **releases its concurrency permit**. +2. A buffered response body larger than a configured cap fails with a new, non-retryable + `HttpError::BodyTooLarge` **before** unbounded allocation. +3. Both remain **runtime-neutral** (driven by the `Timer` seam, deterministically + `MockTimer`-testable) and keep the "every `Body::Error` is `HttpError`" invariant. + +**Non-goals (YAGNI / deferred)** + +- **Per-request overrides** (`RequestStallTimeout`, per-request byte cap) — additive later + if a venue needs per-endpoint tuning; v1 is a single config value each. +- **A total (wall-clock) body deadline** — we implement per-frame *inactivity* (stall) + semantics only, matching `tower_http::TimeoutBody` (its `DeadlineBody` sibling is the + total-deadline variant; not needed here). +- **Buffer-mode *time* bounding** — stays with the existing send `Timeout` layer. +- **Tier-3 buffering-as-a-layer re-cut** (deep review §5.4) — untouched; this design is + forward-compatible with it (the `TimeoutBody` is already a layer; the cap moves with the + buffering when that lands). + +## 3. Design overview + +| Guardrail | Axis | Mode it protects | Where | Mechanism | Error | +|---|---|---|---|---|---| +| Stall timeout | time | Stream | `net-http-api` (new layer) | per-frame inactivity deadline via `Timer` | `HttpError::Timeout` | +| Buffered size cap | memory | Buffer | `net-http-api` `LimitedBody` + `net-http-hyper` leaf | `size_hint().upper()` fast-fail + byte-count while collecting (typed `HttpError`) | `HttpError::BodyTooLarge` | + +## 4. Component A — `Timer` gains an associated `Sleep` type (`net-api`) + +The stall body must **store** its sleep future across many `poll_frame` calls (unlike the +`Timeout` layer, which stack-pins one sleep inside a single `async` block). `Timer::sleep` +currently returns an **unnameable** `impl Future`, which cannot be a struct field without +boxing. Adding an associated type makes the concrete future nameable, so the field is +stored **inline** — zero allocation, no `dyn` (mirrors `tower_http::TimeoutBody`'s inline +`Option`). + +```rust +// crates/adapter/net/api/src/timer.rs +pub trait Timer: Clone + Send + Sync { + type Sleep: Future + Send; // NEW + fn sleep(&self, dur: Duration) -> Self::Sleep; // was `-> impl Future + Send` + fn now(&self) -> Instant; +} +``` + +All **three** existing impls already return named, concrete futures — the change is one +line each, and no caller changes (`self.timer.sleep(dur).await` is identical): + +| impl | `type Sleep = …` | +|---|---| +| `TokioTimer` ([hyper/src/timer.rs](../../../crates/adapter/net/http/hyper/src/timer.rs)) | `tokio::time::Sleep` | +| `MockTimer` ([mock/src/timer.rs:70](../../../crates/adapter/net/mock/src/timer.rs#L70)) | its existing `pub struct Sleep` | +| `FixedTimer` (net-api test double) | `std::future::Ready<()>` | + +Safe because `Timer` is **only ever a generic bound** — there is no `dyn Timer` anywhere in +the workspace (verified), so object-safety is irrelevant. An impl that genuinely could not +name its future may still fall back to `type Sleep = Pin>` locally; the +trait *enables* zero-alloc without forcing it. This also equips the forthcoming WS +heartbeat/stall body (ADR-0033) with the same inline-timer storage. + +## 5. Component B — `TimeoutBody` + `StallTimeoutLayer` (`net-http-api`) + +A new module (e.g. `stall.rs`) mirroring `tower_http::timeout::TimeoutBody`, but generic +over our `Timer` and typed to `HttpError`. + +```rust +pin_project_lite::pin_project! { + pub struct TimeoutBody { + #[pin] inner: B, + #[pin] sleep: Option, // inline; None until first armed / after each reset + timeout: Option, // None => fully inert pass-through + timer: T, + } +} +``` + +`poll_frame` (the load-bearing order — poll the timer **before** the body so its waker +stays registered while the body is `Pending`): + +1. If `timeout` is `None` → forward straight to `inner.poll_frame` (fully inert). +2. Arm `sleep` if unset: `sleep.set(Some(self.timer.sleep(dur)))`. +3. Poll the timer; if `Ready(())` → `Poll::Ready(Some(Err(HttpError::Timeout)))`. +4. `let frame = ready!(inner.poll_frame(cx));` +5. On any frame (data / trailer / terminal `None` / error), `sleep.set(None)` — **lazy + per-frame reset** (inactivity semantics; next poll re-arms). Return the frame. + +**Transparency (ADR-0034 §2):** forward `is_end_stream` and `size_hint` to `inner` — +required so `Guarded`'s already-ended check and any collector stay correct. + +**`StallTimeoutLayer`** mirrors `TimeoutLayer`: its `Service::call` awaits +`inner.call(req)`, then re-wraps the response body: +`Response::from_parts(parts, TimeoutBody::new(body, self.timeout, self.timer.clone()))`. +Requires `B: Body`; `TimeoutBody` is `Send` when `B`/`T` +are (preserving the M5 `Body: Send` return bound). + +**Inert on Buffer mode:** a buffered body is one ready frame, so the timer arms and +immediately resets — it never fires (matches Am#6's "inert on buffered responses"). + +## 6. Component C — buffered size cap via a typed `LimitedBody` (`net-http-api`) + +A small **typed** body wrapper beside `ResponseBody`/`Guarded` in `body.rs`, kept in +`HttpError` end-to-end. `http_body_util::Limited` exists, but its +`Error = Box` forces a runtime downcast and injects a +`Box`-typed body into the pipeline — breaking the crate's defining "one +concrete `HttpError` for service *and* body" invariant (error.rs). A ~30-line typed +wrapper avoids that and is independently unit-testable: + +```rust +pin_project_lite::pin_project! { + pub struct LimitedBody { + #[pin] inner: B, + remaining: u64, + } +} +// poll_frame: count DATA-frame bytes; on overflow emit HttpError::BodyTooLarge; pass +// trailers / terminal None / inner Err through. is_end_stream forwarded; size_hint +// clamped to `remaining` (mirrors Limited's lower>=n / upper.min(n) logic, no lower>upper). +``` + +Leaf `Buffer` path — the collect error is **already** `HttpError`, so there is no shim: + +```rust +BufferMode::Buffer => { + let bytes = match self.max_response_bytes { + Some(cap) => { + let cap = cap as u64; + // Fast-fail on an honest oversized Content-Length, before reading a byte. + if incoming.size_hint().upper().is_some_and(|u| u > cap) { + return Err(HttpError::BodyTooLarge); + } + LimitedBody::new(incoming.map_err(map_hyper_err), cap) + .collect().await?.to_bytes() // BodyTooLarge or a mapped inner error + } + None => incoming.collect().await.map_err(map_hyper_err)?.to_bytes(), + }; + ResponseBody::buffered(bytes) +} +``` + +Peak memory is bounded to ≈ `cap` (the frame that would exceed it is rejected; +`collect()` does not pre-reserve from `size_hint`, so there is no size-hint allocation DoS +— the hyper `to_bytes` #3111 vector). `Stream` mode is untouched (the caller drains). + +**Why the stall body is *not* likewise a wrapper we could have reused:** `http_body_util` +has **no** timeout combinator (a pure body-util crate has no clock), and the only +ready-made one, `tower_http::timeout::TimeoutBody`, is welded to `tokio::time::Sleep` +(defeats the `Timer` seam / smol-async goal / `MockTimer` determinism) and drags in the +`tower`/`tower-http` trees this runtime-neutral crate reimplemented `compose.rs` to avoid +(ADR-0029). So both new body wrappers are hand-rolled and typed — consistent with the +existing `ResponseBody`/`Guarded`. + +## 7. Error surface — `BodyTooLarge` + +- **`net-api`** ([error_kind.rs:14](../../../crates/adapter/net/api/src/error_kind.rs#L14)): + add `ErrorKind::BodyTooLarge` (non-exhaustive enum, additive). + - **Non-retryable for free:** `retry.rs::is_transient` matches only `{Timeout, Connection}` + ([retry.rs:230-231](../../../crates/adapter/net/http/api/src/retry.rs#L230-L231)), so a new + kind is never retried — no `retry.rs` change. (Correct: retrying an oversized body just + re-overflows.) +- **`net-http-api`** ([error.rs:18](../../../crates/adapter/net/http/api/src/error.rs#L18)): + add unit variant `HttpError::BodyTooLarge` (message: "response body exceeded the + configured maximum") + its `HasErrorKind` arm → `ErrorKind::BodyTooLarge`. +- **`trace.rs`** ([trace.rs:30-40](../../../crates/adapter/net/http/api/src/trace.rs#L30-L40)): + add `ErrorKind::BodyTooLarge => "body_too_large"`. **Required** — the `_ => "unknown"` + fallback would otherwise mislabel it (the exact M2-class bug we're avoiding by choosing a + dedicated variant). + +The stall error reuses `HttpError::Timeout` (a body that didn't complete in time *is* a +timeout); it surfaces to the caller during draining (outside the Retry boundary, so Retry +never sees it) and triggers `Guarded`'s permit release. + +## 8. Config surface + +Both are **new required fields** (pre-release breaking additions — no `Default` on these +structs; every construction site sets them explicitly). + +- **`HttpConfig.body_stall_timeout: Option`** + ([stack.rs:30](../../../crates/adapter/net/http/api/src/stack.rs#L30)) — `None` disables + the stall guard (for a legitimately long-idle streaming endpoint); recommended value + `Some(Duration::from_secs(30))`. `validate_config` rejects `Some(Duration::ZERO)` + (a zero stall = insta-stall), symmetric with the existing `timeout` check; `None` is + allowed. +- **`ConnConfig.max_response_bytes: Option`** + ([leaf.rs:81](../../../crates/adapter/net/http/hyper/src/leaf.rs#L81)) — `None` = unbounded + (today's behaviour); recommended value `Some(16 * 1024 * 1024)`. Leaf-owned because the + cap is enforced where buffering happens. + +Construction sites to update (mechanical): `HttpConfig` in +[stack.rs](../../../crates/adapter/net/http/api/src/stack.rs) (doctest + `http_cfg` helper), +[build.rs](../../../crates/adapter/net/http/hyper/src/build.rs), +[examples/client_with_directives.rs](../../../crates/adapter/net/http/hyper/examples/client_with_directives.rs); +`ConnConfig` in [build.rs](../../../crates/adapter/net/http/hyper/src/build.rs), +[leaf.rs `test_conn`](../../../crates/adapter/net/http/hyper/src/leaf.rs), the example. + +## 9. Stack placement & permit-release correctness + +`StallTimeoutLayer` is added as the **innermost** `.layer()` in `stack()` +([stack.rs:174-180](../../../crates/adapter/net/http/api/src/stack.rs#L174-L180)): + +```rust +let svc = LayerBuilder::new() + .layer(TracingLayer::new(timer.clone())) // outermost + .layer(CircuitBreakerLayer::new(cfg.circuit_breaker, timer.clone())) + .layer(RetryLayer::new(cfg.retry, timer.clone())) + .layer(rate) // RateLimit → Guarded + .layer(TimeoutLayer::new(cfg.timeout, timer.clone())) + .layer(StallTimeoutLayer::new(cfg.body_stall_timeout, timer)) // NEW innermost + .wrap(inner); // SetHeaders(Auth(leaf)) +``` + +New order (outer→inner): +`Tracing(CB(Retry(RateLimit(Timeout(StallTimeout(SetHeaders(Auth(leaf))))))))`. +On the **response** path (inner→outer, `Auth`/`SetHeaders`/`Timeout` all body-transparent): + +``` +leaf → … → StallTimeout wraps in TimeoutBody → … → RateLimit wraps in Guarded +final body: Guarded< TimeoutBody< ResponseBody > > +``` + +`Guarded` is **outside** `TimeoutBody`, so a stall yields `Some(Err(Timeout))`, `Guarded` +observes a non-`Ok` frame, and its `poll_frame` releases the permit +([body.rs:189-203](../../../crates/adapter/net/http/api/src/body.rs#L189-L203)) — the wedged +`/history` slot is freed. This ordering is the crux of the design and gets a dedicated +full-stack test. + +## 10. ADR-0034 Amendment #13 + +Append-only amendment (highest current is #12): + +> **13. Response-body bounds — un-defers Am#6's `TimeoutBody`; wires §2's size guard.** +> The streaming mid-stream-stall `TimeoutBody` deferred in Amendment #6 lands as a +> `StallTimeoutLayer` (innermost, inside `RateLimit` so `Guarded` wraps it): a per-frame +> **inactivity** timeout via the `Timer` seam, `HttpError::Timeout` on stall, inert on +> buffered bodies. To store the sleep future inline (no `Box`/`dyn`), `Timer` gains +> `type Sleep: Future + Send`. Independently, `BufferMode::Buffer`'s collect is +> **capped** (`ConnConfig::max_response_bytes`) via a typed `LimitedBody` wrapper plus a +> `size_hint().upper()` fast-fail, completing the max-size guard §2's wrapper-transparency +> was built to support (N1); overflow is a new non-retryable `HttpError::BodyTooLarge` +> (`ErrorKind::BodyTooLarge`, `error_kind="body_too_large"`). Both config values are +> `Option` (disable-able); `HttpConfig.body_stall_timeout` is `validate_config`-checked +> non-zero when `Some`. Cross-refs: ADR-0031 §1 (`Timeout`), ADR-0030 §4 (buffering). + +## 11. Testing strategy (TDD, `MockTimer`-driven) + +Red-green per unit, mirroring the existing `Timeout`/`body.rs` test style (inline body +doubles, no `MockClient` dev-dep cycle). + +**`TimeoutBody`** (unit, `net-http-api`): +- Stall fires: a body that never yields a frame → advancing past the timeout → + `Some(Err(HttpError::Timeout))`. +- Steady body does **not** trip: a frame arrives each interval (reset works) → completes OK. +- `timeout = None` → fully inert (never arms; forwards frames verbatim). +- Buffered/already-ended body → inert (arms then immediately resets; never fires). +- `is_end_stream`/`size_hint` forwarded (parity assertion, as in `body.rs`). + +**Permit release** (unit + full-stack): a `Guarded>` whose inner stalls → +timer advance → permit released while the body is still held (not drop-driven), proving the +ordering; and a full `stack()` test that a wedged streaming `/history` body frees the +concurrency slot. + +**Size cap** (leaf, `net-http-hyper`, real hyper server doubles like the existing +truncating server): +- Honest oversized `Content-Length` → upfront `BodyTooLarge`, no body read. +- Lying/absent length streaming past `cap` → `BodyTooLarge` from `collect`. +- Under-cap body → collects normally. +- `None` cap → unbounded (today's behaviour) still works. + +**Classification** (`stack()`): `BodyTooLarge` returned from the leaf in `Buffer` mode is +**not retried** (leaf hit once); `error_kind` label is `"body_too_large"`. + +**`Timer::Sleep`**: existing `Timeout`/`Retry`/`RateLimit`/`CircuitBreaker` suites re-pass +unchanged (proves the associated-type refactor is behaviour-preserving). + +## 12. Housekeeping / definition of done + +- **CHANGELOG.md `[Unreleased]`** — a "Breaking (pre-release) — net response-body bounds" + entry (new `HttpConfig`/`ConnConfig` fields, new `HttpError`/`ErrorKind` variant, `Timer` + gains `type Sleep`) plus a feature line. +- **Rustdoc + doctests** on every new public item (`TimeoutBody`, `StallTimeoutLayer`, + `LimitedBody`, `HttpError::BodyTooLarge`, `ErrorKind::BodyTooLarge`, the two config fields, + `Timer::Sleep`). +- **`just ci` green**, including **`just doc`** (per prior net-http layer PRs — check/lint/test + miss broken intra-doc links). +- GitHub: open the Tier-2 issue, branch in a worktree under `.claude/worktrees/`, + `Closes #N`, squash-merge; tick the "Streaming stall TimeoutBody" box in #102. + +## 13. Task breakdown (for the implementation plan) + +1. `Timer::Sleep` associated type + 3 impls (+ re-pass existing timing suites). +2. `ErrorKind::BodyTooLarge` + `HttpError::BodyTooLarge` + `HasErrorKind` + `kind_label`. +3. `TimeoutBody` + `StallTimeoutLayer` module (TDD) — the stall unit tests. +4. Wire `StallTimeoutLayer` into `stack()`; `HttpConfig.body_stall_timeout` + + `validate_config`; full-stack permit-release test. +5. Typed `LimitedBody` wrapper (TDD, `net-http-api`) — its unit tests. +6. `ConnConfig.max_response_bytes` + leaf `Buffer`-arm cap using `LimitedBody` — the cap + leaf tests. +7. ADR-0034 Amendment #13; CHANGELOG; update construction sites, doctests, the example. + +## 14. Risks / open questions + +- **Default values** (`30s` stall, `16 MiB` cap) are proposals — confirm against IBKR's + actual payload sizes / idle behaviour. +- **`LimitedBody` `size_hint` clamp** must not report `lower > upper` — mirror + `http_body_util::Limited`'s `lower>=n → set_exact` / else `upper.min(n)` logic; a unit test + on a body whose `size_hint` lower bound exceeds the cap pins it. +- **Lazy reset skew:** re-arming on the next poll (not at frame arrival) measures the fresh + deadline from that poll — the same minor skew `tower_http::TimeoutBody` accepts; immaterial + for a stall (inactivity) bound. From 63498d85ff92d1c5efa2c1e001dbbbbb5db339dc Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 08:15:13 +0000 Subject: [PATCH 02/10] refactor(net)!: add Timer::Sleep type for inline sleep storage --- crates/adapter/net/api/src/timer.rs | 11 ++++++++--- crates/adapter/net/http/hyper/src/timer.rs | 4 ++-- crates/adapter/net/mock/src/timer.rs | 3 ++- 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/crates/adapter/net/api/src/timer.rs b/crates/adapter/net/api/src/timer.rs index 2c1f1e7..dbaf2d8 100644 --- a/crates/adapter/net/api/src/timer.rs +++ b/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 + Send; + /// Complete after `dur` has elapsed. - fn sleep(&self, dur: Duration) -> impl Future + 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 + Send { + type Sleep = std::future::Ready<()>; + fn sleep(&self, _dur: Duration) -> std::future::Ready<()> { std::future::ready(()) } fn now(&self) -> Instant { diff --git a/crates/adapter/net/http/hyper/src/timer.rs b/crates/adapter/net/http/hyper/src/timer.rs index 8219204..3c25469 100644 --- a/crates/adapter/net/http/hyper/src/timer.rs +++ b/crates/adapter/net/http/hyper/src/timer.rs @@ -1,7 +1,6 @@ //! [`TokioTimer`] — the tokio-backed [`Timer`] the resilience stack sleeps on. use oath_adapter_net_api::Timer; -use std::future::Future; use std::time::{Duration, Instant}; /// The tokio-backed [`Timer`]: [`sleep`](Timer::sleep) and [`now`](Timer::now) @@ -13,7 +12,8 @@ use std::time::{Duration, Instant}; pub struct TokioTimer; impl Timer for TokioTimer { - fn sleep(&self, dur: Duration) -> impl Future + Send { + type Sleep = tokio::time::Sleep; + fn sleep(&self, dur: Duration) -> tokio::time::Sleep { tokio::time::sleep(dur) } diff --git a/crates/adapter/net/mock/src/timer.rs b/crates/adapter/net/mock/src/timer.rs index f6eb486..846db22 100644 --- a/crates/adapter/net/mock/src/timer.rs +++ b/crates/adapter/net/mock/src/timer.rs @@ -97,7 +97,8 @@ impl Future for Sleep { } impl Timer for MockTimer { - fn sleep(&self, dur: Duration) -> impl Future + Send { + type Sleep = Sleep; + fn sleep(&self, dur: Duration) -> Sleep { let deadline = { let state = lock(&self.state); state.now + dur From c89e888774a98b355dd2221386a3d32e1d43bab2 Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 08:22:50 +0000 Subject: [PATCH 03/10] feat(net): add BodyTooLarge error kind, variant, and label --- crates/adapter/net/api/src/error_kind.rs | 5 +++++ crates/adapter/net/http/api/src/error.rs | 5 +++++ crates/adapter/net/http/api/src/retry.rs | 19 +++++++++++++++++++ crates/adapter/net/http/api/src/trace.rs | 9 +++++++++ 4 files changed, 38 insertions(+) diff --git a/crates/adapter/net/api/src/error_kind.rs b/crates/adapter/net/api/src/error_kind.rs index 3663e78..adf4553 100644 --- a/crates/adapter/net/api/src/error_kind.rs +++ b/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`]. diff --git a/crates/adapter/net/http/api/src/error.rs b/crates/adapter/net/http/api/src/error.rs index a555bcf..1058bcd 100644 --- a/crates/adapter/net/http/api/src/error.rs +++ b/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] diff --git a/crates/adapter/net/http/api/src/retry.rs b/crates/adapter/net/http/api/src/retry.rs index e498001..b86a1d7 100644 --- a/crates/adapter/net/http/api/src/retry.rs +++ b/crates/adapter/net/http/api/src/retry.rs @@ -413,6 +413,7 @@ mod tests { ErrorKind::Connection => HttpError::connection("reset"), ErrorKind::Throttled => HttpError::Throttled, ErrorKind::Auth => HttpError::auth("expired"), + ErrorKind::BodyTooLarge => HttpError::BodyTooLarge, _ => HttpError::other("boom"), } } @@ -865,6 +866,24 @@ mod tests { Duration::from_millis(80) ); } + + #[tokio::test] + async fn body_too_large_is_not_retried() { + // BodyTooLarge is not a transient kind, so even an eligible request sends once. + let leaf = ScriptLeaf::new(vec![Step::Err(ErrorKind::BodyTooLarge)]); + let svc = RetryLayer::new( + cfg(3, Duration::from_millis(1), Duration::from_millis(1)), + MockTimer::new(), + ) + .layer(leaf.clone()); + let err = svc.call(req(true)).await.unwrap_err(); + assert!(matches!(err, HttpError::BodyTooLarge)); + assert_eq!( + leaf.calls(), + 1, + "BodyTooLarge is non-transient → never retried" + ); + } } #[cfg(test)] diff --git a/crates/adapter/net/http/api/src/trace.rs b/crates/adapter/net/http/api/src/trace.rs index 2859669..ddf25b6 100644 --- a/crates/adapter/net/http/api/src/trace.rs +++ b/crates/adapter/net/http/api/src/trace.rs @@ -37,6 +37,7 @@ pub(crate) const fn kind_label(kind: ErrorKind) -> &'static str { ErrorKind::Client => "client", ErrorKind::Server => "server", ErrorKind::CircuitOpen => "circuit_open", + ErrorKind::BodyTooLarge => "body_too_large", _ => "unknown", // ErrorKind::Unknown and any future non_exhaustive variant } } @@ -615,4 +616,12 @@ mod tests { "circuit_open" ); } + + #[test] + fn body_too_large_has_a_stable_label() { + assert_eq!( + super::kind_label(oath_adapter_net_api::ErrorKind::BodyTooLarge), + "body_too_large" + ); + } } From 3f57ecd9eacb2eae9cebb1275181591cd5ff5379 Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 08:37:49 +0000 Subject: [PATCH 04/10] feat(net): TimeoutBody + StallTimeoutLayer for body stalls --- crates/adapter/net/http/api/src/lib.rs | 2 + crates/adapter/net/http/api/src/stall.rs | 452 +++++++++++++++++++++++ 2 files changed, 454 insertions(+) create mode 100644 crates/adapter/net/http/api/src/stall.rs diff --git a/crates/adapter/net/http/api/src/lib.rs b/crates/adapter/net/http/api/src/lib.rs index 1d09f78..48769e1 100644 --- a/crates/adapter/net/http/api/src/lib.rs +++ b/crates/adapter/net/http/api/src/lib.rs @@ -45,6 +45,7 @@ pub mod retry; mod retry_after; pub mod service; pub mod stack; +pub mod stall; pub mod timeout; pub mod trace; @@ -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}; diff --git a/crates/adapter/net/http/api/src/stall.rs b/crates/adapter/net/http/api/src/stall.rs new file mode 100644 index 0000000..4845178 --- /dev/null +++ b/crates/adapter/net/http/api/src/stall.rs @@ -0,0 +1,452 @@ +//! The `StallTimeout` body-inactivity layer (ADR-0034 Amendment #13). +//! +//! Wraps a *streaming* response body with a per-frame **inactivity** deadline via +//! the runtime-neutral [`Timer`] seam: if no frame arrives within the configured +//! duration the body yields [`HttpError::Timeout`], so a stalled transfer can no +//! longer wedge a `Guarded` concurrency permit. Placed innermost in `stack()` so +//! `Guarded` wraps [`TimeoutBody`] and the stall error releases the permit. +//! Inert on buffered bodies (one ready frame) and when the deadline is `None`. + +use crate::{HttpError, Service}; +use bytes::Bytes; +use http_body::{Body, Frame, SizeHint}; +use oath_adapter_net_api::{Layer, Timer}; +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll, ready}; +use std::time::Duration; + +pin_project_lite::pin_project! { + /// A response body wrapper enforcing a per-frame **inactivity** timeout. + /// + /// On each poll it (re-)arms a [`Timer::sleep`] deadline; if the deadline + /// fires before the inner body produces a frame, it yields + /// [`HttpError::Timeout`]. Each frame resets the deadline (lazily: `sleep` is + /// cleared and re-armed on the next poll). A `None` timeout is fully inert — + /// the wrapper forwards to `inner` unchanged. Forwards `is_end_stream` and + /// `size_hint` (ADR-0034 §2 transparency). + pub struct TimeoutBody { + #[pin] + inner: B, + #[pin] + sleep: Option, + timeout: Option, + timer: T, + } +} + +impl TimeoutBody { + /// Wrap `inner` with a stall timeout. `None` disables it (pass-through). + /// + /// # Example + /// ``` + /// use oath_adapter_net_http_api::TimeoutBody; + /// use oath_adapter_net_mock::MockTimer; + /// use http_body_util::Empty; + /// use bytes::Bytes; + /// use std::time::Duration; + /// + /// let _body = TimeoutBody::new( + /// Empty::::new(), + /// Some(Duration::from_secs(30)), + /// MockTimer::new(), + /// ); + /// ``` + #[must_use] + pub const fn new(inner: B, timeout: Option, timer: T) -> Self { + Self { + inner, + sleep: None, + timeout, + timer, + } + } +} + +impl fmt::Debug for TimeoutBody { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TimeoutBody") + .field("inner", &self.inner) + .field("timeout", &self.timeout) + .finish_non_exhaustive() + } +} + +impl Body for TimeoutBody +where + B: Body, + T: Timer, +{ + type Data = Bytes; + type Error = HttpError; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, HttpError>>> { + let mut this = self.project(); + let Some(dur) = *this.timeout else { + // Disabled: fully transparent. + return this.inner.poll_frame(cx); + }; + // Arm the deadline if unset. + if this.sleep.is_none() { + let nap = this.timer.sleep(dur); + this.sleep.set(Some(nap)); + } + // Poll the timer FIRST so its waker stays registered while the body pends. + if let Some(sleep) = this.sleep.as_mut().as_pin_mut() + && sleep.poll(cx).is_ready() + { + return Poll::Ready(Some(Err(HttpError::Timeout))); + } + let frame = ready!(this.inner.poll_frame(cx)); + this.sleep.set(None); // lazy per-frame reset (inactivity semantics) + Poll::Ready(frame) + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> SizeHint { + self.inner.size_hint() + } +} + +/// The [`StallTimeout`] [`Layer`] factory: holds the (optional) inactivity +/// deadline + clock and wraps any inner service's response body in +/// [`TimeoutBody`]. +pub struct StallTimeoutLayer { + timeout: Option, + timer: T, +} + +impl StallTimeoutLayer { + /// Build the layer. `None` disables the stall timeout (pass-through body). + /// + /// # Example + /// ``` + /// use oath_adapter_net_http_api::StallTimeoutLayer; + /// use oath_adapter_net_mock::MockTimer; + /// use std::time::Duration; + /// + /// let _layer = StallTimeoutLayer::new(Some(Duration::from_secs(30)), MockTimer::new()); + /// ``` + #[must_use] + pub const fn new(timeout: Option, timer: T) -> Self { + Self { timeout, timer } + } +} + +impl Clone for StallTimeoutLayer { + fn clone(&self) -> Self { + Self { + timeout: self.timeout, + timer: self.timer.clone(), + } + } +} + +impl fmt::Debug for StallTimeoutLayer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("StallTimeoutLayer") + .field("timeout", &self.timeout) + .finish_non_exhaustive() + } +} + +impl Layer for StallTimeoutLayer { + type Wrapped = StallTimeout; + fn layer(&self, inner: S) -> StallTimeout { + StallTimeout { + inner, + timeout: self.timeout, + timer: self.timer.clone(), + } + } +} + +/// The `StallTimeout` middleware: wraps the inner service's response body. +/// +/// Wraps the response body in a [`TimeoutBody`]. The send itself is untouched +/// (that is the `Timeout` layer's job); only the streamed body gains the +/// inactivity deadline. +pub struct StallTimeout { + inner: S, + timeout: Option, + timer: T, +} + +impl Clone for StallTimeout { + fn clone(&self) -> Self { + Self { + inner: self.inner.clone(), + timeout: self.timeout, + timer: self.timer.clone(), + } + } +} + +impl fmt::Debug for StallTimeout { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("StallTimeout") + .field("timeout", &self.timeout) + .finish_non_exhaustive() + } +} + +impl Service> for StallTimeout +where + S: Service, Response = http::Response, Error = HttpError> + Sync, + T: Timer, + B: Body, +{ + type Response = http::Response>; + type Error = HttpError; + + #[allow(clippy::manual_async_fn)] + fn call( + &self, + req: http::Request, + ) -> impl Future> + Send { + async move { + let resp = self.inner.call(req).await?; + let (parts, body) = resp.into_parts(); + let body = TimeoutBody::new(body, self.timeout, self.timer.clone()); + Ok(http::Response::from_parts(parts, body)) + } + } +} + +#[cfg(test)] +mod tests { + use super::TimeoutBody; + use crate::HttpError; + use bytes::Bytes; + use http_body::{Body, Frame, SizeHint}; + use oath_adapter_net_mock::MockTimer; + use std::collections::VecDeque; + use std::pin::{Pin, pin}; + use std::task::{Context, Poll, Waker}; + use std::time::Duration; + + use crate::Guarded; + use async_lock::Semaphore; + use std::sync::Arc; + + // A body that never yields a frame — models a wedged/stalled transfer. + struct PendingBody; + impl Body for PendingBody { + type Data = Bytes; + type Error = HttpError; + fn poll_frame( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll, HttpError>>> { + Poll::Pending + } + fn is_end_stream(&self) -> bool { + false + } + fn size_hint(&self) -> SizeHint { + SizeHint::default() + } + } + + #[tokio::test] + async fn stall_fires_when_no_frame_arrives() { + let timer = MockTimer::new(); + let body = TimeoutBody::new(PendingBody, Some(Duration::from_secs(1)), timer.clone()); + let waiter = tokio::spawn(async move { + let mut body = pin!(body); + std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await + }); + tokio::task::yield_now().await; // arms the deadline, inner pends + timer.advance(Duration::from_secs(1)); // fire the stall deadline + let frame = waiter.await.unwrap(); + assert!( + matches!(frame, Some(Err(HttpError::Timeout))), + "a stalled body must yield Timeout" + ); + } + + // A multi-frame body; every frame is immediately ready. + struct Frames { + frames: VecDeque, + } + impl Frames { + fn new(frames: [&'static [u8]; N]) -> Self { + Self { + frames: frames.iter().copied().map(Bytes::from_static).collect(), + } + } + } + impl Body for Frames { + type Data = Bytes; + type Error = HttpError; + fn poll_frame( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll, HttpError>>> { + let this = self.get_mut(); + Poll::Ready(this.frames.pop_front().map(|d| Ok(Frame::data(d)))) + } + fn is_end_stream(&self) -> bool { + self.frames.is_empty() + } + fn size_hint(&self) -> SizeHint { + SizeHint::with_exact(self.frames.iter().map(|f| f.len() as u64).sum()) + } + } + + #[test] + fn each_frame_resets_the_deadline() { + // timeout = 1s; two 500ms gaps (1s total) must NOT trip, because each + // per-frame gap is < 1s. A no-reset impl (deadline armed once) WOULD trip + // once cumulative time reaches 1s. + let timer = MockTimer::new(); + let body = TimeoutBody::new( + Frames::new([b"a", b"b"]), + Some(Duration::from_secs(1)), + timer.clone(), + ); + let mut body = pin!(body); + let mut cx = Context::from_waker(Waker::noop()); + assert!(matches!( + body.as_mut().poll_frame(&mut cx), + Poll::Ready(Some(Ok(_))) + )); // frame a @ t=0 + timer.advance(Duration::from_millis(500)); + assert!(matches!( + body.as_mut().poll_frame(&mut cx), + Poll::Ready(Some(Ok(_))) + )); // frame b @ t=0.5s + timer.advance(Duration::from_millis(500)); // t=1.0s total + assert!(matches!( + body.as_mut().poll_frame(&mut cx), + Poll::Ready(None) + )); // end, NOT a stall + } + + #[tokio::test] + async fn none_timeout_is_inert() { + // A None timeout never arms a deadline: a pending body just stays pending + // even after the clock advances arbitrarily. + let timer = MockTimer::new(); + let body = TimeoutBody::new(PendingBody, None, timer.clone()); + let mut body = pin!(body); + let mut cx = Context::from_waker(Waker::noop()); + assert!(body.as_mut().poll_frame(&mut cx).is_pending()); + timer.advance(Duration::from_secs(3600)); + assert!( + body.as_mut().poll_frame(&mut cx).is_pending(), + "None disables the stall timeout entirely" + ); + } + + #[test] + fn forwards_is_end_stream_and_size_hint() { + let inner = Frames::new([b"ab", b"cde"]); + let ref_hint = inner.size_hint().exact(); + let wrapped = TimeoutBody::new( + Frames::new([b"ab", b"cde"]), + Some(Duration::from_secs(1)), + MockTimer::new(), + ); + assert_eq!(wrapped.size_hint().exact(), ref_hint); // NOT silently unbounded + assert!(!wrapped.is_end_stream()); + let ended = TimeoutBody::new( + Frames::new([]), + Some(Duration::from_secs(1)), + MockTimer::new(), + ); + assert!(ended.is_end_stream()); // forwarded, not the `false` default + } + + // A stall must release the concurrency permit: with Guarded OUTSIDE TimeoutBody, + // the stall error frame is a non-Ok frame that Guarded observes and releases on. + #[expect( + clippy::significant_drop_tightening, + reason = "the test holds the guarded body across polls to prove release at the stall while still alive" + )] + #[test] + fn stall_releases_the_concurrency_permit() { + let sem = Arc::new(Semaphore::new(1)); + let permit = sem.try_acquire_arc().expect("permit free at start"); + let timer = MockTimer::new(); + let inner = TimeoutBody::new(PendingBody, Some(Duration::from_secs(1)), timer.clone()); + let body = Guarded::new(inner, Some(permit)); + let mut body = pin!(body); + let mut cx = Context::from_waker(Waker::noop()); + + assert!(body.as_mut().poll_frame(&mut cx).is_pending()); // arms deadline, permit held + assert!( + sem.try_acquire_arc().is_none(), + "permit held while streaming" + ); + + timer.advance(Duration::from_secs(1)); // fire the stall + assert!(matches!( + body.as_mut().poll_frame(&mut cx), + Poll::Ready(Some(Err(HttpError::Timeout))) + )); + assert!( + sem.try_acquire_arc().is_some(), + "the stall error must release the permit" + ); + } + + use crate::Service; + use http_body_util::BodyExt; + use oath_adapter_net_api::Layer; + + #[derive(Debug)] + struct StubBody { + data: Option, + } + impl Body for StubBody { + type Data = Bytes; + type Error = HttpError; + fn poll_frame( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll, HttpError>>> { + Poll::Ready(self.get_mut().data.take().map(|d| Ok(Frame::data(d)))) + } + fn is_end_stream(&self) -> bool { + self.data.is_none() + } + fn size_hint(&self) -> SizeHint { + self.data.as_ref().map_or_else( + || SizeHint::with_exact(0), + |d| SizeHint::with_exact(d.len() as u64), + ) + } + } + #[derive(Clone)] + struct FastLeaf; + impl Service> for FastLeaf { + type Response = http::Response; + type Error = HttpError; + async fn call(&self, _req: http::Request) -> Result { + Ok(http::Response::new(StubBody { + data: Some(Bytes::from_static(b"ok")), + })) + } + } + + #[tokio::test] + async fn layer_wraps_and_body_streams_through() { + use super::StallTimeoutLayer; + let svc = + StallTimeoutLayer::new(Some(Duration::from_secs(30)), MockTimer::new()).layer(FastLeaf); + let resp = svc + .call(http::Request::new(Bytes::new())) + .await + .expect("fast leaf"); + let body = resp.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(body, Bytes::from_static(b"ok")); // wrapped, streamed through untouched + } +} From aa04b89ed3ce6353271f1859ed2657c05bb060eb Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 09:02:23 +0000 Subject: [PATCH 05/10] feat(net)!: wire StallTimeoutLayer into stack() + body_stall_timeout Adds HttpConfig.body_stall_timeout (validated non-zero-when-Some via BuildError::ZeroDuration) and composes StallTimeoutLayer innermost in stack()'s LayerBuilder chain, just outside SetHeaders/Auth. This puts Guarded outside TimeoutBody so a mid-transfer body stall releases the concurrency permit instead of pinning it (ADR-0034 Amendment #13). Breaking: HttpConfig gains a new required field; updates the hyper build() doctest/test helper and the client_with_directives example. --- crates/adapter/net/http/api/src/stack.rs | 100 ++++++++++++++++-- .../hyper/examples/client_with_directives.rs | 1 + crates/adapter/net/http/hyper/src/build.rs | 2 + 3 files changed, 96 insertions(+), 7 deletions(-) diff --git a/crates/adapter/net/http/api/src/stack.rs b/crates/adapter/net/http/api/src/stack.rs index bfc8aa2..08cc055 100644 --- a/crates/adapter/net/http/api/src/stack.rs +++ b/crates/adapter/net/http/api/src/stack.rs @@ -1,20 +1,23 @@ //! The `stack()` assembly (ADR-0031 §1) + the non-generic `HttpConfig`. //! //! [`stack`] composes the canonical resilience order over any leaf: -//! `Tracing( CircuitBreaker( Retry( RateLimit( Timeout( SetHeaders( Auth( leaf ) ) ) ) ) ) )`. +//! `Tracing( CircuitBreaker( Retry( RateLimit( Timeout( StallTimeout( SetHeaders( Auth( leaf ) ) ) ) ) ) ) )`. //! It builds the one fallible layer ([`RateLimitLayer`], //! which validates pacing coverage + the concurrency-singleton invariant) first, //! so a coverage/param error is a [`BuildError`] before the //! rest is assembled. `Auth`/`SetHeaders` are direct `Service` wrappers (no -//! `Layer` factory), so they pre-wrap the leaf; the five `Layer`-factory layers +//! `Layer` factory), so they pre-wrap the leaf; the six `Layer`-factory layers //! compose over that via [`LayerBuilder`] -//! (first `.layer()` = outermost). The composed value satisfies -//! [`HttpClient`] by blanket impl. +//! (first `.layer()` = outermost). [`StallTimeoutLayer`] sits innermost of the +//! `Layer`-factory chain — just outside `SetHeaders`/`Auth` — so `Guarded` (the +//! concurrency permit) ends up OUTSIDE the stall-bounded body and a stall +//! releases the permit instead of pinning it (ADR-0034 Amendment #13). The +//! composed value satisfies [`HttpClient`] by blanket impl. use crate::rate::{BuildError, RateKey, RateLimitConfig}; use crate::{ Auth, AuthSource, CircuitBreakerConfig, CircuitBreakerLayer, HttpClient, RateLimitLayer, - RetryConfig, RetryLayer, SetHeaders, TimeoutLayer, TracingLayer, + RetryConfig, RetryLayer, SetHeaders, StallTimeoutLayer, TimeoutLayer, TracingLayer, }; use oath_adapter_net_api::{LayerBuilder, Timer}; use std::fmt; @@ -41,6 +44,11 @@ pub struct HttpConfig { /// `timeout`: `RateLimit` sits **outside** `Timeout`, so the permit wait is /// bounded by this — at IBKR's 1/15-min buckets, minutes not seconds. pub rate_limit_max_wait: Duration, + /// Per-frame inactivity deadline for a **streaming** response body; `None` + /// disables it. Bounds a mid-transfer stall so a slow body cannot pin a + /// concurrency permit indefinitely (ADR-0034 Amendment #13). Inert on + /// buffered responses. + pub body_stall_timeout: Option, } impl fmt::Debug for HttpConfig { @@ -51,6 +59,7 @@ impl fmt::Debug for HttpConfig { .field("retry", &self.retry) .field("circuit_breaker", &self.circuit_breaker) .field("rate_limit_max_wait", &self.rate_limit_max_wait) + .field("body_stall_timeout", &self.body_stall_timeout) .finish_non_exhaustive() } } @@ -62,7 +71,7 @@ impl fmt::Debug for HttpConfig { /// total over `K::all()`, carries an out-of-range policy param, or breaches the /// ≤1-concurrency-permit invariant is a [`BuildError`] before the infallible layers /// are assembled. Then composes, outermost-first: -/// `Tracing( CircuitBreaker( Retry( RateLimit( Timeout( SetHeaders( Auth( leaf ) ) ) ) ) ) )`. +/// `Tracing( CircuitBreaker( Retry( RateLimit( Timeout( StallTimeout( SetHeaders( Auth( leaf ) ) ) ) ) ) ) )`. /// `Auth`/`SetHeaders` are direct `Service` wrappers (no `Layer` factory), so they /// pre-wrap the leaf; the composed value satisfies [`HttpClient`] by blanket impl. /// @@ -137,6 +146,7 @@ impl fmt::Debug for HttpConfig { /// }, /// headers: http::HeaderMap::new(), /// rate_limit_max_wait: Duration::from_secs(0), +/// body_stall_timeout: Some(Duration::from_secs(30)), /// }; /// let rates = RateLimitConfig { /// global: LimitPolicy::TokenBucket { rate: 1000, per: Duration::from_secs(1), burst: 1000 }, @@ -176,7 +186,8 @@ where .layer(CircuitBreakerLayer::new(cfg.circuit_breaker, timer.clone())) .layer(RetryLayer::new(cfg.retry, timer.clone())) .layer(rate) - .layer(TimeoutLayer::new(cfg.timeout, timer)) // innermost Layer-factory + .layer(TimeoutLayer::new(cfg.timeout, timer.clone())) + .layer(StallTimeoutLayer::new(cfg.body_stall_timeout, timer)) // innermost .wrap(inner); Ok(svc) } @@ -214,6 +225,11 @@ const fn validate_config(cfg: &HttpConfig) -> Result<(), BuildError> { if cfg.circuit_breaker.retry_after_cap.is_zero() { return Err(BuildError::ZeroDuration("circuit_breaker.retry_after_cap")); } + if let Some(d) = cfg.body_stall_timeout + && d.is_zero() + { + return Err(BuildError::ZeroDuration("body_stall_timeout")); + } if cfg.circuit_breaker.failure_rate_threshold == 0 || cfg.circuit_breaker.failure_rate_threshold > 100 { @@ -430,6 +446,7 @@ mod tests { }, headers: http::HeaderMap::new(), rate_limit_max_wait: max_wait, + body_stall_timeout: Some(Duration::from_secs(30)), } } // Global effectively unlimited; Snapshot 2/s; History concurrency 1. @@ -934,6 +951,75 @@ mod tests { ); } + // A streaming body that never yields a frame (models a mid-transfer stall). + #[derive(Debug)] + struct StallingBody; + impl Body for StallingBody { + type Data = Bytes; + type Error = HttpError; + fn poll_frame( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll, HttpError>>> { + Poll::Pending + } + fn is_end_stream(&self) -> bool { + false + } + fn size_hint(&self) -> SizeHint { + SizeHint::default() + } + } + + // A dedicated minimal leaf (not `ScriptLeaf`, whose `Response` is pinned to + // `http::Response`) that always returns a stalling streaming body. + #[derive(Clone)] + struct StallLeaf; + impl Service> for StallLeaf { + type Response = http::Response; + type Error = HttpError; + async fn call(&self, _req: http::Request) -> Result { + Ok(http::Response::new(StallingBody)) + } + } + + #[tokio::test] + async fn a_stalled_streaming_body_times_out_through_the_stack() { + let timer = MockTimer::new(); + let mut cfg = http_cfg(1, Duration::from_secs(3600), Duration::ZERO); + cfg.body_stall_timeout = Some(Duration::from_secs(1)); // short body-stall bound + let svc = stack(StallLeaf, cfg, timer.clone(), NoAuth, rate_cfg()).expect("total config"); + + // The send returns at headers immediately; the stall bites while draining. + let resp = svc + .call(req(RateScope::Global)) + .await + .expect("headers arrive"); + let waiter = tokio::spawn(async move { + let mut body = std::pin::pin!(resp.into_body()); + std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await + }); + tokio::task::yield_now().await; // body registers the stall deadline + timer.advance(Duration::from_secs(1)); // fire it + let frame = waiter.await.unwrap(); + assert!( + matches!(frame, Some(Err(HttpError::Timeout))), + "a stalled streaming body must surface Timeout when drained" + ); + } + + #[test] + fn zero_body_stall_timeout_is_rejected_at_build() { + let timer = MockTimer::new(); + let leaf = ScriptLeaf::new(timer.clone(), vec![Step::Status(200)]); + let mut cfg = http_cfg(1, Duration::from_secs(1), Duration::ZERO); + cfg.body_stall_timeout = Some(Duration::ZERO); + let Err(err) = stack(leaf, cfg, timer, NoAuth, rate_cfg()) else { + panic!("Some(ZERO) body_stall_timeout must be a BuildError"); + }; + assert_eq!(err, BuildError::ZeroDuration("body_stall_timeout")); + } + #[test] fn http_config_debug_omits_secret_headers() { // `HttpConfig.headers` may carry static API keys; its `Debug` must not diff --git a/crates/adapter/net/http/hyper/examples/client_with_directives.rs b/crates/adapter/net/http/hyper/examples/client_with_directives.rs index 68cb3a2..f82f1ac 100644 --- a/crates/adapter/net/http/hyper/examples/client_with_directives.rs +++ b/crates/adapter/net/http/hyper/examples/client_with_directives.rs @@ -73,6 +73,7 @@ async fn main() { }, headers: http::HeaderMap::new(), rate_limit_max_wait: Duration::from_secs(0), + body_stall_timeout: Some(Duration::from_secs(30)), }; let rates = RateLimitConfig { global: LimitPolicy::TokenBucket { diff --git a/crates/adapter/net/http/hyper/src/build.rs b/crates/adapter/net/http/hyper/src/build.rs index 33acf4c..7ebd536 100644 --- a/crates/adapter/net/http/hyper/src/build.rs +++ b/crates/adapter/net/http/hyper/src/build.rs @@ -41,6 +41,7 @@ use std::fmt; /// circuit_breaker: CircuitBreakerConfig { failure_rate_threshold: 50, window_size: NonZeroU32::new(50).unwrap(), minimum_calls: NonZeroU32::new(10).unwrap(), cooldown: Duration::from_secs(30), retry_after_fallback: Duration::from_secs(900), retry_after_cap: Duration::from_secs(1800), half_open_probes: NonZeroU32::new(1).unwrap() }, /// headers: http::HeaderMap::new(), /// rate_limit_max_wait: Duration::from_secs(0), +/// body_stall_timeout: Some(Duration::from_secs(30)), /// }; /// let rates = RateLimitConfig { /// global: LimitPolicy::TokenBucket { rate: 1000, per: Duration::from_secs(1), burst: 1000 }, @@ -135,6 +136,7 @@ mod tests { }, headers: http::HeaderMap::new(), rate_limit_max_wait: Duration::ZERO, + body_stall_timeout: Some(Duration::from_secs(30)), } } From 5fe34c157796c9351ea6fd0b47ea5725396fae93 Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 09:16:20 +0000 Subject: [PATCH 06/10] =?UTF-8?q?feat(net):=20LimitedBody=20=E2=80=94=20ty?= =?UTF-8?q?ped=20max-bytes=20response-body=20wrapper?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/adapter/net/http/api/src/body.rs | 127 +++++++++++++++++++++++- crates/adapter/net/http/api/src/lib.rs | 2 +- 2 files changed, 127 insertions(+), 2 deletions(-) diff --git a/crates/adapter/net/http/api/src/body.rs b/crates/adapter/net/http/api/src/body.rs index cb13ec6..1e80c56 100644 --- a/crates/adapter/net/http/api/src/body.rs +++ b/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 { + #[pin] + inner: B, + remaining: u64, + } +} + +impl LimitedBody { + /// 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::::new(), 16 * 1024 * 1024); + /// ``` + #[must_use] + pub const fn new(inner: B, max_bytes: u64) -> Self { + Self { + inner, + remaining: max_bytes, + } + } +} + +impl fmt::Debug for LimitedBody { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("LimitedBody") + .field("inner", &self.inner) + .field("remaining", &self.remaining) + .finish() + } +} + +impl Body for LimitedBody +where + B: Body, +{ + type Data = Bytes; + type Error = HttpError; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, 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,37 @@ 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 + } } diff --git a/crates/adapter/net/http/api/src/lib.rs b/crates/adapter/net/http/api/src/lib.rs index 48769e1..3559aef 100644 --- a/crates/adapter/net/http/api/src/lib.rs +++ b/crates/adapter/net/http/api/src/lib.rs @@ -50,7 +50,7 @@ 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}; From b87e74e545ea6d71c8fbd5df5aa27e41f4eda983 Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 09:23:20 +0000 Subject: [PATCH 07/10] test(net): cover LimitedBody size_hint clamp branches + trailers Adds RangeHint (bounded-above-cap and unbounded-upper size_hint branches) and TrailersOnly (non-DATA frame accounting) test doubles. --- crates/adapter/net/http/api/src/body.rs | 99 +++++++++++++++++++++++++ 1 file changed, 99 insertions(+) diff --git a/crates/adapter/net/http/api/src/body.rs b/crates/adapter/net/http/api/src/body.rs index 1e80c56..505b8e4 100644 --- a/crates/adapter/net/http/api/src/body.rs +++ b/crates/adapter/net/http/api/src/body.rs @@ -586,4 +586,103 @@ mod tests { 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, + } + + impl Body for RangeHint { + type Data = Bytes; + type Error = HttpError; + fn poll_frame( + self: Pin<&mut Self>, + _: &mut Context<'_>, + ) -> Poll, 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, 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()); + } } From 49b4dd1e7505e1b60087b711e7eb695d93a5f791 Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 09:35:48 +0000 Subject: [PATCH 08/10] feat(net)!: cap buffered response via ConnConfig::max_response_bytes Add ConnConfig.max_response_bytes: Option, threaded through HyperLeaf, and apply it only to the BufferMode::Buffer arm of HyperLeaf::call. Two-layer cap: an upfront fast-fail when the response's honest Content-Length (size_hint().upper()) already exceeds the cap, then wrap the incoming body in LimitedBody before collect() so an unsized/chunked body is caught mid-collect. None stays unbounded (current behavior); overflow surfaces as HttpError::BodyTooLarge. Streaming responses are untouched (the stall timeout already covers those). Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 1 + crates/adapter/net/http/hyper/Cargo.toml | 1 + .../hyper/examples/client_with_directives.rs | 1 + crates/adapter/net/http/hyper/src/build.rs | 2 + crates/adapter/net/http/hyper/src/leaf.rs | 171 +++++++++++++++++- 5 files changed, 174 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 56560aa..be1f438 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -579,6 +579,7 @@ version = "0.1.0" dependencies = [ "bytes", "http", + "http-body", "http-body-util", "hyper", "hyper-rustls", diff --git a/crates/adapter/net/http/hyper/Cargo.toml b/crates/adapter/net/http/hyper/Cargo.toml index 4a136d4..d084bdf 100644 --- a/crates/adapter/net/http/hyper/Cargo.toml +++ b/crates/adapter/net/http/hyper/Cargo.toml @@ -10,6 +10,7 @@ oath-adapter-net-api = { workspace = true } oath-adapter-net-http-api = { workspace = true } bytes = { workspace = true } http = { workspace = true } +http-body = { workspace = true } http-body-util = { workspace = true } hyper = { workspace = true } hyper-util = { workspace = true } diff --git a/crates/adapter/net/http/hyper/examples/client_with_directives.rs b/crates/adapter/net/http/hyper/examples/client_with_directives.rs index f82f1ac..45ee8be 100644 --- a/crates/adapter/net/http/hyper/examples/client_with_directives.rs +++ b/crates/adapter/net/http/hyper/examples/client_with_directives.rs @@ -92,6 +92,7 @@ async fn main() { http2_keep_alive_interval: None, http2_keep_alive_timeout: Duration::from_secs(10), http2_keep_alive_while_idle: false, + max_response_bytes: Some(16 * 1024 * 1024), }; let client = build(cfg, TokioTimer, NoAuth, rates, conn).expect("valid config"); diff --git a/crates/adapter/net/http/hyper/src/build.rs b/crates/adapter/net/http/hyper/src/build.rs index 7ebd536..a404cbb 100644 --- a/crates/adapter/net/http/hyper/src/build.rs +++ b/crates/adapter/net/http/hyper/src/build.rs @@ -56,6 +56,7 @@ use std::fmt; /// http2_keep_alive_interval: None, /// http2_keep_alive_timeout: Duration::from_secs(10), /// http2_keep_alive_while_idle: false, +/// max_response_bytes: Some(16 * 1024 * 1024), /// }; /// let _client = build(cfg, TokioTimer, NoAuth, rates, conn).expect("valid config"); /// ``` @@ -113,6 +114,7 @@ mod tests { http2_keep_alive_interval: None, http2_keep_alive_timeout: Duration::from_secs(10), http2_keep_alive_while_idle: false, + max_response_bytes: Some(16 * 1024 * 1024), } } diff --git a/crates/adapter/net/http/hyper/src/leaf.rs b/crates/adapter/net/http/hyper/src/leaf.rs index eb22d75..5aa732a 100644 --- a/crates/adapter/net/http/hyper/src/leaf.rs +++ b/crates/adapter/net/http/hyper/src/leaf.rs @@ -7,6 +7,7 @@ use crate::error::{map_hyper_err, map_legacy_err}; use bytes::Bytes; +use http_body::Body; use http_body_util::combinators::MapErr; use http_body_util::{BodyExt, Full}; use hyper::body::Incoming; @@ -14,7 +15,7 @@ use hyper_rustls::HttpsConnector; use hyper_util::client::legacy::Client; use hyper_util::client::legacy::connect::HttpConnector; use hyper_util::rt::{TokioExecutor, TokioTimer as HyperPoolTimer}; -use oath_adapter_net_http_api::{BufferMode, HttpError, ResponseBody, Service}; +use oath_adapter_net_http_api::{BufferMode, HttpError, LimitedBody, ResponseBody, Service}; use rustls::RootCertStore; use rustls::pki_types::CertificateDer; use std::future::Future; @@ -103,6 +104,10 @@ pub struct ConnConfig { /// Send keepalive PINGs even with no active requests. Ignored when /// `http2_keep_alive_interval` is `None`. pub http2_keep_alive_while_idle: bool, + /// Maximum bytes to buffer for a `BufferMode::Buffer` response; `None` = + /// unbounded. Rejects an oversized body with [`HttpError::BodyTooLarge`] + /// (ADR-0034 Amendment #13) — memory-safety for a misbehaving venue. + pub max_response_bytes: Option, } /// The hyper backend leaf. `Clone` (the pool **and** the in-flight tracker are @@ -111,6 +116,7 @@ pub struct ConnConfig { pub struct HyperLeaf { client: Client, Full>, inflight: Arc, + max_response_bytes: Option, } impl std::fmt::Debug for HyperLeaf { @@ -158,6 +164,7 @@ impl Service> for HyperLeaf { // in-flight from invocation until the future completes or is dropped, so // `shutdown` can drain it. let guard = InFlightGuard::enter(&self.inflight); + let max_response_bytes = self.max_response_bytes; async move { let _guard = guard; // ADR-0030 §4: absent extension ⇒ Stream. `BufferMode` is `Copy`. @@ -177,7 +184,23 @@ impl Service> for HyperLeaf { }, BufferMode::Buffer => { // Collect inside the retry boundary → full-body retry coverage. - let bytes = incoming.collect().await.map_err(map_hyper_err)?.to_bytes(); + let bytes = match max_response_bytes { + Some(cap) => { + let cap = cap as u64; + if incoming + .size_hint() + .upper() + .is_some_and(|upper| upper > cap) + { + return Err(HttpError::BodyTooLarge); + } + LimitedBody::new(incoming.map_err(map_hyper_err), cap) + .collect() + .await? + .to_bytes() + }, + None => incoming.collect().await.map_err(map_hyper_err)?.to_bytes(), + }; ResponseBody::buffered(bytes) }, }; @@ -194,6 +217,7 @@ impl Service> for HyperLeaf { /// HTTPS-only unless [`ConnConfig::allow_http`] is set; optional HTTP/2 keepalive. #[must_use] pub fn hyper_leaf(conn: ConnConfig) -> HyperLeaf { + let max_response_bytes = conn.max_response_bytes; let mut http = HttpConnector::new(); http.enforce_http(false); // let the HTTPS wrapper handle `https://` http.set_connect_timeout(Some(conn.connect_timeout)); @@ -234,6 +258,7 @@ pub fn hyper_leaf(conn: ConnConfig) -> HyperLeaf { HyperLeaf { client, inflight: Arc::new(InFlight::default()), + max_response_bytes, } } @@ -261,6 +286,7 @@ mod tests { http2_keep_alive_interval: None, http2_keep_alive_timeout: Duration::from_secs(10), http2_keep_alive_while_idle: false, + max_response_bytes: None, } } @@ -288,6 +314,147 @@ mod tests { format!("http://{addr}") } + // Serves a body of `n` bytes with an explicit Content-Length (via Full). + async fn spawn_body_server(n: usize) -> String { + let payload = Bytes::from(vec![b'x'; n]); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + loop { + let (stream, _) = listener.accept().await.unwrap(); + let payload = payload.clone(); + let io = hyper_util::rt::TokioIo::new(stream); + tokio::spawn(async move { + let svc = hyper::service::service_fn(move |_r| { + let payload = payload.clone(); + async move { + Ok::<_, Infallible>(hyper::Response::new(http_body_util::Full::new( + payload, + ))) + } + }); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(io, svc) + .await; + }); + } + }); + format!("http://{addr}") + } + + #[tokio::test] + async fn buffer_rejects_an_oversized_content_length_upfront() { + let base = spawn_body_server(64).await; + let conn = ConnConfig { + max_response_bytes: Some(16), + ..test_conn() + }; + let leaf = hyper_leaf(conn); + let mut req = http::Request::get(format!("{base}/big")) + .body(Bytes::new()) + .unwrap(); + req.extensions_mut().insert(BufferMode::Buffer); + let err = leaf + .call(req) + .await + .expect_err("64-byte body over a 16-byte cap"); + assert!( + matches!(err, oath_adapter_net_http_api::HttpError::BodyTooLarge), + "expected BodyTooLarge, got {err:?}" + ); + } + + // Sends a chunked (no Content-Length) body of two `chunk`-sized pieces. + async fn spawn_chunked_server(chunk: &'static [u8], count: usize) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = Vec::new(); + let mut tmp = [0_u8; 256]; + loop { + let n = stream.read(&mut tmp).await.unwrap(); + assert!(n > 0, "peer closed before a full request head"); + buf.extend_from_slice(&tmp[..n]); + if buf.windows(4).any(|w| w == b"\r\n\r\n") { + break; + } + } + stream + .write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n") + .await + .unwrap(); + for _ in 0..count { + stream + .write_all(format!("{:x}\r\n", chunk.len()).as_bytes()) + .await + .unwrap(); + stream.write_all(chunk).await.unwrap(); + stream.write_all(b"\r\n").await.unwrap(); + } + stream.write_all(b"0\r\n\r\n").await.unwrap(); + stream.flush().await.unwrap(); + }); + format!("http://{addr}") + } + + #[tokio::test] + async fn buffer_caps_an_unsized_streaming_body_while_collecting() { + // Two 8-byte chunks = 16 bytes, no Content-Length, over a 10-byte cap. + let base = spawn_chunked_server(b"12345678", 2).await; + let conn = ConnConfig { + max_response_bytes: Some(10), + ..test_conn() + }; + let leaf = hyper_leaf(conn); + let mut req = http::Request::get(format!("{base}/chunked")) + .body(Bytes::new()) + .unwrap(); + req.extensions_mut().insert(BufferMode::Buffer); + let err = leaf + .call(req) + .await + .expect_err("16 chunked bytes over a 10-byte cap"); + assert!( + matches!(err, oath_adapter_net_http_api::HttpError::BodyTooLarge), + "expected BodyTooLarge from the streaming cap, got {err:?}" + ); + } + + #[tokio::test] + async fn buffer_under_cap_collects_normally() { + let base = spawn_body_server(8).await; + let conn = ConnConfig { + max_response_bytes: Some(1024), + ..test_conn() + }; + let leaf = hyper_leaf(conn); + let mut req = http::Request::get(format!("{base}/small")) + .body(Bytes::new()) + .unwrap(); + req.extensions_mut().insert(BufferMode::Buffer); + let resp = leaf.call(req).await.expect("8-byte body under a 1KiB cap"); + let body = resp.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(body.len(), 8); + } + + #[tokio::test] + async fn buffer_with_no_cap_is_unbounded() { + let base = spawn_body_server(64).await; + let conn = ConnConfig { + max_response_bytes: None, + ..test_conn() + }; + let leaf = hyper_leaf(conn); + let mut req = http::Request::get(format!("{base}/big")) + .body(Bytes::new()) + .unwrap(); + req.extensions_mut().insert(BufferMode::Buffer); + let resp = leaf.call(req).await.expect("no cap → unbounded"); + let body = resp.into_body().collect().await.unwrap().to_bytes(); + assert_eq!(body.len(), 64); + } + #[tokio::test] async fn leaf_round_trips_a_plain_http_body() { let base = spawn_echo_server(b"pong").await; From 3b407897445626d2f1b1606b62e5f83d23e7a291 Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 09:47:57 +0000 Subject: [PATCH 09/10] docs(net): ADR-0034 Am#13 + CHANGELOG for response-body bounds Records the streaming stall-timeout (StallTimeoutLayer/TimeoutBody) and buffered size-cap (LimitedBody/ConnConfig::max_response_bytes) work (Tasks 1-6 on this branch) as ADR-0034 Amendment #13, un-deferring Amendment #6's TimeoutBody and wiring Section 2's size guard. Adds the matching Unreleased CHANGELOG entries (Added + a breaking-change note under Changed). --- CHANGELOG.md | 13 +++++++++++++ ...struction-surface-auth-guarded-boot-coverage.md | 14 ++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e98461..c7452f9 100644 --- a/CHANGELOG.md +++ b/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 + Send` (was `fn sleep(&self) -> impl Future`), so body + wrappers store the sleep future inline without boxing. `HttpConfig` gains + `body_stall_timeout: Option` and `ConnConfig` gains + `max_response_bytes: Option` (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 diff --git a/docs/adr/0034-http-construction-surface-auth-guarded-boot-coverage.md b/docs/adr/0034-http-construction-surface-auth-guarded-boot-coverage.md index c807d2d..ebc6f5e 100644 --- a/docs/adr/0034-http-construction-surface-auth-guarded-boot-coverage.md +++ b/docs/adr/0034-http-construction-surface-auth-guarded-boot-coverage.md @@ -296,3 +296,17 @@ carries the full reasoning. The breaker's `throttle_cooldown` field is renamed **`retry_after_fallback`**, and a new **`retry_after_cap`** bounds an honored value. The `HTTP-date` form and alternate/absolute headers stay deferred (they need a wall-clock `Timer` seam). +13. **Response-body bounds — un-defers Am#6's `TimeoutBody`; wires §2's size guard.** + The streaming mid-stream-stall `TimeoutBody` deferred in Amendment #6 lands as a + `StallTimeoutLayer` (innermost, inside `RateLimit` so `Guarded` wraps it): a + per-frame **inactivity** timeout via the `Timer` seam, `HttpError::Timeout` on + stall, inert on buffered bodies and when disabled. To store the sleep future + inline (no `Box`/`dyn`), `Timer` gains `type Sleep: Future + Send`. + Independently, `BufferMode::Buffer`'s collect is **capped** + (`ConnConfig::max_response_bytes`) via a typed `LimitedBody` wrapper plus a + `size_hint().upper()` fast-fail, completing the max-size guard §2's + wrapper-transparency was built to support (N1); overflow is a new non-retryable + `HttpError::BodyTooLarge` (`ErrorKind::BodyTooLarge`, `error_kind="body_too_large"`). + Both config values are `Option` (disable-able); `HttpConfig.body_stall_timeout` + is `validate_config`-checked non-zero when `Some`. Cross-refs: ADR-0031 §1 + (`Timeout`), ADR-0030 §4 (buffering). From a92e29465f3b72c7db46a3a2bb9974f417a04790 Mon Sep 17 00:00:00 2001 From: NotAProfDev <84450364+NotAProfDev@users.noreply.github.com> Date: Fri, 10 Jul 2026 11:31:59 +0000 Subject: [PATCH 10/10] test(net): pin BodyTooLarge breaker-neutral; doc zero-cap + stall gap --- crates/adapter/net/http/api/src/circuit_breaker.rs | 12 ++++++++++++ crates/adapter/net/http/api/src/stall.rs | 4 ++++ crates/adapter/net/http/hyper/src/leaf.rs | 3 +++ 3 files changed, 19 insertions(+) diff --git a/crates/adapter/net/http/api/src/circuit_breaker.rs b/crates/adapter/net/http/api/src/circuit_breaker.rs index dce2fe4..3c23752 100644 --- a/crates/adapter/net/http/api/src/circuit_breaker.rs +++ b/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); diff --git a/crates/adapter/net/http/api/src/stall.rs b/crates/adapter/net/http/api/src/stall.rs index 4845178..e774dc2 100644 --- a/crates/adapter/net/http/api/src/stall.rs +++ b/crates/adapter/net/http/api/src/stall.rs @@ -6,6 +6,10 @@ //! longer wedge a `Guarded` concurrency permit. Placed innermost in `stack()` so //! `Guarded` wraps [`TimeoutBody`] and the stall error releases the permit. //! Inert on buffered bodies (one ready frame) and when the deadline is `None`. +//! This guard bounds mid-transfer *inactivity*, not total transferred size: a +//! steady, never-idle `BufferMode::Stream` response is bounded by neither this +//! guard nor the buffered-size cap, by design — in Stream mode the caller owns +//! accumulation, and the cap applies only to `BufferMode::Buffer`. use crate::{HttpError, Service}; use bytes::Bytes; diff --git a/crates/adapter/net/http/hyper/src/leaf.rs b/crates/adapter/net/http/hyper/src/leaf.rs index 5aa732a..ec546bb 100644 --- a/crates/adapter/net/http/hyper/src/leaf.rs +++ b/crates/adapter/net/http/hyper/src/leaf.rs @@ -107,6 +107,9 @@ pub struct ConnConfig { /// Maximum bytes to buffer for a `BufferMode::Buffer` response; `None` = /// unbounded. Rejects an oversized body with [`HttpError::BodyTooLarge`] /// (ADR-0034 Amendment #13) — memory-safety for a misbehaving venue. + /// Unlike `body_stall_timeout`, `Some(0)` is not rejected at boot: it is a + /// valid (if extreme) policy meaning "reject every non-empty buffered body", + /// not "no limit" — don't misread it as unbounded. pub max_response_bytes: Option, }