diff --git a/crates/buzz-acp/src/backoff.rs b/crates/buzz-acp/src/backoff.rs new file mode 100644 index 0000000000..5377ccb825 --- /dev/null +++ b/crates/buzz-acp/src/backoff.rs @@ -0,0 +1,157 @@ +//! Jitter helpers for retry, backoff, and rate-limit gate delays. +//! +//! Jitter uses the nanosecond sub-second component of the system clock as a +//! cheap entropy source (no `rand` dependency). The factor computation is a +//! pure function of that value so the reachable range can be tested directly — +//! a wall-clock helper cannot be driven to its domain boundary, which is how +//! the divisor defect below survived every existing bound assertion. +//! +//! Two shapes, and the distinction is load-bearing: +//! +//! * [`jittered_duration`] is **symmetric** (±20%). Correct for self-chosen +//! backoff ladders, where waking early only costs an extra attempt. +//! * [`extend_with`] is **one-sided** (+0–20%, never shorter). Required +//! wherever the base duration is an authoritative deadline supplied by the +//! relay: shortening a `retry in {N}s` hint wakes into a window the relay has +//! already told us is closed, which earns a fresh denial and burns a counter +//! increment for no chance of success. + +use std::time::Duration; + +/// Nanoseconds in a second — the true domain bound of `Duration::subsec_nanos`. +/// +/// The previous implementations divided by `u32::MAX` (4,294,967,295), which is +/// 4.295x larger than the largest value `subsec_nanos()` can return. That +/// capped the symmetric factor at 0.893 instead of approaching 1.2, making +/// "±20% jitter" unconditionally negative and averaging −15%. +const NANOS_PER_SEC: f64 = 1_000_000_000.0; + +/// Map a sub-second nanosecond count onto `[0.0, 1.0]`. +/// +/// Clamped so the function is total for any `u32`; inputs above one second are +/// unreachable from `subsec_nanos()` but must not push the factor out of range. +fn nanos_fraction(nanos: u32) -> f64 { + (f64::from(nanos) / NANOS_PER_SEC).min(1.0) +} + +/// Symmetric jitter factor in `[0.8, 1.2)` over the domain `[0, 1s)`. +pub(crate) fn symmetric_jitter_factor(nanos: u32) -> f64 { + 0.8 + nanos_fraction(nanos) * 0.4 +} + +/// One-sided jitter factor in `[1.0, 1.2)` over the domain `[0, 1s)`. +/// +/// Never returns less than 1.0, so a delay built from it can never fall below +/// its base. This is a structural guarantee, not a policy applied by callers. +pub(crate) fn nonnegative_jitter_factor(nanos: u32) -> f64 { + 1.0 + nanos_fraction(nanos) * 0.2 +} + +/// Sub-second component of the current wall clock, used as the jitter source. +pub(crate) fn clock_nanos() -> u32 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .subsec_nanos() +} + +/// Apply symmetric ±20% jitter to a self-chosen backoff duration. +pub(crate) fn jittered_duration(base: Duration) -> Duration { + base.mul_f64(symmetric_jitter_factor(clock_nanos())) +} + +/// Extend an authoritative deadline by 0–20%, never shortening it. +/// +/// The entropy sample is an explicit parameter so tests can drive the whole +/// domain deterministically. With a wall-clock-only helper the "never +/// shortens" property is only probabilistically testable: a symmetric-jitter +/// regression satisfies it on roughly half of all draws, which reads as a +/// surviving mutant rather than a flaky assertion. Callers pass +/// [`clock_nanos`]. +pub(crate) fn extend_with(base: Duration, nanos: u32) -> Duration { + base.mul_f64(nonnegative_jitter_factor(nanos)) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The largest value `Duration::subsec_nanos()` can return. Asserting the + /// ceiling at `u32::MAX` instead would re-encode the very divisor defect + /// these tests exist to catch, so the domain bound is pinned explicitly. + const MAX_SUBSEC_NANOS: u32 = 999_999_999; + + /// Symmetric factor reaches its documented floor and (near) ceiling. + /// + /// The ceiling assertion is the discriminating one: the `u32::MAX` divisor + /// caps this at 0.893, while every range assertion of the form + /// `0.8 <= f < 1.2` passes in both the broken and the fixed implementation. + #[test] + fn symmetric_factor_spans_its_documented_range() { + assert_eq!(symmetric_jitter_factor(0), 0.8, "floor must be exactly 0.8"); + let ceiling = symmetric_jitter_factor(MAX_SUBSEC_NANOS); + assert!( + ceiling > 1.199 && ceiling < 1.2, + "ceiling {ceiling} must approach 1.2 from below — a ceiling near 0.893 \ + means the factor is divided by u32::MAX instead of 1e9" + ); + } + + /// One-sided factor never shortens its base and reaches +20%. + #[test] + fn nonnegative_factor_spans_its_documented_range() { + assert_eq!( + nonnegative_jitter_factor(0), + 1.0, + "floor must be exactly 1.0 — a one-sided factor may never shorten" + ); + let ceiling = nonnegative_jitter_factor(MAX_SUBSEC_NANOS); + assert!( + ceiling > 1.199 && ceiling < 1.2, + "ceiling {ceiling} must approach 1.2 from below" + ); + } + + /// No input can make the one-sided factor shorten a deadline. Swept across + /// the whole `u32` range, not just the reachable sub-second domain. + #[test] + fn nonnegative_factor_is_never_below_one() { + for step in 0..=1_000u32 { + let nanos = (u32::MAX / 1_000).saturating_mul(step); + let factor = nonnegative_jitter_factor(nanos); + assert!( + (1.0..=1.2 + 1e-9).contains(&factor), + "factor {factor} out of [1.0, 1.2] at nanos={nanos}" + ); + } + } + + /// Both factors stay in range for out-of-domain inputs (clamped). + /// + /// `subsec_nanos()` can never return these values; the clamp exists so the + /// factor functions are total. Compared with a tolerance because + /// `0.8 + 1.0 * 0.4` is not exactly 1.2 in binary floating point. + #[test] + fn factors_are_clamped_above_one_second() { + assert!((symmetric_jitter_factor(u32::MAX) - 1.2).abs() < 1e-9); + assert!((nonnegative_jitter_factor(u32::MAX) - 1.2).abs() < 1e-9); + } + + /// The wrappers apply their factor to the base duration. + #[test] + fn wrappers_stay_within_their_factor_ranges() { + let base = Duration::from_secs(5); + for _ in 0..64 { + let symmetric = jittered_duration(base); + assert!( + symmetric >= base.mul_f64(0.8) && symmetric <= base.mul_f64(1.2), + "symmetric {symmetric:?} out of range" + ); + let extended = extend_with(base, clock_nanos()); + assert!( + extended >= base && extended <= base.mul_f64(1.2), + "extended {extended:?} shortened the base" + ); + } + } +} diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 811253e4ac..eb848d742c 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1,6 +1,7 @@ #![deny(unsafe_code)] mod acp; +mod backoff; mod config; mod engram_fetch; mod filter; @@ -1086,13 +1087,7 @@ impl SlotCircuit { // Exponential backoff: 1s * 2^(recent-1), capped at 30s, with ±20% jitter. let base = RESPAWN_BASE_DELAY.saturating_mul(1u32 << (recent - 1).min(5)); let capped = base.min(RESPAWN_MAX_DELAY); - let jitter = (std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .subsec_nanos() as f64) - / 1_000_000_000.0; // 0.0..1.0 - let factor = 0.8 + jitter * 0.4; // 0.8..1.2 - CrashVerdict::Respawn(capped.mul_f64(factor)) + CrashVerdict::Respawn(crate::backoff::jittered_duration(capped)) } /// Mark a spawn failure — opens the circuit so the slot isn't retried diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 5c960de202..29787ee334 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -453,15 +453,7 @@ impl EventQueue { // Exponential backoff: BASE * 2^(attempt-1), capped at MAX, with ±20% jitter. let base_secs = BASE_RETRY_DELAY_SECS.saturating_mul(1u64 << (attempt - 1).min(6)); let capped_secs = base_secs.min(MAX_RETRY_DELAY_SECS); - // Jitter: multiply by 0.8..1.2 using subsecond nanos as entropy source. - let jitter = { - let nanos = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .subsec_nanos(); - 0.8 + (nanos as f64 / u32::MAX as f64) * 0.4 - }; - let delay = Duration::from_secs_f64(capped_secs as f64 * jitter); + let delay = crate::backoff::jittered_duration(Duration::from_secs(capped_secs)); tracing::warn!( channel_id = %channel_id, diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index aea5cee077..b1b391fec4 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -126,6 +126,7 @@ use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, Web use tracing::{debug, info, warn}; use uuid::Uuid; +use crate::backoff::{extend_with, jittered_duration}; use crate::config::ChannelFilter; /// Metadata about a channel, populated at discovery time. @@ -251,6 +252,61 @@ const REST_RETRY_BASE_DELAYS: [Duration; 3] = [ Duration::from_millis(2000), ]; +/// Upper bound on a REST retry sleep honouring a relay `retry in {N}s` hint. +/// +/// The ladder's own rungs top out at 2s, far below the relay's 60s quota +/// window, so a hint must be allowed to exceed them or honouring it is +/// pointless. The cap exists only to bound a pathological hint; it sits above +/// the longest window so a legitimate hint is never truncated into it. The +/// `const` assertion pins that property rather than the literal. +const REST_RETRY_HINT_MAX: Duration = Duration::from_secs(90); +const _: () = assert!( + REST_RETRY_HINT_MAX.as_secs() >= 60, + "the hint cap must outlast the relay's longest quota window, or a client \ + told to wait out a 60s window wakes inside it and is denied again" +); + +/// The delay before the next REST attempt. +/// +/// `hint_secs` is the relay's parsed `retry in {N}s` value when the previous +/// attempt was refused with a rate-limit response, and `None` otherwise. +/// +/// Without a hint this is the self-chosen ladder rung with symmetric jitter: +/// waking early merely costs another attempt. With one, the relay has named +/// the window TTL, and the ladder's sub-2s rungs are all shorter than any +/// window it would name — retrying on them guarantees a denial that still +/// costs a counter increment, because the limiter's `INCR` runs on refused +/// checks too. So the hint wins whenever it is longer, and it is extended +/// one-sided so jitter can never pull the wake-up back inside the window. +/// +/// Pure, with entropy as a parameter, for the same reason as `gate_delay`: +/// the "never wakes before the hint" property is only meaningfully asserted +/// if the jitter endpoints can be driven rather than sampled. +/// +/// Reachability: most REST callers wrap these requests in their own short +/// `tokio::time::timeout` (500ms–5s), so a multi-second hint sleep is +/// cancelled by the caller's budget rather than slept out. That is the +/// intended direction — the alternative was retrying inside a window the +/// relay had closed, which is denied anyway and still costs a counter +/// increment, so the caller fails either way and the relay is charged less. +/// The unbudgeted callers (the setup nudge, engram fetch) get the full +/// benefit. +fn rest_retry_delay(rung: Duration, hint_secs: Option, jitter_nanos: u32) -> Duration { + match hint_secs { + Some(secs) => { + let hint = Duration::from_secs(secs).min(REST_RETRY_HINT_MAX); + // A hint shorter than the rung we would have waited anyway is no + // reason to retry sooner. + if hint > rung { + extend_with(hint, jitter_nanos) + } else { + jittered_duration(rung) + } + } + None => jittered_duration(rung), + } +} + fn unix_now_secs() -> u64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -322,13 +378,17 @@ impl RestClient { Fut: std::future::Future>, { let mut last_err = None; + // The relay's `retry in {N}s` hint from the previous attempt, if it was + // refused with a rate-limit response. + let mut retry_hint_secs: Option = None; for (attempt, delay) in std::iter::once(None) .chain(REST_RETRY_BASE_DELAYS.iter().map(|d| Some(*d))) .enumerate() { if let Some(base) = delay { - let jittered = jittered_duration(base); + let jittered = + rest_retry_delay(base, retry_hint_secs, crate::backoff::clock_nanos()); tracing::debug!( "retrying {method} {path} (attempt {attempt}) in {:.1}s", jittered.as_secs_f64() @@ -336,14 +396,34 @@ impl RestClient { tokio::time::sleep(jittered).await; } - match build_request().await { + // Every arm yields the hint that governs the NEXT sleep, so each + // outcome states its own policy instead of inheriting one from + // wherever an assignment happens to sit. + retry_hint_secs = match build_request().await { Ok(resp) if resp.status().is_success() => return Ok(resp), Ok(resp) if is_retriable_status(resp.status()) => { let status = resp.status(); - tracing::warn!("{method} {path} returned retriable HTTP {status}"); + // Read the body on a 429 for the relay's `retry in {N}s` + // hint. Only the sleep length depends on it, so a body we + // cannot read simply leaves the ladder rung in charge. + let hint = if status == reqwest::StatusCode::TOO_MANY_REQUESTS { + resp.text() + .await + .ok() + .as_deref() + .and_then(parse_rate_limit_retry_secs) + } else { + None + }; + if let Some(secs) = hint { + tracing::warn!("{method} {path} rate-limited, relay asked for {secs}s"); + } else { + tracing::warn!("{method} {path} returned retriable HTTP {status}"); + } last_err = Some(RelayError::Http(format!( "{method} {path} returned HTTP {status}" ))); + hint } Ok(resp) => { return Err(RelayError::Http(format!( @@ -355,9 +435,27 @@ impl RestClient { Err(e) if e.is_timeout() || e.is_connect() => { tracing::warn!("{method} {path} network error: {e}"); last_err = Some(RelayError::Http(e.to_string())); + // Deliberately keep the previous 429's hint. A network + // error says nothing about the quota window, so dropping + // the hint here would send the next attempt back to a + // sub-2s rung and inside a window the relay already + // closed. The cost is one-sided: over-sleeping wastes + // time we were told to wait anyway, while under-sleeping + // earns a fresh denial that still costs a counter + // increment, because the limiter's `INCR` runs on refused + // checks too. + // + // This is only exactly right for the sleep immediately + // after the 429. If the network error itself burned most + // of the hinted window we over-sleep by up to a second + // window — accepted, same cheap direction. Do not + // "optimize" that by subtracting elapsed time: the + // subtraction reintroduces under-sleep whenever the clock + // arithmetic is off, which is the expensive direction. + retry_hint_secs } Err(e) => return Err(RelayError::Http(e.to_string())), - } + }; } Err(last_err @@ -1161,15 +1259,31 @@ impl BgState { /// the desktop TypeScript client, which uses a 10s no-hint default — both /// values are conservative enough; the relay hint wins when present. /// + /// Jitter here is **one-sided** ([`crate::backoff::extend_with`]): the hint is the + /// relay's authoritative window TTL, so the gate may only ever be longer + /// than it. Symmetric jitter would wake us inside a window the relay has + /// already said is closed, earning a fresh denial plus a wasted counter + /// increment (the limiter's `INCR` runs on denied checks too). + /// /// The gate takes the **maximum** of any existing deadline and the newly /// computed one so overlapping CLOSED/NOTICE messages can't shorten a gate /// that is already set further out. /// /// Returns the gate deadline that was set. fn set_rate_limit_gate(&mut self, retry_secs: u64) -> tokio::time::Instant { - let secs = if retry_secs < 2 { 5 } else { retry_secs }; - let base = Duration::from_secs(secs); - let deadline = tokio::time::Instant::now() + jittered_duration(base); + self.set_rate_limit_gate_with(retry_secs, crate::backoff::clock_nanos()) + } + + /// [`set_rate_limit_gate`] with an explicit jitter sample. + /// + /// Delegates the arithmetic to the pure [`gate_delay`]; this wrapper exists + /// only to apply the result to `self`. + fn set_rate_limit_gate_with( + &mut self, + retry_secs: u64, + jitter_nanos: u32, + ) -> tokio::time::Instant { + let deadline = tokio::time::Instant::now() + gate_delay(retry_secs, jitter_nanos); let gate = match self.rate_limit_gate { Some(existing) if existing > deadline => existing, _ => deadline, @@ -1498,11 +1612,28 @@ async fn execute_connected_command( // indicators are worthless and sending them would consume admission // budget the relay already rejected us on. // - // INVARIANT: apart from observer frames (parked above), the WS publish - // path carries only ephemeral kinds (typing indicators). The silent - // drop-while-gated relies on that invariant. If a future caller - // publishes durable events through this path, it must extend the - // kind guard above to avoid silently discarding user data. + // INVARIANT: the WS publish path carries only ephemeral kinds + // (presence 20001, typing 20002, observer frames 24200 — the last + // parked above rather than dropped). The silent drop-while-gated + // below is only correct under that invariant: dropping a durable + // event here would discard user data while reporting success to + // the caller, since `publish_event` returns `Ok` once the command + // is queued and never learns what happened to it. + // + // The assertion encodes the boundary rather than widening the + // guard. Note that `is_ephemeral` is not usable as the drop + // predicate here: observer frames are themselves range-ephemeral, + // so testing it would send them down the drop path and undo the + // parking above. A durable event reaching this point is a bug at + // the *caller* — it should use `RestClient::submit_event`, which + // surfaces the relay's real response (see `publish_setup_nudge`). + debug_assert!( + buzz_core::kind::is_ephemeral(u32::from(event.kind.as_u16())), + "durable kind {} reached the WS publish path, which silently \ + drops while rate-gated or disconnected; publish durable events \ + via RestClient::submit_event instead", + event.kind.as_u16() + ); if state.check_rate_gate().is_some() { debug!("rate-gated: dropping ephemeral PublishEvent (typing indicator)"); return true; @@ -3345,16 +3476,28 @@ pub(crate) fn parse_rate_limit_retry_secs(msg: &str) -> Option { after[..len].parse::().ok() } -/// Add ±20% jitter to a backoff duration using the nanosecond sub-second -/// component of the system clock as a cheap entropy source (no `rand` dep). -fn jittered_duration(base: Duration) -> Duration { - let nanos = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .subsec_nanos(); - // factor ∈ [0.8, 1.2) - let factor = 0.8 + (nanos as f64 / u32::MAX as f64) * 0.4; - base.mul_f64(factor) +/// Sub-2s hints (including a missing hint parsed as 0) floor to this many +/// seconds, matching the desktop TypeScript client's no-hint default. +const RATE_LIMIT_GATE_FLOOR_SECS: u64 = 5; + +/// The rate-limit gate delay for a relay hint and an explicit jitter sample. +/// +/// Pure: entropy enters only as `jitter_nanos`, and the wall clock is acquired +/// by the caller. That is what makes the gate's safety property testable at +/// exact endpoints — a helper that reads `SystemTime` internally can only be +/// sampled, never driven, and paused Tokio time does not freeze `SystemTime`. +/// +/// Jitter is **one-sided**: the hint is the relay's authoritative window TTL, +/// so the delay may only ever exceed it. Symmetric jitter would wake us inside +/// a window the relay has already said is closed, earning a fresh denial plus +/// a wasted counter increment (the limiter's `INCR` runs on denied checks too). +fn gate_delay(retry_secs: u64, jitter_nanos: u32) -> Duration { + let secs = if retry_secs < 2 { + RATE_LIMIT_GATE_FLOOR_SECS + } else { + retry_secs + }; + extend_with(Duration::from_secs(secs), jitter_nanos) } /// Classify a `RelayError` as a DNS resolution failure. @@ -4336,11 +4479,15 @@ mod tests { assert_eq!(sub_id, "ch-12345678-1234-5678-1234-567812345678"); } - /// Build a real signed Nostr event for testing BgState. + /// Build a real signed durable kind:1 note for testing BgState. /// /// Uses `custom_created_at` so tests can control the timestamp. /// The event ID is determined by the nostr signing process — we don't /// control it, but we return it so callers can use it for dedup tests. + /// + /// Not valid as a `RelayCommand::PublishEvent` payload for + /// `execute_connected_command`: that path carries ephemeral kinds only and + /// asserts as much. Use [`make_test_typing_event`] for publish-path tests. fn make_test_event(keys: &nostr::Keys, created_at_secs: u64) -> Event { let ts = nostr::Timestamp::from(created_at_secs); EventBuilder::new(nostr::Kind::TextNote, "test") @@ -4350,6 +4497,15 @@ mod tests { .expect("signing should succeed") } + /// A kind:20002 typing indicator — a realistic payload for the WS publish + /// path, which carries ephemeral kinds only. + fn make_test_typing_event(keys: &nostr::Keys) -> Event { + EventBuilder::new(Kind::Custom(KIND_TYPING_INDICATOR as u16), "") + .tags([Tag::parse(["h", &Uuid::new_v4().to_string()]).expect("h tag")]) + .sign_with_keys(keys) + .expect("signing should succeed") + } + async fn test_ws_pair() -> (WsStream, WebSocketStream) { let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await @@ -4513,7 +4669,8 @@ mod tests { let mut state = BgState::new(); let channel_id = Uuid::new_v4(); seed_test_subscription(&mut state, channel_id); - let event = make_test_event(&nostr::Keys::generate(), 2_000); + // A typing indicator: the WS publish path carries ephemeral kinds only. + let event = make_test_typing_event(&nostr::Keys::generate()); let event_id = event.id.to_hex(); let task = tokio::spawn(async move { @@ -5765,7 +5922,7 @@ mod tests { "gate must be active immediately after arming" ); - // Advance virtual time past the max jitter (1.2 × 5 s = 6 s). + // Advance virtual time past the max gate length (1.2 × 5 s = 6 s). tokio::time::advance(Duration::from_secs(7)).await; assert!( @@ -5778,6 +5935,337 @@ mod tests { ); } + /// The effective gate base for a hint: sub-2s hints floor to 5s. + #[cfg(test)] + fn effective_gate_secs(hint: u64) -> u64 { + if hint < 2 { + 5 + } else { + hint + } + } + + /// `gate_delay` is exact at both endpoints of the jitter domain. + /// + /// This is the discriminating test, and it is a **pure-function** test for + /// a reason. Entropy enters `gate_delay` only as a parameter, so both + /// endpoints are driven rather than sampled: + /// + /// * `nanos = 0` ⇒ one-sided factor exactly 1.0 ⇒ delay is exactly base. + /// A symmetric-jitter regression yields `0.8 * base` here, deterministically. + /// * `nanos = 999_999_999` ⇒ factor approaches 1.2 ⇒ delay approaches + /// `1.2 * base`. The `u32::MAX` divisor caps it at `0.893 * base`, + /// so this endpoint is what catches the divisor defect; a range assertion + /// of the form `0.8*base <= d < 1.2*base` cannot, because the broken range + /// is a strict subset of the asserted one. The upper bound is inclusive: + /// `Duration` is nanosecond-resolution, so e.g. `2s * 1.1999999998` + /// rounds to exactly `2.4s` and a strict `<` would be unsatisfiable. + /// + /// Scope: this kills a swap of one-sided for symmetric *inside* the seam, + /// and the divisor defect. A mutation that bypasses the seam entirely (call + /// site reverting to a wall-clock helper) is a structural change this unit + /// test cannot honestly claim to kill — `#[deny(unused)]` on the parameter + /// and the integration test below are what cover that. + #[test] + fn gate_delay_is_exact_at_both_jitter_endpoints() { + for hint in [0u64, 1, 2, 5, 6, 51] { + let base = Duration::from_secs(effective_gate_secs(hint)); + + assert_eq!( + gate_delay(hint, 0), + base, + "a {hint}s hint at zero jitter must be exactly {base:?}; \ + symmetric jitter would yield 0.8 x base here" + ); + + let ceiling = gate_delay(hint, 999_999_999); + assert!( + ceiling > base.mul_f64(1.1999) && ceiling <= base.mul_f64(1.2), + "a {hint}s hint at max jitter gave {ceiling:?}, which must reach \ + 1.2 x {base:?} — a value near 0.893 x base means the \ + factor is divided by u32::MAX instead of 1e9" + ); + } + } + + /// `gate_delay` never returns less than the relay's hint, for any sample. + #[test] + fn gate_delay_never_falls_below_the_hint() { + for hint in [0u64, 1, 2, 5, 6, 51] { + let base = Duration::from_secs(effective_gate_secs(hint)); + for step in 0..=1_000u32 { + let nanos = (999_999_999f64 * f64::from(step) / 1_000.0) as u32; + let delay = gate_delay(hint, nanos); + assert!( + delay >= base && delay <= base.mul_f64(1.2), + "a {hint}s hint with jitter nanos={nanos} gave {delay:?}, \ + outside [{base:?}, 1.2 x base]" + ); + } + } + } + + /// A relay hint longer than the ladder rung wins, and is never shortened. + /// + /// The ladder tops out at 2s while the relay's quota windows run to 60s, + /// so this is the case that matters: without it the client retries inside + /// a window the relay has already closed, is denied, and still pays a + /// counter increment for the attempt. + /// + /// Endpoints are driven, not sampled, for the same reason as `gate_delay`: + /// a symmetric-jitter regression satisfies a range assertion on about half + /// of all draws, which reads as a surviving mutant rather than a failure. + #[test] + fn rest_retry_delay_honours_a_hint_longer_than_the_rung() { + let rung = Duration::from_millis(500); + for hint in [1u64, 2, 5, 51, 60] { + let base = Duration::from_secs(hint); + + assert_eq!( + rest_retry_delay(rung, Some(hint), 0), + base, + "a {hint}s hint at zero jitter must be exactly {base:?} — \ + symmetric jitter would yield 0.8 x base and wake inside the window" + ); + + let ceiling = rest_retry_delay(rung, Some(hint), 999_999_999); + assert!( + ceiling > base.mul_f64(1.1999) && ceiling <= base.mul_f64(1.2), + "a {hint}s hint at max jitter gave {ceiling:?}, which must reach \ + 1.2 x {base:?}" + ); + } + } + + /// No jitter sample can pull a honoured hint below the relay's window. + #[test] + fn rest_retry_delay_never_wakes_before_the_hint() { + let rung = Duration::from_millis(500); + for hint in [1u64, 5, 51, 60] { + let base = Duration::from_secs(hint); + for step in 0..=1_000u32 { + let nanos = (999_999_999f64 * f64::from(step) / 1_000.0) as u32; + let delay = rest_retry_delay(rung, Some(hint), nanos); + assert!( + delay >= base && delay <= base.mul_f64(1.2), + "a {hint}s hint with jitter nanos={nanos} gave {delay:?}, \ + outside [{base:?}, 1.2 x base]" + ); + } + } + } + + /// With no hint, and with a hint shorter than the rung, the self-chosen + /// ladder rung governs — symmetric jitter is correct there because waking + /// early only costs an attempt nobody asked us to defer. + /// + /// The sub-rung hint case is the one worth pinning: `Some(0)` must not + /// collapse the delay to zero and spin the ladder. + #[test] + fn rest_retry_delay_falls_back_to_the_rung() { + let rung = Duration::from_secs(2); + for hint in [None, Some(0), Some(1), Some(2)] { + for nanos in [0u32, 500_000_000, 999_999_999] { + let delay = rest_retry_delay(rung, hint, nanos); + assert!( + delay >= rung.mul_f64(0.8) && delay <= rung.mul_f64(1.2), + "hint={hint:?} nanos={nanos} gave {delay:?}, which is not \ + the {rung:?} ladder rung with symmetric jitter" + ); + } + } + } + + /// A pathological hint is capped, and the cap still outlasts the window. + /// + /// The cap is what stops a bad or hostile hint parking the client for + /// hours; the `const` assertion beside it is what stops the cap being + /// retuned back below the 60s window the hint exists to outlast. + #[test] + fn rest_retry_delay_caps_a_pathological_hint() { + let rung = Duration::from_millis(500); + let delay = rest_retry_delay(rung, Some(86_400), 0); + assert_eq!( + delay, REST_RETRY_HINT_MAX, + "a day-long hint must be capped at {REST_RETRY_HINT_MAX:?}" + ); + assert!( + REST_RETRY_HINT_MAX >= Duration::from_secs(60), + "the cap must still outlast the relay's longest quota window" + ); + } + + /// The 429 hint actually reaches the sleep — the wiring, not the maths. + /// + /// The `rest_retry_delay` tests above are pure and cannot see the call + /// site: deleting the body read (so the hint is never parsed) leaves every + /// one of them green. This test is what kills that mutant. It serves a 429 + /// carrying a 51s hint, then a 200, and asserts the retry slept for the + /// hint rather than the 500ms ladder rung. + /// + /// Time is paused, so the sleep is virtual and the assertion is on the + /// clock the sleep advanced, not on wall-clock duration. + #[tokio::test(start_paused = true)] + async fn rest_retry_sleeps_for_the_relay_hint_not_the_ladder_rung() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind test HTTP server"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + // First attempt: refused with the relay's real rejection text. + // Second: accepted. + let bodies = [ + ( + "429 Too Many Requests", + r#"{"error":"rate-limited: quota exceeded (api); retry in 51s"}"#, + ), + ("200 OK", "{}"), + ]; + for (status, body) in bodies { + let Ok((mut socket, _)) = listener.accept().await else { + return; + }; + let mut buf = vec![0u8; 8192]; + let _ = socket.read(&mut buf).await; + let response = format!( + "HTTP/1.1 {status}\r\nContent-Type: application/json\r\n\ + Content-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let _ = socket.write_all(response.as_bytes()).await; + } + }); + + let client = RestClient { + http: reqwest::Client::new(), + base_url, + keys: Keys::generate(), + auth_tag_json: None, + }; + + let start = tokio::time::Instant::now(); + let resp = client + .submit_event(&make_test_event(&Keys::generate(), 1_000)) + .await; + let elapsed = start.elapsed(); + + assert!(resp.is_ok(), "the second attempt succeeds: {resp:?}"); + assert!( + elapsed >= Duration::from_secs(51), + "the retry slept {elapsed:?}, but the relay asked for 51s — a sleep \ + near the 500ms ladder rung means the hint was never read from the \ + 429 body, and the retry lands inside the window and is denied" + ); + assert!( + elapsed < Duration::from_secs(70), + "slept {elapsed:?}: the hint should be honoured, not compounded" + ); + server.abort(); + } + + /// A network error after a 429 keeps the relay's hint for the next sleep. + /// + /// This pins the policy in the `is_timeout() || is_connect()` arm, which + /// is otherwise invisible: the arm assigns the hint it already held, so + /// deleting the assignment (yielding `None` instead) is a silent revert + /// that every pure `rest_retry_delay` test stays green under. The test + /// above cannot see it either — it never produces a network error. + /// + /// Shape: the first attempt is refused with a 51s hint, then the listener + /// is dropped so every later attempt is refused at connect. Sleeps are + /// therefore hint, hint, hint (~153s) when the hint is carried, and hint, + /// 1s rung, 2s rung (~54s) when it is dropped. The assertion sits in the + /// gap, so it discriminates on the sleep the ladder actually took rather + /// than on any single call's arguments. + /// + /// A network error says nothing about the quota window, so retrying on a + /// sub-2s rung would land inside a window the relay already closed and + /// earn a denial that still costs a counter increment. + #[tokio::test(start_paused = true)] + async fn rest_retry_keeps_the_hint_across_a_network_error() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind test HTTP server"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + // Serve exactly one 429 carrying the hint, then drop the listener + // so every subsequent attempt fails at connect. + if let Ok((mut socket, _)) = listener.accept().await { + let body = r#"{"error":"rate-limited: quota exceeded (api); retry in 51s"}"#; + let mut buf = vec![0u8; 8192]; + let _ = socket.read(&mut buf).await; + let response = format!( + "HTTP/1.1 429 Too Many Requests\r\nContent-Type: application/json\r\n\ + Content-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let _ = socket.write_all(response.as_bytes()).await; + let _ = socket.flush().await; + } + drop(listener); + }); + + let client = RestClient { + http: reqwest::Client::new(), + base_url, + keys: Keys::generate(), + auth_tag_json: None, + }; + + let start = tokio::time::Instant::now(); + let resp = client + .submit_event(&make_test_event(&Keys::generate(), 1_000)) + .await; + let elapsed = start.elapsed(); + + assert!( + resp.is_err(), + "every attempt after the 429 is refused at connect, so the call fails: {resp:?}" + ); + assert!( + elapsed >= Duration::from_secs(150), + "the ladder slept {elapsed:?} across three retries after a 51s hint. \ + Around 54s means the hint was dropped by the network-error arm and \ + the last two sleeps fell back to the 1s and 2s rungs, waking inside \ + a window the relay had already closed" + ); + assert!( + elapsed < Duration::from_secs(200), + "slept {elapsed:?}: three hint-length sleeps, not a compounding one" + ); + server.abort(); + } + + /// Corroboration: the armed gate honours the computed delay. + /// + /// Integration-level, so it can only sample the entropy the call site + /// actually uses. It states the safety property end to end; the + /// deterministic receipts live in the pure-function tests above. + #[tokio::test(start_paused = true)] + async fn rate_limit_gate_arms_for_the_computed_delay() { + for hint in [0u64, 1, 2, 5, 6, 51] { + let base = Duration::from_secs(effective_gate_secs(hint)); + for nanos in [0u32, 500_000_000, 999_999_999] { + let mut state = BgState::new(); + let before = tokio::time::Instant::now(); + let gate = state.set_rate_limit_gate_with(hint, nanos); + assert_eq!( + gate - before, + gate_delay(hint, nanos), + "armed gate for a {hint}s hint (nanos={nanos}) must equal gate_delay" + ); + assert!( + gate - before >= base, + "armed gate for a {hint}s hint (nanos={nanos}) fell below the hint" + ); + } + } + } + /// set_rate_limit_gate takes the max of overlapping deadlines. #[tokio::test(start_paused = true)] async fn rate_limit_gate_extends_to_max() { @@ -6100,7 +6588,7 @@ mod tests { "gate must be active while membership sub is pending" ); - // Advance past the gate (max jitter: 1.2 × 5s = 6s). + // Advance past the gate (max gate length: 1.2 × 5s = 6s). tokio::time::advance(Duration::from_secs(7)).await; assert!( diff --git a/crates/buzz-acp/src/setup_mode.rs b/crates/buzz-acp/src/setup_mode.rs index b1a9372ea4..1abdc4467d 100644 --- a/crates/buzz-acp/src/setup_mode.rs +++ b/crates/buzz-acp/src/setup_mode.rs @@ -74,7 +74,7 @@ use crate::{ author_allowed, config::Config, event_mentions_agent, filter, - relay::{HarnessRelay, RelayEventPublisher}, + relay::{HarnessRelay, RestClient}, }; // ── Payload ─────────────────────────────────────────────────────────────────── @@ -380,7 +380,6 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> } } - let publisher = relay.event_publisher(); let rest_client = relay.rest_client(); let channel_info = crate::pool::ChannelInfoResolver::new(channel_info_map, rest_client.clone()); @@ -450,19 +449,27 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> .await .is_some(); - // Pure gate: author gate verdict + event-id dedup. + // Pure gate: author gate verdict + event-id dedup check. if !should_nudge_for_event( buzz_event.event.id, allowed, filter_matched, - &mut nudged_event_ids, + &nudged_event_ids, ) { continue; } - // Build and publish the setup nudge. - if let Err(e) = publish_setup_nudge( - &publisher, + // Build and publish the setup nudge over the HTTP bridge. + // + // The nudge is a durable kind:9 message. It must not ride the WS + // publish path: that path silently discards non-observer publishes + // while the rate-limit gate is armed and while disconnected, reporting + // success to the caller either way. For an ephemeral typing indicator + // that is correct; for a user-visible reply it is permanent data loss + // logged as "nudge published". `submit_event` surfaces the relay's + // actual response instead. + match publish_setup_nudge( + &rest_client, &config.keys, buzz_event.channel_id, &buzz_event.event, @@ -470,13 +477,21 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> ) .await { - tracing::warn!("setup-mode: failed to publish nudge: {e}"); - } else { - tracing::info!( - channel_id = %buzz_event.channel_id, - event_id = %buzz_event.event.id, - "setup-mode: nudge published" - ); + Err(e) => { + // Dedup is NOT recorded: the nudge was not delivered, so a + // later retry for this event must still be allowed. Recording + // before the send is what made a dropped nudge permanent. + tracing::warn!("setup-mode: failed to publish nudge: {e}"); + } + Ok(()) => { + // Record dedup only after the relay accepted the nudge. + nudged_event_ids.insert(buzz_event.event.id); + tracing::info!( + channel_id = %buzz_event.channel_id, + event_id = %buzz_event.event.id, + "setup-mode: nudge published" + ); + } } } @@ -487,8 +502,14 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> /// /// Callers compute the async gates (`author_allowed`, `filter::match_event`) /// up-front, then pass the boolean results here. This helper handles -/// everything that is synchronous and stateful: the author gate verdict -/// and event-id dedup. +/// everything that is synchronous: the author gate verdict and the dedup +/// *check*. +/// +/// Recording the dedup entry is deliberately NOT done here. The caller inserts +/// the event id only after the relay has accepted the nudge, so a nudge that +/// fails to send can still be retried. Inserting at check time — as this +/// function used to — made any dropped nudge permanently un-retryable, because +/// the event was marked as nudged before anything was delivered. /// /// Returns `true` when the event should produce a nudge. #[must_use] @@ -496,7 +517,7 @@ pub(crate) fn should_nudge_for_event( event_id: EventId, author_allowed: bool, filter_matched: bool, - nudged_event_ids: &mut HashSet, + nudged_event_ids: &HashSet, ) -> bool { if !author_allowed { tracing::debug!("setup-mode: event filtered by author gate"); @@ -505,7 +526,7 @@ pub(crate) fn should_nudge_for_event( if !filter_matched { return false; } - if !nudged_event_ids.insert(event_id) { + if nudged_event_ids.contains(&event_id) { tracing::debug!(%event_id, "setup-mode: skipping already-nudged event"); return false; } @@ -592,8 +613,14 @@ async fn handle_setup_membership( /// /// Threading: flat reply to the thread root if one exists; otherwise reply /// to the triggering event itself. P-tags the asker. +/// +/// Published over the HTTP bridge rather than the WS publish path. The nudge +/// is a durable kind:9 message, and the WS path is documented to carry only +/// ephemeral kinds: it drops non-observer publishes while rate-gated or +/// disconnected and still returns `Ok`. Going through `submit_event` means a +/// failure is actually reported to the caller. async fn publish_setup_nudge( - publisher: &RelayEventPublisher, + rest_client: &RestClient, keys: &nostr::Keys, channel_id: Uuid, triggering_event: &nostr::Event, @@ -637,8 +664,8 @@ async fn publish_setup_nudge( .sign_with_keys(keys) .map_err(|e| anyhow::anyhow!("failed to sign setup nudge: {e}"))?; - publisher - .publish_event(signed) + rest_client + .submit_event(&signed) .await .map_err(|e| anyhow::anyhow!("failed to publish setup nudge: {e}"))?; @@ -1000,13 +1027,13 @@ mod tests { #[test] fn test_non_allowlisted_author_returns_no_nudge() { // author_allowed = false → should return false regardless of other args. - let mut dedup: HashSet = HashSet::new(); + let dedup: HashSet = HashSet::new(); let event_id = fake_event_id(0xAA); let result = should_nudge_for_event( event_id, false, // author NOT allowed true, // filter matched — would otherwise nudge - &mut dedup, + &dedup, ); assert!(!result, "non-allowlisted author must not produce a nudge"); @@ -1017,29 +1044,247 @@ mod tests { ); } + /// Dedup suppresses a replay only once the send has been recorded. + /// + /// The gate now *checks* the dedup set without mutating it; the caller + /// records the id after the relay accepts the nudge. So a replay is + /// accepted until that recording happens, and rejected afterwards. #[test] fn test_same_event_id_twice_nudges_exactly_once() { - // The first call with a given event-id should return true; the second - // call with the identical id must return false (replay dedup). let mut dedup: HashSet = HashSet::new(); let event_id = fake_event_id(0xBB); let first = should_nudge_for_event( event_id, true, // allowed true, // matched - &mut dedup, + &dedup, ); assert!(first, "first occurrence must be accepted"); + // The caller records the id only after a successful publish. + dedup.insert(event_id); + // Simulate reconnect replay: same event arrives again. let second = should_nudge_for_event( event_id, true, // allowed true, // matched - &mut dedup, + &dedup, ); assert!( !second, - "replay of the same event-id must be rejected (dedup)" + "replay of the same event-id must be rejected once recorded" + ); + } + + /// A nudge that failed to send stays retryable. + /// + /// This is the F9/F11 regression: dedup used to be recorded by the gate + /// itself, before the publish was attempted. Combined with a WS publish + /// path that silently drops durable events while rate-gated or + /// disconnected, that made a dropped nudge permanent — the user never got + /// a reply, and the retry was suppressed by an entry recorded for a + /// message that was never delivered. The caller must leave the set + /// untouched on failure. + #[test] + fn test_failed_nudge_remains_retryable() { + let mut dedup: HashSet = HashSet::new(); + let event_id = fake_event_id(0xCC); + + assert!( + should_nudge_for_event(event_id, true, true, &dedup), + "first attempt must be accepted" + ); + + // Publish fails: the caller records nothing. + assert!( + dedup.is_empty(), + "the gate must not record dedup on its own — recording is the \ + caller's job, after the relay accepts the nudge" + ); + + assert!( + should_nudge_for_event(event_id, true, true, &dedup), + "a nudge that was never delivered must still be retryable" + ); + + // Now it succeeds and the caller records it. + dedup.insert(event_id); + assert!( + !should_nudge_for_event(event_id, true, true, &dedup), + "once delivered, the replay must be suppressed" + ); + } + + // ── nudge transport tests ───────────────────────────────────────────────── + // + // The dedup tests above are pure: they prove the gate stopped recording, + // but they cannot see which transport the nudge rides. That distinction is + // the other half of the fix, and reverting `publish_setup_nudge` to the WS + // publisher leaves every pure test green. + // + // These drive `publish_setup_nudge` against a real socket so the transport + // is observable: the request must arrive over HTTP, and a relay refusal + // must come back as `Err` rather than being swallowed. + + /// A one-shot HTTP server that records the request line and body, and + /// replies with `status`. Returns the bound base URL and a handle to the + /// captured request. + async fn capturing_bridge( + status: &'static str, + body: &'static str, + ) -> ( + String, + std::sync::Arc>>, + tokio::task::JoinHandle<()>, + ) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind test HTTP server"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let captured = std::sync::Arc::new(tokio::sync::Mutex::new(Vec::new())); + let server_captured = captured.clone(); + let server = tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut buf = vec![0u8; 16384]; + let n = socket.read(&mut buf).await.unwrap_or(0); + server_captured + .lock() + .await + .push(String::from_utf8_lossy(&buf[..n]).to_string()); + let response = format!( + "HTTP/1.1 {status}\r\nContent-Type: application/json\r\n\ + Content-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let _ = socket.write_all(response.as_bytes()).await; + } + }); + (base_url, captured, server) + } + + fn test_rest_client(base_url: String) -> crate::relay::RestClient { + crate::relay::RestClient { + http: reqwest::Client::new(), + base_url, + keys: nostr::Keys::generate(), + auth_tag_json: None, + } + } + + fn trigger_event(keys: &nostr::Keys) -> nostr::Event { + nostr::EventBuilder::new( + nostr::Kind::Custom(KIND_STREAM_MESSAGE as u16), + "how do I set you up?", + ) + .tags([]) + .sign_with_keys(keys) + .expect("sign trigger event") + } + + fn test_payload() -> SetupPayload { + SetupPayload { + agent_name: "Fizz".into(), + agent_pubkey: "aabbccddeeff0011".into(), + requirements: vec![], + } + } + + /// The nudge goes out over the HTTP bridge, as a durable kind:9. + /// + /// This is the transport half of F9/F11. `publish_event` (the WS path) + /// returns `Ok` as soon as the command is queued and never touches a + /// socket in-process, so if this fix were reverted no HTTP request would + /// arrive and this test fails on the request count. + #[tokio::test] + async fn nudge_is_published_over_the_http_bridge() { + let (base_url, captured, server) = capturing_bridge("200 OK", "{}").await; + let keys = nostr::Keys::generate(); + let trigger = trigger_event(&keys); + + publish_setup_nudge( + &test_rest_client(base_url), + &keys, + Uuid::new_v4(), + &trigger, + &test_payload(), + ) + .await + .expect("a 200 from the bridge must be reported as success"); + + let requests = captured.lock().await; + assert_eq!(requests.len(), 1, "the nudge must reach the HTTP bridge"); + let request = &requests[0]; + assert!( + request.starts_with("POST /events "), + "nudge must POST to the events endpoint, got: {}", + request.lines().next().unwrap_or_default() + ); + assert!( + request.contains(&format!("\"kind\":{KIND_STREAM_MESSAGE}")), + "the nudge is a durable kind:{KIND_STREAM_MESSAGE} message" + ); + server.abort(); + } + + /// A relay refusal is surfaced to the caller, not swallowed. + /// + /// This is what makes the dedup-after-`Ok` ordering meaningful: the caller + /// can only decline to record the dedup entry if the failure is actually + /// reported. The WS path returned `Ok` for a dropped publish, which is why + /// a rate-gated nudge was lost while being logged as "nudge published". + #[tokio::test] + async fn a_refused_nudge_is_reported_as_an_error() { + // 403 is non-retriable, so the retry ladder returns immediately. + let (base_url, captured, server) = capturing_bridge("403 Forbidden", "denied").await; + let keys = nostr::Keys::generate(); + let trigger = trigger_event(&keys); + + let result = publish_setup_nudge( + &test_rest_client(base_url), + &keys, + Uuid::new_v4(), + &trigger, + &test_payload(), + ) + .await; + + assert!( + result.is_err(), + "a relay refusal must be an Err — reporting Ok is the data-loss bug" + ); + assert_eq!(captured.lock().await.len(), 1, "the attempt was made"); + server.abort(); + } + + /// With no relay listening at all, the nudge fails rather than reporting + /// success. This is the disconnected case: the WS path accepted the + /// publish into a queue nobody was draining and told the caller `Ok`. + #[tokio::test] + async fn a_nudge_with_no_relay_listening_fails() { + // Bind and immediately drop the listener so the port is closed. + let base_url = { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind to claim a port"); + format!("http://{}", listener.local_addr().unwrap()) + }; + let keys = nostr::Keys::generate(); + let trigger = trigger_event(&keys); + + let result = publish_setup_nudge( + &test_rest_client(base_url), + &keys, + Uuid::new_v4(), + &trigger, + &test_payload(), + ) + .await; + + assert!( + result.is_err(), + "a nudge that never reached a relay must not be reported as published" ); } diff --git a/crates/buzz-cli/src/client.rs b/crates/buzz-cli/src/client.rs index ee8868ad92..c32e04b0f9 100644 --- a/crates/buzz-cli/src/client.rs +++ b/crates/buzz-cli/src/client.rs @@ -125,9 +125,29 @@ const RETRY_MAX_ATTEMPTS: u32 = 3; /// `RETRY_BASE_SECS[i]` is the ceiling for attempt `i` before attempt `i+1`. const RETRY_BASE_SECS: [f64; 2] = [0.5, 1.5]; -/// Maximum seconds to honour a relay-provided `retry in Ns` hint from a 429. -/// Defensive cap against pathological hints; real relay hints observed up to ~24 s. -const RETRY_IN_MAX_SECS: u64 = 30; +/// Bounds for honouring a relay-provided `retry in Ns` hint from a 429. +/// +/// The floor exists because the hint is a *window TTL*, not a suggestion: +/// retrying before it expires is guaranteed to be denied again, and each denied +/// attempt still costs a counter increment at the relay. A `retry in 0s` hint +/// (or any sub-second value) would otherwise sleep zero and retry instantly, +/// turning the client into the storm the hint is trying to prevent. +/// +/// The ceiling is a defensive cap against a pathological hint. It must exceed +/// the relay's longest quota window, or the client wakes inside a window it was +/// told to sit out: hints of 51s have been observed in production against a 60s +/// window, so a 30s cap guaranteed a wasted attempt. Sleeps happen between +/// requests, so this is independent of the per-request `BUZZ_TIMEOUT_SECS`. +const RETRY_IN_MIN_SECS: u64 = 1; +const RETRY_IN_MAX_SECS: u64 = 90; + +/// Clamp a relay `retry in Ns` hint into the honoured range. +/// +/// Pure and total, so both call sites share one policy and the endpoints are +/// directly testable. +fn clamp_retry_hint_secs(secs: u64) -> Duration { + Duration::from_secs(secs.clamp(RETRY_IN_MIN_SECS, RETRY_IN_MAX_SECS)) +} /// Returns a full-jitter delay for attempt `i`: a random duration in `[0, RETRY_BASE_SECS[i])`. fn jitter_delay(attempt: u32) -> Duration { @@ -631,7 +651,8 @@ impl BuzzClient { /// failures and mid-body TCP drops. /// - `Err(CliError::Relay { status: 429 | 502 | 503 | 504, .. })` — transient relay /// or proxy errors. For 429 the `retry in Ns` hint from the body is used as the - /// delay (capped at `RETRY_IN_MAX_SECS`); all others use exponential jitter. + /// delay (clamped to `[RETRY_IN_MIN_SECS, RETRY_IN_MAX_SECS]`); all others use + /// exponential jitter. /// /// Use this variant for all operations (reads, writes, uploads); the retry boundary /// covers the entire operation including response body transfer. @@ -659,7 +680,7 @@ impl BuzzClient { } CliError::Relay { status: 429, body } => { let d = parse_retry_hint_text(body) - .map(|s| Duration::from_secs(s.min(RETRY_IN_MAX_SECS))) + .map(clamp_retry_hint_secs) .unwrap_or_else(|| jitter_delay(attempt)); Some(d) } @@ -943,7 +964,7 @@ impl BuzzClient { // timeout/body-loss after the relay may have acted). if !is_last { let delay = parse_retry_hint_text(msg) - .map(|s| Duration::from_secs(s.min(RETRY_IN_MAX_SECS))) + .map(clamp_retry_hint_secs) .unwrap_or_else(|| jitter_delay(attempt)); tokio::time::sleep(delay).await; continue; @@ -1442,8 +1463,9 @@ mod retry_tests { use std::time::Duration; use super::{ - env_duration_secs, is_moderation_kind, jitter_delay, parse_retry_hint_text, - parse_retry_in_secs, RETRY_BASE_SECS, RETRY_IN_MAX_SECS, RETRY_MAX_ATTEMPTS, + clamp_retry_hint_secs, env_duration_secs, is_moderation_kind, jitter_delay, + parse_retry_hint_text, parse_retry_in_secs, RETRY_BASE_SECS, RETRY_IN_MAX_SECS, + RETRY_IN_MIN_SECS, RETRY_MAX_ATTEMPTS, }; // ---- parse_retry_in_secs ---- @@ -1460,12 +1482,63 @@ mod retry_tests { assert_eq!(parse_retry_in_secs(body), Some(3)); } + /// A `retry in 0s` hint parses as `Some(0)`: the parser reports what the + /// relay said, faithfully. Turning 0 into a zero-length sleep is the + /// clamp's job to prevent, not the parser's — see + /// `clamp_retry_hint_secs_floors_zero_and_caps_pathological`. #[test] fn parse_retry_in_zero_seconds() { let body = r#"{"error":"retry in 0s"}"#; assert_eq!(parse_retry_in_secs(body), Some(0)); } + // ---- clamp_retry_hint_secs ---- + + /// The clamp is what stands between a relay hint and a retry storm. + /// + /// Endpoints are asserted exactly, in both directions: + /// + /// * A `retry in 0s` hint previously produced `Duration::from_secs(0)` — an + /// instant retry into a window the relay had just closed, which is the + /// storm the hint exists to prevent. It must now floor to a real sleep. + /// * A hint above the cap must be capped, and the cap must sit above the + /// relay's longest quota window. The old 30s cap was *below* the 51s + /// hints observed in production against a 60s window, so it guaranteed a + /// wasted attempt. Asserting `>= 60` pins the property (outlasts the + /// window) rather than the number, so retuning the cap does not + /// silently reintroduce the defect. + #[test] + fn clamp_retry_hint_secs_floors_zero_and_caps_pathological() { + assert_eq!( + clamp_retry_hint_secs(0), + Duration::from_secs(RETRY_IN_MIN_SECS), + "a `retry in 0s` hint must never sleep zero and retry instantly" + ); + assert_eq!( + clamp_retry_hint_secs(u64::MAX), + Duration::from_secs(RETRY_IN_MAX_SECS), + "a pathological hint must be capped" + ); + + // Values inside the range pass through untouched, including the 51s + // hint that the previous 30s cap silently truncated. + for secs in [RETRY_IN_MIN_SECS, 2, 24, 51, RETRY_IN_MAX_SECS] { + assert_eq!( + clamp_retry_hint_secs(secs), + Duration::from_secs(secs), + "a {secs}s hint is within range and must be honoured exactly" + ); + } + + const { assert!(RETRY_IN_MIN_SECS > 0) }; + const { + assert!( + RETRY_IN_MAX_SECS >= 60, + "the cap must outlast the relay's longest quota window" + ) + }; + } + #[test] fn parse_garbled_body_returns_none() { assert_eq!(parse_retry_in_secs("not json at all"), None); diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index a118ff453f..bffe5b551a 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -39,10 +39,10 @@ async fn enforce_http_admission( { Ok(()) => Ok(()), Err(crate::admission::AdmissionError::Exceeded { reset_in_secs }) => { - metrics::counter!("buzz_admission_rejections_total", "transport" => "http", "reason" => "quota").increment(1); + metrics::counter!("buzz_admission_rejections_total", "transport" => "http", "reason" => "quota", "limit_type" => LimitType::ApiCalls.key_suffix()).increment(1); Err(api_error( StatusCode::TOO_MANY_REQUESTS, - &format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"), + &crate::connection::quota_rejection_reason(&LimitType::ApiCalls, reset_in_secs), )) } Err(crate::admission::AdmissionError::Unavailable) => { diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 96e266779f..a1457e1a69 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -15,6 +15,7 @@ use tracing::{debug, info, trace, warn}; use uuid::Uuid; use buzz_auth::{generate_challenge, AuthContext, LimitType}; +use buzz_core::kind::{event_kind_u32, is_ephemeral}; use buzz_core::tenant::TenantContext; use nostr::Filter; @@ -625,11 +626,16 @@ async fn enforce_ws_admission( ClientMessage::Req { sub_id, .. } => Some(sub_id.as_str()), _ => None, }; - if !send_admission_result(conn, ws_result, sub_id) { + if !send_admission_result(conn, ws_result, sub_id, LimitType::WsEvents) { return false; } - if is_event { + // The Messages budget covers durable writes only. Ephemeral kinds + // (20000–29999) are never persisted — buzz-db rejects them with + // `EphemeralEventRejected` — so they are admitted on the WsEvents + // budget above alone, mirroring the `is_ephemeral` gate the EVENT + // handler already applies for scope checks. + if event_bills_messages(msg) { let message_limit = if is_agent { limits.agent_standard_messages_per_min } else { @@ -644,7 +650,7 @@ async fn enforce_ws_admission( message_limit, ) .await; - if !send_admission_result(conn, message_result, None) { + if !send_admission_result(conn, message_result, None, LimitType::Messages) { return false; } } @@ -652,18 +658,43 @@ async fn enforce_ws_admission( true } +/// Whether a client frame bills the `Messages` (durable-write) budget. +/// +/// Only EVENT frames carrying a non-ephemeral kind do. Ephemeral kinds +/// (20000–29999) are never persisted, so they ride on `WsEvents` alone. +fn event_bills_messages(msg: &ClientMessage) -> bool { + match msg { + ClientMessage::Event(event) => !is_ephemeral(event_kind_u32(event)), + _ => false, + } +} + +/// Rejection reason for a quota denial, naming which budget was exhausted. +/// +/// Keeps the `rate-limited:` prefix and the `retry in {N}s` hint that all +/// clients parse by substring; the `({suffix})` discriminator sits between +/// them so log readers and the rejection metric can tell WsEvents, Messages, +/// and ApiCalls denials apart. +pub(crate) fn quota_rejection_reason(limit_type: &LimitType, reset_in_secs: u64) -> String { + format!( + "rate-limited: quota exceeded ({}); retry in {reset_in_secs}s", + limit_type.key_suffix() + ) +} + fn send_admission_result( conn: &ConnectionState, result: Result<(), crate::admission::AdmissionError>, sub_id: Option<&str>, + limit_type: LimitType, ) -> bool { match result { Ok(()) => true, Err(crate::admission::AdmissionError::Exceeded { reset_in_secs }) => { - metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "quota").increment(1); + metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "quota", "limit_type" => limit_type.key_suffix()).increment(1); conn.send(request_rejection_message( sub_id, - &format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"), + "a_rejection_reason(&limit_type, reset_in_secs), )); false } @@ -785,6 +816,66 @@ mod tests { assert_eq!(notice, serde_json::json!(["NOTICE", reason])); } + fn signed_event(kind: u32) -> nostr::Event { + nostr::EventBuilder::new(nostr::Kind::Custom(kind as u16), "test") + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign test event") + } + + #[test] + fn durable_events_bill_the_messages_budget() { + // kind:1 note and kind:9 channel message are durable. + assert!(event_bills_messages(&ClientMessage::Event(signed_event(1)))); + assert!(event_bills_messages(&ClientMessage::Event(signed_event(9)))); + } + + #[test] + fn ephemeral_events_do_not_bill_the_messages_budget() { + // Presence (20001), typing (20002), and observer frames (24200) are + // range-ephemeral: never persisted, so never billed against Messages. + for kind in [20000, 20001, 20002, 24200, 29999] { + assert!( + !event_bills_messages(&ClientMessage::Event(signed_event(kind))), + "kind {kind} is ephemeral and must not bill Messages" + ); + } + // Boundary: first kind past the ephemeral range is durable again. + assert!(event_bills_messages(&ClientMessage::Event(signed_event( + 30000 + )))); + } + + #[test] + fn non_event_frames_do_not_bill_the_messages_budget() { + assert!(!event_bills_messages(&ClientMessage::Req { + sub_id: "sub".to_owned(), + filters: vec![Filter::new()], + })); + assert!(!event_bills_messages(&ClientMessage::Count { + sub_id: "sub".to_owned(), + filters: vec![Filter::new()], + })); + assert!(!event_bills_messages(&ClientMessage::Close( + "sub".to_owned() + ))); + } + + #[test] + fn quota_rejection_text_names_the_limit_type_and_keeps_the_retry_hint() { + // Clients parse the `retry in {N}s` hint by substring; the limit_type + // discriminator must not disturb that or the `rate-limited:` prefix. + for (limit_type, tag) in [ + (LimitType::Messages, "msg"), + (LimitType::WsEvents, "ws"), + (LimitType::ApiCalls, "api"), + ] { + let text = quota_rejection_reason(&limit_type, 7); + assert!(text.starts_with("rate-limited:")); + assert!(text.contains(&format!("({tag})"))); + assert!(text.contains("retry in 7s")); + } + } + #[tokio::test] async fn send_loop_batches_queued_data_frames_into_one_flush() { let (data_tx, data_rx) = mpsc::channel(MAX_WS_SEND_BATCH);