tidaldb/tidal-net/src/circuit_breaker.rs

//! Per-peer circuit breaker for gRPC connections.
//!
//! State machine: Closed → Open → `HalfOpen` → Closed.
//! After `threshold` consecutive failures, the breaker opens for `reset_duration`.
//! The first [`check`](CircuitBreaker::check) after the reset period transitions
//! to `HalfOpen` and admits **exactly one** probe: that single `check()` returns
//! `Ok(())`, and every subsequent `check()` returns [`CircuitOpenError`] until the
//! outstanding probe resolves via [`record_success`](CircuitBreaker::record_success)
//! (→ Closed) or [`record_failure`](CircuitBreaker::record_failure) (→ Open again).
//! This single-probe invariant is enforced by the state machine itself, not by a
//! separate flag: `check()` returns `Ok(())` only on the `Open → HalfOpen`
//! transition and errors for every call while already in `HalfOpen`, so being in
//! the fieldless `HalfOpen` variant *is* the "a probe is outstanding" marker. No
//! `in_flight` field is needed (or present).
//!
//! Struct-shaped error: `CircuitOpenError` carries no fields and represents the single "breaker is open" outcome; an enum with one variant would be ceremony with no readability gain.

use std::{
    sync::Mutex,
    time::{Duration, Instant},
};

/// Error returned when the circuit breaker is open and not yet ready to probe.
#[derive(Debug, Clone, thiserror::Error)]
#[error("circuit breaker is open")]
pub struct CircuitOpenError;

/// A read-only snapshot of the breaker's state for observability (m11p8).
///
/// Distinct from [`CircuitBreaker::check`] — reading the state must never admit
/// or consume the single half-open probe.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BreakerState {
    /// Requests flow normally.
    Closed,
    /// Open: requests are rejected until the reset period elapses.
    Open,
    /// A single probe has been admitted and is in flight.
    HalfOpen,
}

impl BreakerState {
    /// The metric encoding: 0 closed, 1 open, 2 half-open (matches the
    /// `tidaldb_cluster_peer_breaker_state` gauge).
    #[must_use]
    pub const fn as_gauge(self) -> u8 {
        match self {
            Self::Closed => 0,
            Self::Open => 1,
            Self::HalfOpen => 2,
        }
    }
}

/// A circuit breaker that tracks consecutive failures for a single peer.
pub struct CircuitBreaker {
    state: Mutex<CircuitState>,
    threshold: u32,
    reset_duration: Duration,
    /// Wall-time of the last gRPC round-trip that PROVED the peer is alive: a
    /// `record_success` (segment accepted) OR a `record_backpressure` (the peer
    /// replied `accepted=false` — its queue was full, but the RPC round-tripped,
    /// so its gRPC server is up). `record_failure` (a genuine transport error)
    /// deliberately does NOT refresh this: a dead/partitioned peer must let it
    /// go stale. Read by the leader's `/cluster/status` aggregator to tell a
    /// SLOW-but-alive peer (HTTP control-plane starved under an apply burst, yet
    /// still acking replication) apart from a GENUINELY unreachable one — the
    /// false-partition fix. `None` until the first round-trip.
    last_contact: Mutex<Option<Instant>>,
}

#[derive(Debug)]
enum CircuitState {
    Closed {
        consecutive_failures: u32,
    },
    Open {
        opened_at: Instant,
    },
    /// A single probe has been admitted and is in flight. `check()` returns
    /// `Ok(())` exactly once on the Open→`HalfOpen` transition; while in this
    /// state every further `check()` errors until `record_success` (→ Closed)
    /// or `record_failure` (→ Open) resolves the outstanding probe.
    HalfOpen,
}

impl CircuitBreaker {
    /// Create a new circuit breaker.
    #[must_use]
    pub const fn new(threshold: u32, reset_duration: Duration) -> Self {
        Self {
            state: Mutex::new(CircuitState::Closed {
                consecutive_failures: 0,
            }),
            threshold,
            reset_duration,
            last_contact: Mutex::new(None),
        }
    }

    /// Stamp "the peer responded just now" — called on any gRPC round-trip that
    /// reached the peer (success or backpressure). Fail-soft on lock poisoning
    /// (skip the stamp; a missed refresh only makes a live peer look slightly
    /// staler, never falsely alive).
    fn touch_contact(&self) {
        if let Ok(mut last) = self.last_contact.lock() {
            *last = Some(Instant::now());
        }
    }

    /// How long since the last gRPC round-trip that proved the peer alive, or
    /// `None` if it has never responded. The leader's status aggregator treats a
    /// peer as reachable-despite-HTTP-timeout iff this is `Some` and recent.
    ///
    /// Read-only; never admits or consumes the half-open probe. Returns `None`
    /// (treated as "no recent contact") on lock poisoning — fail toward the
    /// honest unreachable verdict rather than masking a real partition.
    #[must_use]
    pub fn last_contact_elapsed(&self) -> Option<Duration> {
        self.last_contact
            .lock()
            .ok()
            .and_then(|g| g.map(|t| t.elapsed()))
    }

    /// Check if a request is allowed.
    ///
    /// Returns `Ok(())` if the breaker is closed, or on the single Open→`HalfOpen`
    /// transition that admits one probe, and `Ok(())` on lock poisoning (fail-open:
    /// allow the request through).
    ///
    /// # Half-open single-probe invariant
    ///
    /// The first `check()` after `reset_duration` elapses transitions Open→`HalfOpen`
    /// and returns `Ok(())` — that is the one admitted probe. While that probe is
    /// outstanding the breaker stays `HalfOpen`, and **every subsequent `check()`
    /// returns [`CircuitOpenError`]** until the probe resolves via
    /// [`record_success`](Self::record_success) (→ Closed) or
    /// [`record_failure`](Self::record_failure) (→ Open). N concurrent callers can
    /// therefore never all probe a peer that just failed; exactly one gets through.
    ///
    /// # Errors
    ///
    /// Returns [`CircuitOpenError`] if the breaker is open and the reset period
    /// has not yet elapsed, or if it is `HalfOpen` with a probe already in flight.
    pub fn check(&self) -> Result<(), CircuitOpenError> {
        let Ok(mut state) = self.state.lock() else {
            tracing::warn!("circuit breaker lock poisoned; failing open");
            return Ok(());
        };
        match *state {
            CircuitState::Closed { .. } => Ok(()),
            // A probe is already outstanding; admit no more until it resolves.
            CircuitState::HalfOpen => Err(CircuitOpenError),
            CircuitState::Open { opened_at } => {
                if opened_at.elapsed() >= self.reset_duration {
                    // Admit exactly one probe: the transition itself is the
                    // single Ok. The next `check()` will see `HalfOpen` and error
                    // until this probe's record_success/record_failure resolves it.
                    *state = CircuitState::HalfOpen;
                    Ok(())
                } else {
                    Err(CircuitOpenError)
                }
            }
        }
    }

    /// Peek the current state for observability WITHOUT admitting a probe.
    ///
    /// Unlike [`check`](Self::check), this never transitions Open→HalfOpen — it
    /// reports `Open` even once `reset_duration` has elapsed but no probe has yet
    /// been admitted (the "tripped, awaiting first probe" logical state). For a
    /// metrics gauge that distinction is immaterial; what matters is that reading
    /// the gauge never steals the single probe `check()` would admit. Fail-open
    /// (reports `Closed`) on lock poisoning, consistent with `check`.
    #[must_use]
    pub fn query_state(&self) -> BreakerState {
        let Ok(state) = self.state.lock() else {
            return BreakerState::Closed;
        };
        match *state {
            CircuitState::Closed { .. } => BreakerState::Closed,
            CircuitState::Open { .. } => BreakerState::Open,
            CircuitState::HalfOpen => BreakerState::HalfOpen,
        }
    }

    /// Record a successful request. Resets the failure count.
    pub fn record_success(&self) {
        // The peer round-tripped an accepted segment: it is alive. Stamp this
        // BEFORE touching the state machine so the liveness signal is refreshed
        // even if the state lock is poisoned below.
        self.touch_contact();
        let Ok(mut state) = self.state.lock() else {
            tracing::warn!("circuit breaker lock poisoned; ignoring success");
            return;
        };
        *state = CircuitState::Closed {
            consecutive_failures: 0,
        };
    }

    /// Record a follower backpressure reply (`accepted=false`).
    ///
    /// Backpressure is normal flow control from a healthy-but-busy follower whose
    /// inbound channel is full — the WAL is durable on the leader and the follower
    /// catches up once capacity frees. Counting it as a failure would let routine
    /// congestion open the breaker and convert transient backpressure into a
    /// self-inflicted `reset_duration` replication stall (see `error.rs`:
    /// `SegmentRejected` is classified transient). Only genuine transport/gRPC
    /// errors call [`record_failure`](Self::record_failure).
    ///
    /// Effect by state:
    /// - **Closed:** no-op. It neither opens the breaker nor resets a genuine
    ///   failure streak (a follower alternating real errors with backpressure must
    ///   still trip the breaker — see `backpressure_does_not_reset_failure_streak`).
    /// - **`HalfOpen`:** **resolves the outstanding probe by closing the breaker.**
    ///   A backpressure reply proves the *transport* is healthy (the RPC round-tripped;
    ///   only the follower's queue was full), so the half-open probe succeeded at the
    ///   layer the breaker protects. Resolving it is mandatory: under the single-probe
    ///   invariant a `HalfOpen` breaker admits no further `check()` until the probe is
    ///   resolved, so a no-op here would wedge the breaker permanently the first time a
    ///   probe drew a backpressure reply.
    /// - **Open:** no-op (no probe to resolve).
    pub fn record_backpressure(&self) {
        // A backpressure reply (`accepted=false`) means the RPC round-tripped —
        // only the follower's queue was full. That PROVES the peer's gRPC server
        // is alive, so refresh the liveness stamp here exactly as `record_success`
        // does. This is the load-bearing case for the false-partition fix: under a
        // sustained apply burst the follower's inbound channel fills and every
        // ship draws backpressure, so success stamps stop — but the peer is very
        // much alive and this keeps its liveness fresh.
        self.touch_contact();
        let Ok(mut state) = self.state.lock() else {
            tracing::warn!("circuit breaker lock poisoned; ignoring backpressure");
            return;
        };
        // Only a HalfOpen probe is resolved; Closed/Open are left untouched so
        // backpressure never opens the breaker nor resets a genuine failure streak.
        if matches!(*state, CircuitState::HalfOpen) {
            *state = CircuitState::Closed {
                consecutive_failures: 0,
            };
        }
    }

    /// Record a failed request. Increments the failure count and opens if threshold reached.
    pub fn record_failure(&self) {
        let Ok(mut state) = self.state.lock() else {
            tracing::warn!("circuit breaker lock poisoned; ignoring failure");
            return;
        };
        match *state {
            CircuitState::Closed {
                consecutive_failures,
            } => {
                let new_count = consecutive_failures + 1;
                if new_count >= self.threshold {
                    *state = CircuitState::Open {
                        opened_at: Instant::now(),
                    };
                } else {
                    *state = CircuitState::Closed {
                        consecutive_failures: new_count,
                    };
                }
            }
            CircuitState::HalfOpen => {
                // Probe failed; re-open.
                *state = CircuitState::Open {
                    opened_at: Instant::now(),
                };
            }
            CircuitState::Open { .. } => {
                // Already open, nothing to do.
            }
        }
    }

    /// Record a ship RPC that exceeded its deadline WITHOUT a reply
    /// (`tonic::Code::DeadlineExceeded`) — distinct from a genuine transport error.
    ///
    /// A timeout is AMBIGUOUS: a slow-but-ALIVE follower and a genuinely blackholed
    /// peer both produce it. Under a sustained 1536-D apply burst a follower's
    /// transport runtime can be momentarily starved by the CPU-heavy HNSW apply on
    /// its single segment-receiver thread, so an individual ship misses the leader's
    /// `request_timeout` even though the follower is up and still acking other ships.
    /// Counting that as a [`record_failure`](Self::record_failure) is the write-burst
    /// false-partition: both followers' breakers latch Open for `reset_duration`, the
    /// commit index stalls, ack=quorum writes 503, and the retries re-burst the same
    /// starved followers — no self-heal.
    ///
    /// We disambiguate with the same liveness stamp the status aggregator uses: a
    /// timeout received no reply so it does NOT refresh `last_contact`; instead it
    /// opens the breaker ONLY when there is no RECENT proof of life (no round-tripped
    /// success/backpressure within `reset_duration`). That preserves genuine-blackhole
    /// detection (a peer that round-trips nothing for `reset_duration` IS effectively
    /// unreachable, and a real severance also surfaces as a connect-level
    /// `Unavailable`/transport error → [`record_failure`](Self::record_failure), not a
    /// deadline) while a follower that keeps acking (success/backpressure refresh the
    /// stamp) is never ejected for a slow ship.
    ///
    /// Effect by state:
    /// - **Closed, recent contact:** neutral no-op — neither advances toward Open nor
    ///   resets a genuine failure streak (mirrors [`record_backpressure`](Self::record_backpressure)).
    /// - **Closed, stale/no contact:** counts toward Open exactly like a failure.
    /// - **`HalfOpen`:** re-opens (conservative) — a timed-out probe proves nothing and
    ///   must not wedge `HalfOpen` (which would admit no further probes).
    /// - **Open:** no-op.
    pub fn record_timeout(&self) {
        // No reply arrived, so (unlike success/backpressure) we do NOT touch_contact.
        let recent = self
            .last_contact_elapsed()
            .is_some_and(|e| e < self.reset_duration);
        let Ok(mut state) = self.state.lock() else {
            tracing::warn!("circuit breaker lock poisoned; ignoring timeout");
            return;
        };
        match *state {
            CircuitState::Closed {
                consecutive_failures,
            } => {
                if recent {
                    tracing::debug!(
                        "ship deadline-exceeded but peer proved liveness within reset window; \
                         classifying slow-but-alive (breaker neutral)"
                    );
                } else {
                    tracing::debug!(
                        "ship deadline-exceeded with stale liveness; counting toward breaker open"
                    );
                    let new_count = consecutive_failures + 1;
                    *state = if new_count >= self.threshold {
                        CircuitState::Open {
                            opened_at: Instant::now(),
                        }
                    } else {
                        CircuitState::Closed {
                            consecutive_failures: new_count,
                        }
                    };
                }
            }
            CircuitState::HalfOpen => {
                // Re-open rather than leave a wedged HalfOpen: a timed-out probe is
                // inconclusive, and the reset timer re-arms for the next probe.
                *state = CircuitState::Open {
                    opened_at: Instant::now(),
                };
            }
            CircuitState::Open { .. } => {}
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn closed_allows_requests() {
        let cb = CircuitBreaker::new(3, Duration::from_secs(30));
        assert!(cb.check().is_ok());
    }

    #[test]
    fn opens_after_threshold_failures() {
        let cb = CircuitBreaker::new(3, Duration::from_secs(30));
        cb.record_failure();
        cb.record_failure();
        assert!(cb.check().is_ok()); // 2 failures, threshold is 3
        cb.record_failure();
        assert!(cb.check().is_err()); // 3 failures, now open
    }

    #[test]
    fn success_resets_failure_count() {
        let cb = CircuitBreaker::new(3, Duration::from_secs(30));
        cb.record_failure();
        cb.record_failure();
        cb.record_success();
        cb.record_failure();
        cb.record_failure();
        assert!(cb.check().is_ok()); // reset happened, only 2 failures since
    }

    #[test]
    fn transitions_to_half_open_after_reset() {
        let cb = CircuitBreaker::new(2, Duration::from_millis(1));
        cb.record_failure();
        cb.record_failure();
        assert!(cb.check().is_err()); // open

        std::thread::sleep(Duration::from_millis(5));
        assert!(cb.check().is_ok()); // half-open after reset period
    }

    #[test]
    fn half_open_failure_reopens() {
        let cb = CircuitBreaker::new(2, Duration::from_millis(1));
        cb.record_failure();
        cb.record_failure();

        std::thread::sleep(Duration::from_millis(5));
        assert!(cb.check().is_ok()); // half-open
        cb.record_failure(); // probe failed
        assert!(cb.check().is_err()); // re-opened
    }

    #[test]
    fn half_open_admits_exactly_one_probe() {
        // Single-probe invariant: once `check()` transitions Open→HalfOpen and
        // returns Ok (the one admitted probe), every subsequent `check()` must
        // error until the probe resolves. Without this, N concurrent callers in
        // half-open would all probe a peer that just failed.
        let cb = CircuitBreaker::new(2, Duration::from_millis(1));
        cb.record_failure();
        cb.record_failure();
        assert!(cb.check().is_err(), "breaker should be open");

        std::thread::sleep(Duration::from_millis(5));

        // First check after reset admits the single probe.
        assert!(
            cb.check().is_ok(),
            "first half-open check admits exactly one probe"
        );
        // The probe is still outstanding: no success/failure recorded yet.
        assert!(
            cb.check().is_err(),
            "second half-open check must error while the probe is in flight"
        );
        assert!(
            cb.check().is_err(),
            "further half-open checks keep erroring until the probe resolves"
        );

        // Resolving the probe with success closes the breaker and re-admits traffic.
        cb.record_success();
        assert!(cb.check().is_ok(), "closed after probe succeeds");
    }

    #[test]
    fn half_open_probe_failure_then_check_stays_closed_to_traffic() {
        // The complementary resolution path: if the single probe fails, the
        // breaker re-opens and `check()` errors (now because it is Open, not
        // because a probe is in flight).
        let cb = CircuitBreaker::new(2, Duration::from_millis(1));
        cb.record_failure();
        cb.record_failure();

        std::thread::sleep(Duration::from_millis(5));
        assert!(cb.check().is_ok(), "single probe admitted");
        assert!(cb.check().is_err(), "second check errors: probe in flight");

        cb.record_failure(); // probe failed → re-open
        assert!(cb.check().is_err(), "re-opened after probe failure");
    }

    #[test]
    fn half_open_success_closes() {
        let cb = CircuitBreaker::new(2, Duration::from_millis(1));
        cb.record_failure();
        cb.record_failure();

        std::thread::sleep(Duration::from_millis(5));
        assert!(cb.check().is_ok()); // half-open
        cb.record_success(); // probe succeeded
        assert!(cb.check().is_ok()); // closed
    }

    #[test]
    fn backpressure_never_opens_breaker() {
        // CRITICAL-7 regression: a healthy-but-busy follower replying
        // `accepted=false` is the path `send_to` routes to `record_backpressure`.
        // Far more than `threshold` consecutive backpressure replies must leave
        // the breaker closed so the leader keeps shipping (no self-inflicted
        // replication stall).
        let cb = CircuitBreaker::new(5, Duration::from_secs(30));
        for _ in 0..10 {
            cb.record_backpressure();
        }
        assert!(
            cb.check().is_ok(),
            "backpressure must not open the breaker; a subsequent send must still be permitted"
        );
    }

    #[test]
    fn half_open_backpressure_resolves_probe_and_closes() {
        // A backpressure reply to the single half-open probe proves the transport
        // is healthy (the RPC round-tripped; only the follower's queue was full),
        // so it must resolve the probe by closing the breaker. Otherwise the
        // single-probe invariant would wedge the breaker in HalfOpen forever the
        // first time a probe drew `accepted=false`.
        let cb = CircuitBreaker::new(2, Duration::from_millis(1));
        cb.record_failure();
        cb.record_failure();

        std::thread::sleep(Duration::from_millis(5));
        assert!(cb.check().is_ok(), "single probe admitted");
        assert!(cb.check().is_err(), "second check errors: probe in flight");

        cb.record_backpressure(); // probe round-tripped, follower just busy
        assert!(
            cb.check().is_ok(),
            "backpressure resolves the half-open probe and re-admits traffic"
        );
    }

    #[test]
    fn genuine_failures_still_open_breaker() {
        // The opposite arm of `send_to` (`Err(status)`) calls `record_failure`,
        // which must still open the breaker after `threshold` consecutive
        // transport errors — backpressure handling must not weaken this.
        let cb = CircuitBreaker::new(5, Duration::from_secs(30));
        for _ in 0..5 {
            cb.record_failure();
        }
        assert!(
            cb.check().is_err(),
            "genuine transport errors must still open the breaker at threshold"
        );
    }

    #[test]
    fn backpressure_does_not_count_toward_open_among_failures() {
        // Backpressure must be neutral: interleaving it with genuine failures
        // neither advances the failure count toward the threshold nor resets it.
        // Here 4 real failures + many backpressures stays below the 5-failure
        // threshold (closed), and the 5th real failure opens it.
        let cb = CircuitBreaker::new(5, Duration::from_secs(30));
        for _ in 0..4 {
            cb.record_failure();
            cb.record_backpressure();
        }
        assert!(
            cb.check().is_ok(),
            "4 failures interleaved with backpressure must remain below threshold"
        );
        cb.record_failure(); // 5th genuine failure
        assert!(
            cb.check().is_err(),
            "the 5th genuine failure must open the breaker"
        );
    }

    #[test]
    fn success_refreshes_last_contact() {
        // The false-partition liveness signal: an accepted ship proves the peer
        // is alive, so `last_contact_elapsed` becomes `Some(~0)`.
        let cb = CircuitBreaker::new(3, Duration::from_secs(30));
        assert!(
            cb.last_contact_elapsed().is_none(),
            "no contact before the first round-trip"
        );
        cb.record_success();
        let elapsed = cb
            .last_contact_elapsed()
            .expect("a success stamps last_contact");
        assert!(
            elapsed < Duration::from_secs(1),
            "freshly stamped contact must read as recent: {elapsed:?}"
        );
    }

    #[test]
    fn backpressure_refreshes_last_contact() {
        // The LOAD-BEARING case for the apply-burst fix: under sustained
        // backpressure the follower replies `accepted=false` (gRPC round-tripped,
        // its queue was just full). That proves it is ALIVE, so the liveness
        // stamp must refresh exactly as a success does — otherwise a busy-but-up
        // follower would be misread as partitioned the moment success stamps stop.
        let cb = CircuitBreaker::new(5, Duration::from_secs(30));
        assert!(cb.last_contact_elapsed().is_none());
        cb.record_backpressure();
        let elapsed = cb
            .last_contact_elapsed()
            .expect("backpressure stamps last_contact (the peer responded)");
        assert!(
            elapsed < Duration::from_secs(1),
            "backpressure contact must read as recent: {elapsed:?}"
        );
    }

    #[test]
    fn failure_does_not_refresh_last_contact() {
        // A genuine transport error (a dead/partitioned peer) must NOT refresh
        // the liveness stamp — that is what lets the leader's status aggregator
        // tell a real partition (stamp goes stale) from a slow-but-alive peer.
        let cb = CircuitBreaker::new(3, Duration::from_secs(30));
        // First a real contact, so there IS a stamp to (not) refresh.
        cb.record_success();
        let after_success = cb.last_contact_elapsed().expect("stamped by success");
        // Now several genuine failures: the stamp must only AGE, never reset.
        for _ in 0..5 {
            cb.record_failure();
        }
        let after_failures = cb
            .last_contact_elapsed()
            .expect("the prior stamp is retained, never cleared");
        assert!(
            after_failures >= after_success,
            "failures must not move the stamp forward (it can only age): \
             {after_failures:?} >= {after_success:?}"
        );
    }

    #[test]
    fn never_contacted_peer_has_no_stamp() {
        // A peer the leader has never round-tripped reports no contact, so the
        // status aggregator treats it as unreachable (the honest default — no
        // evidence of life).
        let cb = CircuitBreaker::new(3, Duration::from_secs(30));
        assert!(
            cb.last_contact_elapsed().is_none(),
            "a peer with no gRPC round-trip has no liveness evidence"
        );
    }

    #[test]
    fn backpressure_does_not_reset_failure_streak() {
        // Backpressure must not reset the consecutive-failure count the way
        // `record_success` does — otherwise a follower alternating real errors
        // with backpressure could never trip the breaker. 4 failures, a
        // backpressure, then a 5th failure must open it.
        let cb = CircuitBreaker::new(5, Duration::from_secs(30));
        for _ in 0..4 {
            cb.record_failure();
        }
        cb.record_backpressure();
        assert!(cb.check().is_ok()); // still 4 failures, below threshold
        cb.record_failure(); // 5th consecutive genuine failure
        assert!(
            cb.check().is_err(),
            "backpressure must not reset the failure streak; the 5th failure opens the breaker"
        );
    }

    #[test]
    fn timeout_with_recent_contact_never_opens() {
        // THE write-burst false-partition fix: a follower that round-tripped a reply
        // recently (success/backpressure) is slow-but-ALIVE, so a burst of ship
        // deadline-exceeds must NOT open the breaker. Far more than `threshold`
        // timeouts stay closed while liveness is fresh.
        let cb = CircuitBreaker::new(5, Duration::from_secs(30));
        cb.record_success(); // recent proof of life
        for _ in 0..20 {
            cb.record_timeout();
        }
        assert!(
            cb.check().is_ok(),
            "timeouts from a peer with recent liveness must not open the breaker"
        );
    }

    #[test]
    fn timeout_without_proof_of_life_opens_like_failure() {
        // Genuine-blackhole detection preserved: a peer the leader has never
        // round-tripped (no liveness stamp) that only times out IS effectively
        // unreachable, so timeouts accrue toward Open exactly like failures.
        let cb = CircuitBreaker::new(5, Duration::from_secs(30));
        for _ in 0..5 {
            cb.record_timeout();
        }
        assert!(
            cb.check().is_err(),
            "timeouts with no proof of life must open the breaker (blackhole detection)"
        );
    }

    #[test]
    fn timeout_after_liveness_goes_stale_opens() {
        // Recent contact suppresses timeout-driven opening; once the stamp ages past
        // `reset_duration` with no round-trips at all, timeouts resume opening.
        let cb = CircuitBreaker::new(3, Duration::from_millis(50));
        cb.record_success(); // stamp liveness
        std::thread::sleep(Duration::from_millis(60)); // stamp now older than reset_duration
        for _ in 0..3 {
            cb.record_timeout();
        }
        assert_eq!(
            cb.query_state(),
            BreakerState::Open,
            "once liveness is stale, timeouts must open the breaker"
        );
    }

    #[test]
    fn timeout_does_not_refresh_last_contact() {
        // A timeout received NO reply, so (unlike success/backpressure) it must not
        // stamp liveness — otherwise a dead peer's own timeouts would keep it
        // perpetually 'recent' and mask a genuine partition.
        let cb = CircuitBreaker::new(3, Duration::from_secs(30));
        assert!(cb.last_contact_elapsed().is_none());
        cb.record_timeout();
        assert!(
            cb.last_contact_elapsed().is_none(),
            "a timeout (no reply) must not refresh the liveness stamp"
        );
    }

    #[test]
    fn timeout_neutral_among_failures_with_recent_contact() {
        // With recent contact a timeout is neutral: it neither advances the failure
        // streak toward Open nor resets it. 4 failures + a backpressure (stamps
        // recent liveness) + a timeout stays below threshold; the 5th failure opens.
        let cb = CircuitBreaker::new(5, Duration::from_secs(30));
        for _ in 0..4 {
            cb.record_failure();
        }
        cb.record_backpressure(); // stamps recent liveness; neutral to the streak
        cb.record_timeout(); // recent -> neutral, streak stays at 4
        assert!(
            cb.check().is_ok(),
            "4 failures + a neutral timeout must stay below threshold"
        );
        cb.record_failure(); // 5th genuine failure
        assert!(
            cb.check().is_err(),
            "the 5th genuine failure must still open the breaker"
        );
    }

    #[test]
    fn half_open_timeout_reopens_without_wedging() {
        // A timed-out probe is inconclusive; it must re-open (re-arm the reset
        // timer) rather than leave a wedged HalfOpen that admits no further probes.
        let cb = CircuitBreaker::new(2, Duration::from_millis(1));
        cb.record_failure();
        cb.record_failure();
        std::thread::sleep(Duration::from_millis(5));
        assert!(cb.check().is_ok(), "half-open probe admitted");
        assert_eq!(cb.query_state(), BreakerState::HalfOpen);
        cb.record_timeout(); // probe timed out
        assert_eq!(
            cb.query_state(),
            BreakerState::Open,
            "a half-open timeout must re-open, never wedge HalfOpen"
        );
    }
}