feat(net)!: rolling-window (error-rate) circuit breaker (ADR-0031 Am#3)

CHANGELOG.md

@@ -77,6 +77,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   renamed `retry_after_fallback` (the `429` reopen wait when no usable `Retry-After` is
   present), and a new `retry_after_cap` bounds an honored `Retry-After` (both validated
   non-zero at `stack()`/`build()`).
+- **Breaking (pre-release) — net-http circuit-breaker trip policy.** The `CircuitBreaker`
+  now trips on a **rolling error-rate window** (last-N outcomes) rather than consecutive
+  failures, so a venue failing a sustained fraction of interleaved traffic is detected
+  (a 50 %-error host no longer resets an alarm on every success) — ADR-0031 Amendment #3.
+  `CircuitBreakerConfig` drops `failure_threshold` and gains `failure_rate_threshold`
+  (percent, `1..=100`), `window_size`, and `minimum_calls` (validated at boot). The
+  `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`).
 
 ### Added
 

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

@@ -40,6 +40,7 @@ pub mod error;
 pub mod meter;
 pub mod rate;
 pub mod rate_limit;
+mod rate_window;
 pub mod retry;
 mod retry_after;
 pub mod service;

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

@@ -57,9 +57,15 @@ pub(crate) fn route_label(req: &http::Request<Bytes>) -> Cow<'static, str> {
         .map_or(Cow::Borrowed("other"), |t| t.0.clone())
 }
 
-/// Record a circuit-breaker phase transition into `to`.
-pub(crate) fn breaker_transition(to: &'static str) {
-    metrics::counter!(CIRCUIT_TRANSITIONS, "to" => to).increment(1);
+/// Record a circuit-breaker phase transition into `to`, with an optional `reason`
+/// (present only for `to = "open"`: `"rate" | "throttle" | "probe_failed" | "abandoned"`).
+pub(crate) fn breaker_transition(to: &'static str, reason: Option<&'static str>) {
+    match reason {
+        Some(reason) => {
+            metrics::counter!(CIRCUIT_TRANSITIONS, "to" => to, "reason" => reason).increment(1);
+        },
+        None => metrics::counter!(CIRCUIT_TRANSITIONS, "to" => to).increment(1),
+    }
 }
 
 /// Count one local pacing rejection for `route`.
@@ -111,28 +117,25 @@ mod tests {
     }
 
     #[test]
-    fn breaker_transition_increments_a_phase_labelled_counter() {
+    fn breaker_transition_carries_to_and_optional_reason() {
         let recorder = DebuggingRecorder::new();
         let snap = recorder.snapshotter();
         metrics::with_local_recorder(&recorder, || {
-            breaker_transition("open");
-            breaker_transition("open");
+            breaker_transition("open", Some("rate"));
+            breaker_transition("half_open", None);
         });
-        let counter = snap
-            .snapshot()
-            .into_vec()
-            .into_iter()
-            .find(|(k, _, _, _)| k.key().name() == "http_circuit_breaker_transitions_total")
-            .expect("counter emitted");
-        assert!(
-            counter
-                .0
-                .key()
-                .labels()
-                .any(|l| l.key() == "to" && l.value() == "open"),
-            "labelled to=open"
-        );
-        assert_eq!(counter.3, DebugValue::Counter(2), "two transitions counted");
+        let snapshot = snap.snapshot().into_vec();
+        let opened = snapshot.iter().any(|(k, _, _, v)| {
+            k.key().name() == "http_circuit_breaker_transitions_total"
+                && k.key()
+                    .labels()
+                    .any(|l| l.key() == "to" && l.value() == "open")
+                && k.key()
+                    .labels()
+                    .any(|l| l.key() == "reason" && l.value() == "rate")
+                && matches!(v, DebugValue::Counter(n) if *n == 1)
+        });
+        assert!(opened, "to=open carries reason=rate");
     }
 
     #[test]

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

@@ -149,6 +149,21 @@ pub enum BuildError {
     /// state). Symmetric with the pacing-parameter validation (deep review §2A).
     #[error("config field `{0}` must be a positive Duration, but is zero")]
     ZeroDuration(&'static str),
+    /// `circuit_breaker.failure_rate_threshold` is outside `1..=100` — a `0` would trip
+    /// on the first sample and a value `> 100` could never trip.
+    #[error("config field `circuit_breaker.failure_rate_threshold` must be in 1..=100, but is {0}")]
+    RateThresholdRange(u8),
+    /// `circuit_breaker.minimum_calls` (`{0}`) exceeds `window_size` (`{1}`) — the window
+    /// can never hold enough samples to reach the floor, so the breaker could never trip
+    /// on rate.
+    #[error("config field `circuit_breaker.minimum_calls` ({0}) must be <= `window_size` ({1})")]
+    MinCallsExceedWindow(u32, u32),
+    /// `circuit_breaker.window_size` (`{0}`) exceeds the sanity ceiling (`{1}`) — a
+    /// window far larger than any sensible "recent health" span is almost always a
+    /// units/typo mistake, and `RateWindow::new` would eagerly allocate (and
+    /// re-allocate on every Half-Open→Closed recovery) a `VecDeque` of that size.
+    #[error("config field `circuit_breaker.window_size` ({0}) exceeds the sanity ceiling ({1})")]
+    WindowSizeTooLarge(u32, u32),
 }
 
 /// Validate that `cfg` is a **total**, param-sane pacing configuration.

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

@@ -0,0 +1,151 @@
+//! A fixed-capacity rolling window of recent breaker outcomes tracking the failure
+//! rate for the error-rate trip policy (ADR-0031 Amendment #3).
+//!
+//! Clock-free: only host-health outcomes enter — a transport failure / `5xx`
+//! ([`Outcome::Failure`]) or a reached-host `2xx`/`3xx` ([`Outcome::Success`]). A
+//! `4xx`/`Auth` (`Class::Ignored`) is never pushed, and a venue `429` trips the breaker
+//! immediately without a window sample. A recovered host earns a fresh window via
+//! [`RateWindow::new`].
+//!
+//! Wired into [`crate::circuit_breaker`]'s `Breaker::record` as the Closed-state trip
+//! policy.
+
+use std::collections::VecDeque;
+use std::num::NonZeroU32;
+
+/// One breaker-relevant, host-health-bearing outcome that enters the window.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+// `pub(crate)` in a private module: clippy's nursery `redundant_pub_crate` wants
+// `pub`, but that would trip the workspace `unreachable_pub` lint instead (a `pub`
+// item not reachable outside the crate). `pub(crate)` states the true visibility;
+// silence the losing alternative, matching `clock.rs`/`retry_after.rs`.
+#[allow(clippy::redundant_pub_crate)]
+pub(crate) enum Outcome {
+    /// A transport failure (`Connection`/`Timeout`) or a `5xx` response.
+    Failure,
+    /// A reached-host success (`2xx`/`3xx`).
+    Success,
+}
+
+/// The last-`N` outcomes as a ring, with a running failure count so the rate is O(1).
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[allow(clippy::redundant_pub_crate)] // see `Outcome` above
+pub(crate) struct RateWindow {
+    cap: usize,
+    samples: VecDeque<Outcome>,
+    failures: u32,
+}
+
+impl RateWindow {
+    /// An empty window of capacity `window_size`. The single backing allocation is
+    /// sized once here; `push` never reallocates (it evicts before exceeding `cap`).
+    pub(crate) fn new(window_size: NonZeroU32) -> Self {
+        let cap = window_size.get() as usize;
+        Self {
+            cap,
+            samples: VecDeque::with_capacity(cap),
+            failures: 0,
+        }
+    }
+
+    /// Record one outcome (O(1)); evict the oldest once full, keeping `failures` exact.
+    pub(crate) fn push(&mut self, o: Outcome) {
+        if self.samples.len() == self.cap {
+            // Window full: drop the oldest. It was counted, so if it was a Failure the
+            // running count is >= 1 here and the decrement cannot underflow.
+            if self.samples.pop_front() == Some(Outcome::Failure) {
+                self.failures -= 1;
+            }
+        }
+        if o == Outcome::Failure {
+            self.failures += 1;
+        }
+        self.samples.push_back(o);
+    }
+
+    /// The current live sample count.
+    // `cap` (and so `samples.len()`, which never exceeds it — `push` evicts before
+    // growing past `cap`) originates from `NonZeroU32::get()` in `new`, so the value
+    // always fits back in a `u32`; the cast cannot truncate.
+    #[allow(clippy::cast_possible_truncation)]
+    pub(crate) fn len(&self) -> u32 {
+        self.samples.len() as u32
+    }
+
+    /// Trip iff at least `min_calls` samples **and** failure rate >= `threshold_pct`.
+    /// Integer cross-multiply — no float in the resilience path; `>=` trips. Widen to
+    /// `u64` so `failures * 100` cannot overflow for a large `window_size`.
+    pub(crate) fn should_trip(&self, min_calls: u32, threshold_pct: u32) -> bool {
+        let len = self.len();
+        len >= min_calls
+            && u64::from(self.failures) * 100 >= u64::from(threshold_pct) * u64::from(len)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{Outcome, RateWindow};
+    use std::num::NonZeroU32;
+
+    fn win(cap: u32) -> RateWindow {
+        RateWindow::new(NonZeroU32::new(cap).unwrap())
+    }
+
+    fn push_n(w: &mut RateWindow, o: Outcome, n: u32) {
+        for _ in 0..n {
+            w.push(o);
+        }
+    }
+
+    #[test]
+    fn empty_window_never_trips() {
+        assert!(!win(50).should_trip(10, 50), "no samples < min_calls");
+    }
+
+    #[test]
+    fn below_min_calls_never_trips_even_at_full_failure() {
+        let mut w = win(50);
+        push_n(&mut w, Outcome::Failure, 9); // 100% failure, but only 9 < min_calls 10
+        assert!(!w.should_trip(10, 50));
+    }
+
+    #[test]
+    fn all_failures_trips_exactly_at_min_calls() {
+        let mut w = win(50);
+        push_n(&mut w, Outcome::Failure, 9);
+        assert!(!w.should_trip(10, 50), "9 samples");
+        w.push(Outcome::Failure); // 10th
+        assert!(w.should_trip(10, 50), "reached min_calls at 100%");
+    }
+
+    #[test]
+    fn interleaved_fifty_percent_trips_at_threshold_fifty() {
+        let mut w = win(50);
+        for _ in 0..10 {
+            w.push(Outcome::Failure);
+            w.push(Outcome::Success);
+        } // 10 F + 10 S = 20 samples, rate 50%
+        assert!(
+            w.should_trip(10, 50),
+            "50% failure rate meets the 50% threshold (>= trips)"
+        );
+    }
+
+    #[test]
+    fn just_below_threshold_does_not_trip() {
+        let mut w = win(100);
+        push_n(&mut w, Outcome::Failure, 49);
+        push_n(&mut w, Outcome::Success, 51); // 49/100 = 49% < 50%
+        assert!(!w.should_trip(10, 50));
+    }
+
+    #[test]
+    fn eviction_keeps_the_failure_count_exact() {
+        let mut w = win(10);
+        push_n(&mut w, Outcome::Failure, 10); // window full, all failures
+        assert!(w.should_trip(10, 50));
+        push_n(&mut w, Outcome::Success, 10); // evicts all 10 failures
+        assert_eq!(w.len(), 10, "capacity holds");
+        assert!(!w.should_trip(10, 50), "window is now all successes");
+    }
+}