diff --git a/Cargo.lock b/Cargo.lock index 36b2020858..0b9e8577ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -883,6 +883,7 @@ dependencies = [ name = "buzz-cli" version = "0.1.0" dependencies = [ + "axum", "base64", "buzz-core", "buzz-persona", @@ -896,6 +897,7 @@ dependencies = [ "hex", "infer", "nostr", + "rand 0.10.1", "reqwest 0.13.4", "serde", "serde_json", diff --git a/crates/buzz-cli/Cargo.toml b/crates/buzz-cli/Cargo.toml index 530a1217b1..a12260b526 100644 --- a/crates/buzz-cli/Cargo.toml +++ b/crates/buzz-cli/Cargo.toml @@ -76,6 +76,11 @@ dirs = "6" # WebSocket client — ephemeral event publish (kind:20001 is WS-only on the relay) buzz-ws-client = { path = "../buzz-ws-client" } +# Random number generation — full jitter for exponential backoff in with_retry +rand = { workspace = true } + [dev-dependencies] # Scratch files for channel-templates.json fixtures in tests tempfile = "3" +# Minimal HTTP test server for retry/policy integration tests +axum = { workspace = true } diff --git a/crates/buzz-cli/src/client.rs b/crates/buzz-cli/src/client.rs index 85a8563056..d0dd2677a9 100644 --- a/crates/buzz-cli/src/client.rs +++ b/crates/buzz-cli/src/client.rs @@ -1,3 +1,4 @@ +use std::future::Future; use std::time::Duration; use base64::engine::general_purpose::STANDARD as B64; @@ -117,6 +118,85 @@ fn relay_server_tag(relay_url: &str) -> Option { } } +/// Maximum number of attempts per request (initial attempt + two retries). +const RETRY_MAX_ATTEMPTS: u32 = 3; + +/// Base sleep durations for full-jitter exponential backoff. +/// `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; + +/// Returns a full-jitter delay for attempt `i`: a random duration in `[0, RETRY_BASE_SECS[i])`. +fn jitter_delay(attempt: u32) -> Duration { + Duration::from_secs_f64(RETRY_BASE_SECS[attempt as usize] * rand::random::()) +} + +/// Read an env var as a `u64` of seconds and return the corresponding `Duration`. +/// Falls back to `default` if the var is unset, unparseable, or zero (zero is treated +/// as invalid to prevent accidentally disabling all timeouts). +fn env_duration_secs(name: &str, default: u64) -> Duration { + std::env::var(name) + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|&n| n > 0) + .map(Duration::from_secs) + .unwrap_or_else(|| Duration::from_secs(default)) +} + +/// Scan a plain-text string for a `retry in s` pattern and return `N`. +/// +/// Matches the literal substring `retry in ` followed by one or more ASCII digits +/// and the character `s`. Works on both extracted field values (`rate-limited: +/// quota exceeded; retry in 4s`) and substrings of raw relay JSON bodies. +/// Returns `None` when the pattern is absent or the digit sequence is empty. +fn parse_retry_hint_text(text: &str) -> Option { + const PREFIX: &str = "retry in "; + let after = text.find(PREFIX).map(|i| &text[i + PREFIX.len()..])?; + let end = after + .find(|c: char| !c.is_ascii_digit()) + .unwrap_or(after.len()); + if end == 0 || after.as_bytes().get(end) != Some(&b's') { + return None; + } + after[..end].parse::().ok() +} + +/// Parse a `retry in Ns` hint from a relay 429 JSON body. +/// +/// Extracts the `error` or `message` field and delegates to +/// `parse_retry_hint_text`. Returns `None` when the body is not valid JSON or +/// the extracted field does not contain the pattern. +#[cfg(test)] +fn parse_retry_in_secs(body: &str) -> Option { + let text = serde_json::from_str::(body) + .ok() + .and_then(|v| { + v.get("error") + .or_else(|| v.get("message")) + .and_then(|m| m.as_str().map(str::to_string)) + })?; + parse_retry_hint_text(&text) +} + +/// Extract the `error` or `message` field from a relay JSON error body. +/// +/// Production relay error bodies are shaped as `{"error":"..."}` (via `api_error()`). +/// Returns the extracted field value, or `None` if the body is not valid JSON or +/// neither field is present. The raw body should be retained for diagnostics when +/// `None` is returned. +fn extract_relay_message_field(body: &str) -> Option { + serde_json::from_str::(body) + .ok() + .and_then(|v| { + v.get("error") + .or_else(|| v.get("message")) + .and_then(|m| m.as_str().map(str::to_string)) + }) +} + fn should_retry_legacy_upload(status: reqwest::StatusCode) -> bool { matches!( status, @@ -124,6 +204,49 @@ fn should_retry_legacy_upload(status: reqwest::StatusCode) -> bool { ) } +/// Returns `true` for moderation command kinds (9040–9044). +/// +/// These events execute immediately at the relay without dedup, so they must +/// not be blindly retried on ambiguous outcomes. +fn is_moderation_kind(kind: u16) -> bool { + matches!(kind, 9040..=9044) +} + +/// Returns `true` for HTTP status codes that indicate a successful response +/// (equivalent to `reqwest::StatusCode::is_success()` for u16). +fn resp_was_success(status: u16) -> bool { + (200..300).contains(&status) +} + +/// Returns `true` if a stored-event exhaustion error is ambiguous (the relay +/// may have executed the command) and should be converted to `DeliveryUnknown`. +/// +/// Connect failures are definitively pre-relay (never executed) so they remain +/// retryable. Canonical pre-ingest 429 (`Relay{status:429}`) was provably +/// rejected before storage — also retryable. Everything else (timeout, body +/// loss, decode error, proxy 502-504) may have crossed the relay's storage +/// boundary and must not invite an outer re-sign. +fn is_stored_event_exhaustion_ambiguous(e: &CliError) -> bool { + match e { + CliError::Network(net_err) => { + // Connect is definitively pre-relay. + if net_err.is_connect() { + return false; + } + // Timeout, body, decode, request — ambiguous. + net_err.is_timeout() || net_err.is_body() || net_err.is_decode() || net_err.is_request() + } + // Canonical pre-ingest 429 — relay did not store. + CliError::Relay { status: 429, .. } => false, + // Proxy 502-504 — relay may have accepted before the proxy failed. + CliError::Relay { + status: 502..=504, .. + } => true, + // All other variants are not retried by with_retry_body; not ambiguous. + _ => false, + } +} + fn is_safe_media_path_segment(sha256_ext: &str) -> bool { let segments: Vec<&str> = sha256_ext.split('.').collect(); match segments.as_slice() { @@ -224,6 +347,43 @@ fn sign_blossom_get(keys: &Keys, media_url: &str) -> Result { )) } +fn sign_blossom_upload( + keys: &Keys, + sha256: &str, + mime: &str, + relay_url: &str, +) -> Result { + use base64::engine::general_purpose::URL_SAFE_NO_PAD; + use nostr::Timestamp; + + let now = Timestamp::now().as_secs(); + let expiry: u64 = if mime.starts_with("video/") { + 3600 + } else { + 600 + }; + let exp_str = (now + expiry).to_string(); + + let mut tags = vec![ + Tag::parse(["t", "upload"]).map_err(|e| CliError::Other(e.to_string()))?, + Tag::parse(["x", sha256]).map_err(|e| CliError::Other(e.to_string()))?, + Tag::parse(["expiration", &exp_str]).map_err(|e| CliError::Other(e.to_string()))?, + ]; + if let Some(domain) = relay_server_tag(relay_url) { + tags.push(Tag::parse(["server", &domain]).map_err(|e| CliError::Other(e.to_string()))?); + } + + let auth_event = EventBuilder::new(Kind::from(24242), "Upload file") + .tags(tags) + .sign_with_keys(keys) + .map_err(|e| CliError::Other(format!("signing failed: {e}")))?; + + Ok(format!( + "Nostr {}", + URL_SAFE_NO_PAD.encode(auth_event.as_json().as_bytes()) + )) +} + #[cfg(test)] mod media_download_tests { use super::*; @@ -369,6 +529,15 @@ pub struct BuzzClient { } impl BuzzClient { + /// Create a new client pointing at `relay_url`. + /// + /// Timeout defaults are tuned for degraded WAN links and can be overridden + /// via environment variables: + /// + /// - `BUZZ_CONNECT_TIMEOUT_SECS` — TCP connect timeout (default 15 s) + /// - `BUZZ_TIMEOUT_SECS` — per-request total timeout (default 30 s) + /// + /// A value of zero for either variable is treated as invalid and falls back to the default. pub fn new( relay_url: String, keys: Keys, @@ -376,8 +545,8 @@ impl BuzzClient { auth_tag_json: Option, ) -> Result { let http = reqwest::Client::builder() - .timeout(Duration::from_secs(10)) - .connect_timeout(Duration::from_secs(5)) + .timeout(env_duration_secs("BUZZ_TIMEOUT_SECS", 30)) + .connect_timeout(env_duration_secs("BUZZ_CONNECT_TIMEOUT_SECS", 15)) .build() .map_err(|e| CliError::Other(e.to_string()))?; Ok(Self { @@ -451,6 +620,66 @@ impl BuzzClient { } } + /// Execute `op` up to `RETRY_MAX_ATTEMPTS` times, including body-transfer failures + /// and transient relay error statuses. + /// + /// The closure is expected to consume the response body and return the parsed result + /// as `T`. Retries on non-last attempts when `op` returns: + /// + /// - `Err(CliError::Network(e))` where `e.is_connect() || e.is_timeout() || + /// e.is_request() || e.is_body() || e.is_decode()` — covers both connection + /// 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. + /// + /// Use this variant for all operations (reads, writes, uploads); the retry boundary + /// covers the entire operation including response body transfer. + async fn with_retry_body<'a, T, F, Fut>(&'a self, op: F) -> Result + where + F: Fn() -> Fut, + Fut: Future> + 'a, + T: 'a, + { + for attempt in 0..RETRY_MAX_ATTEMPTS { + let is_last = attempt == RETRY_MAX_ATTEMPTS - 1; + match op().await { + Ok(value) => return Ok(value), + Err(e) => { + if !is_last { + let delay = match &e { + CliError::Network(net_err) + if net_err.is_connect() + || net_err.is_timeout() + || net_err.is_request() + || net_err.is_body() + || net_err.is_decode() => + { + Some(jitter_delay(attempt)) + } + CliError::Relay { status: 429, body } => { + let d = parse_retry_hint_text(body) + .map(|s| Duration::from_secs(s.min(RETRY_IN_MAX_SECS))) + .unwrap_or_else(|| jitter_delay(attempt)); + Some(d) + } + CliError::Relay { + status: 502..=504, .. + } => Some(jitter_delay(attempt)), + _ => None, + }; + if let Some(d) = delay { + tokio::time::sleep(d).await; + continue; + } + } + return Err(e); + } + } + } + unreachable!("loop exhausts all RETRY_MAX_ATTEMPTS") + } + async fn query_pages( &self, mut filter: serde_json::Value, @@ -543,17 +772,29 @@ impl BuzzClient { /// Each filter is ORed by the relay (standard Nostr REQ behavior). pub async fn query_multi(&self, filters: &[serde_json::Value]) -> Result { let url = format!("{}/query", self.relay_url); - let body_bytes = serde_json::to_vec(filters) - .map_err(|e| CliError::Other(format!("filter serialization failed: {e}")))?; - let auth = sign_nip98(&self.keys, "POST", &url, Some(&body_bytes))?; - let req = self - .http - .post(&url) - .header("Authorization", &auth) - .header("Content-Type", "application/json") - .body(body_bytes); - let resp = self.with_auth_tag(req).send().await?; - self.handle_response(resp).await + let body = bytes::Bytes::from( + serde_json::to_vec(filters) + .map_err(|e| CliError::Other(format!("filter serialization failed: {e}")))?, + ); + self.with_retry_body(|| { + let body = body.clone(); + let url = url.clone(); + async move { + let auth = sign_nip98(&self.keys, "POST", &url, Some(&body))?; + let resp = self + .with_auth_tag( + self.http + .post(&url) + .header("Authorization", auth) + .header("Content-Type", "application/json") + .body(body), + ) + .send() + .await?; + self.handle_response(resp).await + } + }) + .await } /// Execute a one-shot count via the HTTP bridge. @@ -561,19 +802,29 @@ impl BuzzClient { #[allow(dead_code)] pub async fn count(&self, filter: &serde_json::Value) -> Result { let url = format!("{}/count", self.relay_url); - let body_bytes = serde_json::to_vec(&[filter]) - .map_err(|e| CliError::Other(format!("filter serialization failed: {e}")))?; - let auth = sign_nip98(&self.keys, "POST", &url, Some(&body_bytes))?; - - let req = self - .http - .post(&url) - .header("Authorization", &auth) - .header("Content-Type", "application/json") - .body(body_bytes); - let resp = self.with_auth_tag(req).send().await?; - - self.handle_response(resp).await + let body = bytes::Bytes::from( + serde_json::to_vec(&[filter]) + .map_err(|e| CliError::Other(format!("filter serialization failed: {e}")))?, + ); + self.with_retry_body(|| { + let body = body.clone(); + let url = url.clone(); + async move { + let auth = sign_nip98(&self.keys, "POST", &url, Some(&body))?; + let resp = self + .with_auth_tag( + self.http + .post(&url) + .header("Authorization", auth) + .header("Content-Type", "application/json") + .body(body), + ) + .send() + .await?; + self.handle_response(resp).await + } + }) + .await } /// GET an authed relay endpoint (NIP-98), returning the raw JSON body. @@ -584,28 +835,234 @@ impl BuzzClient { /// stored events. pub async fn get_authed(&self, path: &str) -> Result { let url = format!("{}{path}", self.relay_url); - let auth = sign_nip98(&self.keys, "GET", &url, None)?; - let req = self.http.get(&url).header("Authorization", &auth); - let resp = self.with_auth_tag(req).send().await?; - self.handle_response(resp).await + self.with_retry_body(|| { + let url = url.clone(); + async move { + let auth = sign_nip98(&self.keys, "GET", &url, None)?; + let resp = self + .with_auth_tag(self.http.get(&url).header("Authorization", auth)) + .send() + .await?; + self.handle_response(resp).await + } + }) + .await } /// Submit a signed Nostr event via POST /events. + /// + /// For non-idempotent moderation command kinds (9040–9044), an ambiguous + /// outcome (mid-request error, body loss, non-ingest 429, or 502/503/504) + /// surfaces as `CliError::DeliveryUnknown` instead of being retried. These + /// events execute at the relay *before* any dedup check, so a blind re-send + /// can duplicate the mutation. Only confirmed-unreceived failures (TCP + /// connect error or a pre-ingest 429 carrying a `rate-limited:` body) are + /// safe to retry. + /// + /// All other event kinds retain the standard retry policy. pub async fn submit_event(&self, event: nostr::Event) -> Result { + let kind = event.kind.as_u16(); + if is_moderation_kind(kind) { + self.submit_moderation_event(event).await + } else { + self.submit_stored_event(event).await + } + } + + /// Submit a moderation command (kinds 9040–9044) with non-idempotent retry policy. + async fn submit_moderation_event(&self, event: nostr::Event) -> Result { let url = format!("{}/events", self.relay_url); - let body_bytes = serde_json::to_vec(&event) - .map_err(|e| CliError::Other(format!("event serialization failed: {e}")))?; - let auth = sign_nip98(&self.keys, "POST", &url, Some(&body_bytes))?; + let body = bytes::Bytes::from( + serde_json::to_vec(&event) + .map_err(|e| CliError::Other(format!("event serialization failed: {e}")))?, + ); - let req = self - .http - .post(&url) - .header("Authorization", &auth) - .header("Content-Type", "application/json") - .body(body_bytes); - let resp = self.with_auth_tag(req).send().await?; + for attempt in 0..RETRY_MAX_ATTEMPTS { + let is_last = attempt == RETRY_MAX_ATTEMPTS - 1; + + // Re-sign NIP-98 each attempt: the nonce tag generates a fresh + // event ID, keeping retries safe against the relay's replay guard. + let auth = sign_nip98(&self.keys, "POST", &url, Some(&body))?; + let send_result: Result = self + .with_auth_tag( + self.http + .post(&url) + .header("Authorization", auth) + .header("Content-Type", "application/json") + .body(body.clone()), + ) + .send() + .await + .map_err(CliError::from); + + match send_result { + Err(e) => { + if let CliError::Network(ref net_err) = e { + // Only connect-failure is safe to retry: the relay never saw + // the request. Timeout and mid-request errors are ambiguous. + if !is_last && net_err.is_connect() { + tokio::time::sleep(jitter_delay(attempt)).await; + continue; + } + if net_err.is_connect() { + // Final attempt: definitively never reached the relay — retryable. + return Err(e); + } + if net_err.is_timeout() + || net_err.is_request() + || net_err.is_body() + || net_err.is_decode() + { + // Ambiguous: the relay may have executed this command. + return Err(CliError::DeliveryUnknown(format!( + "moderation command (kind {}) outcome unknown: {}", + event.kind.as_u16(), + net_err + ))); + } + } + return Err(e); + } + Ok(resp) => { + let status = resp.status().as_u16(); + if status == 429 { + // Only retry if the relay's own ingest layer rejected it: + // the extracted error/message field must start with + // "rate-limited:". A proxy-level 429 (or JSON-wrapped body + // whose field does not start with "rate-limited:") leaves + // relay execution state ambiguous. + let body_text = resp.text().await.unwrap_or_default(); + let extracted = extract_relay_message_field(&body_text); + let msg = extracted.as_deref().unwrap_or(&body_text); + if msg.starts_with("rate-limited:") { + // Canonical pre-ingest 429: the relay provably did not execute + // the command. Retry while budget remains; on exhaustion return + // Relay(429) (retryable:true) — the caller may retry the exact + // same command. DeliveryUnknown is reserved for outcomes where + // relay execution is genuinely ambiguous (proxy 429, 502-504, + // 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))) + .unwrap_or_else(|| jitter_delay(attempt)); + tokio::time::sleep(delay).await; + continue; + } + // Budget exhausted — still pre-ingest, still safe to retry. + return Err(CliError::Relay { + status: 429, + body: body_text, + }); + } + // Non-canonical 429 (proxy-level or unrecognised body): outcome unknown. + return Err(CliError::DeliveryUnknown(format!( + "moderation command (kind {}) outcome unknown: HTTP 429", + event.kind.as_u16() + ))); + } + if matches!(status, 502..=504) { + // Proxy-level error: the relay may have received and executed + // the command before the proxy failed. + return Err(CliError::DeliveryUnknown(format!( + "moderation command (kind {}) outcome unknown: HTTP {status}", + event.kind.as_u16() + ))); + } + // 2xx or definitive error (4xx other than 429): read body normally. + let body_text = resp.text().await.map_err(|e| { + // Body loss after relay confirmed receipt is ambiguous for + // non-idempotent commands. + CliError::DeliveryUnknown(format!( + "moderation command (kind {}) outcome unknown: response body lost: {e}", + event.kind.as_u16() + )) + })?; + // Map the body through handle_response's error logic inline. + if !resp_was_success(status) { + let message = serde_json::from_str::(&body_text) + .ok() + .and_then(|v| { + v.get("error") + .or_else(|| v.get("message")) + .and_then(|m| m.as_str()) + .map(str::to_string) + }) + .unwrap_or(body_text); + let message = if status == 403 && std::env::var("BUZZ_AUTH_TAG").is_ok() { + format!( + "{message} (BUZZ_AUTH_TAG is set — it may be stale or revoked; try unsetting it)" + ) + } else { + message + }; + return Err(CliError::Relay { + status, + body: message, + }); + } + return Ok(body_text); + } + } + } + unreachable!("loop exhausts all RETRY_MAX_ATTEMPTS") + } - self.handle_response(resp).await + /// Submit a stored event (all non-moderation kinds) with the standard retry policy. + /// + /// The full operation — network send AND response body read — is inside the retry + /// boundary so that a dropped body after a 200 header is retried with the same + /// serialized event bytes (and a fresh per-attempt NIP-98 auth event). + /// + /// **Exhaustion policy:** after all attempts, connect failures and canonical + /// pre-ingest 429 remain retryable (`CliError::Network`/`CliError::Relay{429}`) + /// because the relay provably never executed them. Any other final failure + /// (timeout, request, body loss, decode, proxy 502-504) is ambiguous — the + /// relay may have stored the event — so we surface `DeliveryUnknown` + /// (retryable:false) to prevent an outer re-sign creating a duplicate write. + /// Content-addressed uploads are exempt: same bytes ⇒ same hash, so outer + /// re-run is safe regardless of the failure kind. + async fn submit_stored_event(&self, event: nostr::Event) -> Result { + let url = format!("{}/events", self.relay_url); + let body = bytes::Bytes::from( + serde_json::to_vec(&event) + .map_err(|e| CliError::Other(format!("event serialization failed: {e}")))?, + ); + let result = self + .with_retry_body(|| { + let body = body.clone(); + let url = url.clone(); + async move { + // Re-sign NIP-98 each attempt: the nonce tag generates a fresh + // event ID, keeping retries safe against the relay's replay guard. + let auth = sign_nip98(&self.keys, "POST", &url, Some(&body))?; + let resp = self + .with_auth_tag( + self.http + .post(&url) + .header("Authorization", auth) + .header("Content-Type", "application/json") + .body(body), + ) + .send() + .await?; + self.handle_response(resp).await + } + }) + .await; + + // Translate ambiguous final errors to DeliveryUnknown so an outer agent + // following retryable:true does not re-sign and risk a duplicate write. + // Connect failures stay Network (retryable:true) — definitively never received. + // Canonical pre-ingest 429 (Relay{429}) stays retryable — definitively not stored. + if let Err(ref e) = result { + if is_stored_event_exhaustion_ambiguous(e) { + let kind_u16 = event.kind.as_u16(); + return Err(CliError::DeliveryUnknown(format!( + "stored event (kind {kind_u16}) outcome unknown after all attempts: {e}" + ))); + } + } + result } /// Publish an ephemeral event via WebSocket with NIP-42 authentication. @@ -615,8 +1072,12 @@ impl BuzzClient { /// EVENT send, OK wait, and graceful close. pub async fn publish_ephemeral_event(&self, event: nostr::Event) -> Result { let ws_url = to_ws_url(&self.relay_url); + // Hard cap — inner wait ceilings sum to 70 s; connect time and network RTT are + // additional overhead absorbed by this budget. + // See buzz_ws_client::{AUTH_CHALLENGE_TIMEOUT_SECS, AUTH_OK_TIMEOUT_SECS, + // PUBLISH_OK_TIMEOUT_SECS} for the inner ceilings. let ok = - buzz_ws_client::publish_event(&ws_url, event, &self.keys, self.auth_tag.as_ref(), 10) + buzz_ws_client::publish_event(&ws_url, event, &self.keys, self.auth_tag.as_ref(), 75) .await .map_err(|e| CliError::Other(e.to_string()))?; @@ -673,40 +1134,8 @@ impl BuzzClient { // 4. SHA-256 let sha256 = hex::encode(Sha256::digest(&bytes)); - // 5. Sign Blossom auth event (kind:24242) - use nostr::Timestamp; - let now = Timestamp::now().as_secs(); - let expiry = if mime.starts_with("video/") { - 3600 - } else { - 600 - }; - let exp_str = (now + expiry).to_string(); - - let mut blossom_tags = vec![ - Tag::parse(["t", "upload"]).map_err(|e| CliError::Other(e.to_string()))?, - Tag::parse(["x", &sha256]).map_err(|e| CliError::Other(e.to_string()))?, - Tag::parse(["expiration", &exp_str]).map_err(|e| CliError::Other(e.to_string()))?, - ]; - // Extract server domain from relay URL for BUD-11 server tag - if let Some(domain) = relay_server_tag(&self.relay_url) { - blossom_tags - .push(Tag::parse(["server", &domain]).map_err(|e| CliError::Other(e.to_string()))?); - } - - let auth_event = EventBuilder::new(Kind::from(24242), "Upload file") - .tags(blossom_tags) - .sign_with_keys(&self.keys) - .map_err(|e| CliError::Other(format!("signing failed: {e}")))?; - - // 6. Base64url encode the auth event for the header - use base64::engine::general_purpose::URL_SAFE_NO_PAD; - let auth_header = format!( - "Nostr {}", - URL_SAFE_NO_PAD.encode(auth_event.as_json().as_bytes()) - ); - - // 7. PUT request to the BUD-02 /upload endpoint with a generous timeout. + // 5. PUT request to the BUD-02 /upload endpoint with a generous timeout. + // Auth is signed per attempt — matches the per-attempt signing pattern in download_media. let upload_timeout = if mime.starts_with("video/") { Duration::from_secs(600) } else { @@ -714,65 +1143,116 @@ impl BuzzClient { }; let url = format!("{}/upload", self.relay_url); let upload_body = bytes::Bytes::from(bytes); - let req = self - .http - .put(&url) - .timeout(upload_timeout) - .header("Authorization", &auth_header) - .header("Content-Type", &mime) - .header("X-SHA-256", &sha256); - - let mut resp = self - .with_auth_tag(req) - .body(upload_body.clone()) - .send() - .await?; - if should_retry_legacy_upload(resp.status()) { - let legacy_url = format!("{}/media/upload", self.relay_url); - let legacy_req = self - .http - .put(&legacy_url) - .timeout(upload_timeout) - .header("Authorization", &auth_header) - .header("Content-Type", &mime) - .header("X-SHA-256", &sha256); - resp = self - .with_auth_tag(legacy_req) - .body(upload_body) - .send() - .await?; - } - if !resp.status().is_success() { - let status = resp.status().as_u16(); - let body = resp.text().await.unwrap_or_default(); - return Err(CliError::Relay { status, body }); + + // The full upload operation — network send AND response body read — lives inside + // with_retry_body so that a dropped body after 200 headers is retried with the + // same file bytes and a fresh Blossom auth per attempt. + let result: Result = self + .with_retry_body(|| { + let upload_body = upload_body.clone(); + let url = url.clone(); + let mime = mime.clone(); + let sha256 = sha256.clone(); + async move { + let auth_header = + sign_blossom_upload(&self.keys, &sha256, &mime, &self.relay_url)?; + let resp = self + .with_auth_tag( + self.http + .put(&url) + .timeout(upload_timeout) + .header("Authorization", auth_header) + .header("Content-Type", &mime) + .header("X-SHA-256", &sha256) + .body(upload_body), + ) + .send() + .await?; + let status = resp.status(); + if !status.is_success() { + let s = status.as_u16(); + let body = resp.text().await.unwrap_or_default(); + return Err(CliError::Relay { status: s, body }); + } + resp.json::().await.map_err(CliError::from) + } + }) + .await; + + // If the primary /upload endpoint definitively doesn't exist on this relay version + // (404 or 405), fall back to the legacy /media/upload endpoint. The 404/405 switch + // itself is not retried; only transient failures on the selected legacy endpoint are. + match result { + Ok(desc) => return Ok(desc), + Err(CliError::Relay { status: s, body: _ }) + if should_retry_legacy_upload( + reqwest::StatusCode::from_u16(s).unwrap_or(reqwest::StatusCode::NOT_FOUND), + ) => + { + // Fall through to legacy endpoint below. + } + Err(e) => return Err(e), } - resp.json::() - .await - .map_err(|e| CliError::Other(format!("invalid upload response: {e}"))) + let legacy_url = format!("{}/media/upload", self.relay_url); + self.with_retry_body(|| { + let upload_body = upload_body.clone(); + let legacy_url = legacy_url.clone(); + let mime = mime.clone(); + let sha256 = sha256.clone(); + async move { + let auth_header = sign_blossom_upload(&self.keys, &sha256, &mime, &self.relay_url)?; + let resp = self + .with_auth_tag( + self.http + .put(&legacy_url) + .timeout(upload_timeout) + .header("Authorization", auth_header) + .header("Content-Type", &mime) + .header("X-SHA-256", &sha256) + .body(upload_body), + ) + .send() + .await?; + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + return Err(CliError::Relay { status, body }); + } + resp.json::().await.map_err(CliError::from) + } + }) + .await } /// Download a Blossom media blob using BUD-01 `t=get` auth. pub async fn download_media(&self, input: &str) -> Result { let url = media_url_from_input(&self.relay_url, input)?; - let auth_header = sign_blossom_get(&self.keys, &url)?; + // Use a dedicated client: 120 s timeout, no redirect forwarding. let client = reqwest::Client::builder() .timeout(Duration::from_secs(120)) // Do not forward Authorization or x-auth-tag to redirect targets. .redirect(reqwest::redirect::Policy::none()) .build() .map_err(|e| CliError::Other(format!("http client init failed: {e}")))?; - let req = client.get(&url).header("Authorization", auth_header); - - let resp = self.with_auth_tag(req).send().await?; - if !resp.status().is_success() { - let status = resp.status().as_u16(); - let body = resp.text().await.unwrap_or_default(); - return Err(CliError::Relay { status, body }); - } - - resp.bytes().await.map_err(CliError::Network) + self.with_retry_body(|| { + let url = url.clone(); + let client = client.clone(); + async move { + let auth_header = sign_blossom_get(&self.keys, &url)?; + let resp = self + .with_auth_tag(client.get(&url).header("Authorization", auth_header)) + .send() + .await?; + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + return Err(CliError::Relay { status, body }); + } + resp.bytes().await.map_err(CliError::Network) + } + }) + .await } async fn handle_response(&self, resp: reqwest::Response) -> Result { @@ -951,6 +1431,869 @@ pub fn normalize_write_response(raw: &str) -> String { raw.to_string() } +#[cfg(test)] +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, + }; + + // ---- parse_retry_in_secs ---- + + #[test] + fn parse_relay_json_with_error_field() { + let body = r#"{"error":"rate-limited: quota exceeded; retry in 5s"}"#; + assert_eq!(parse_retry_in_secs(body), Some(5)); + } + + #[test] + fn parse_relay_json_with_message_field() { + let body = r#"{"message":"back off; retry in 3s please"}"#; + assert_eq!(parse_retry_in_secs(body), Some(3)); + } + + #[test] + fn parse_retry_in_zero_seconds() { + let body = r#"{"error":"retry in 0s"}"#; + assert_eq!(parse_retry_in_secs(body), Some(0)); + } + + #[test] + fn parse_garbled_body_returns_none() { + assert_eq!(parse_retry_in_secs("not json at all"), None); + } + + #[test] + fn parse_missing_retry_pattern_returns_none() { + let body = r#"{"error":"rate-limited, please slow down"}"#; + assert_eq!(parse_retry_in_secs(body), None); + } + + #[test] + fn parse_empty_body_returns_none() { + assert_eq!(parse_retry_in_secs(""), None); + } + + // ---- parse_retry_hint_text ---- + + #[test] + fn hint_text_plain_extracted_field_returns_secs() { + // Shape produced by handle_response: JSON extracted, plain text arrives. + assert_eq!( + parse_retry_hint_text("rate-limited: quota exceeded; retry in 4s"), + Some(4) + ); + } + + #[test] + fn hint_text_raw_json_body_returns_secs() { + // Shape from download_media's inline error path: raw JSON body preserved. + assert_eq!( + parse_retry_hint_text(r#"{"error":"rate-limited: retry in 7s"}"#), + Some(7) + ); + } + + #[test] + fn hint_text_plain_no_pattern_returns_none() { + assert_eq!(parse_retry_hint_text("rate-limited: slow down"), None); + } + + #[test] + fn hint_text_empty_returns_none() { + assert_eq!(parse_retry_hint_text(""), None); + } + + // ---- is_moderation_kind ---- + + #[test] + fn moderation_kind_covers_9040_through_9044() { + for kind in 9040u16..=9044 { + assert!(is_moderation_kind(kind), "kind {kind} should be moderation"); + } + } + + #[test] + fn non_moderation_kinds_are_not_moderation() { + for kind in [1u16, 9039, 9045, 39000, 20000, 30023] { + assert!( + !is_moderation_kind(kind), + "kind {kind} should not be moderation" + ); + } + } + + // ---- jitter bounds ---- + + #[test] + fn jitter_stays_within_base() { + for attempt in 0..RETRY_BASE_SECS.len() as u32 { + let base = RETRY_BASE_SECS[attempt as usize]; + for _ in 0..100 { + let delay = jitter_delay(attempt).as_secs_f64(); + assert!( + (0.0..=base).contains(&delay), + "jitter {delay} out of [0, {base}]" + ); + } + } + } + + // ---- constant sanity ---- + + #[test] + fn retry_constants_are_sensible() { + assert_eq!(RETRY_MAX_ATTEMPTS, 3); + assert_eq!(RETRY_BASE_SECS.len(), (RETRY_MAX_ATTEMPTS - 1) as usize); + const { assert!(RETRY_IN_MAX_SECS > 0) }; + } + + // ---- env_duration_secs ---- + + #[test] + fn env_duration_secs_parsing() { + // All assertions share one env var key; sequential set/remove prevents races. + const KEY: &str = "BUZZ_CLI_TEST_DURATION_SECS"; + + // Valid numeric value is parsed. + std::env::set_var(KEY, "42"); + assert_eq!(env_duration_secs(KEY, 30), Duration::from_secs(42)); + + // Non-numeric falls back to default. + std::env::set_var(KEY, "not-a-number"); + assert_eq!(env_duration_secs(KEY, 30), Duration::from_secs(30)); + + // Zero is treated as invalid and falls back to default. + std::env::set_var(KEY, "0"); + assert_eq!(env_duration_secs(KEY, 30), Duration::from_secs(30)); + + // Unset uses the default. + std::env::remove_var(KEY); + assert_eq!(env_duration_secs(KEY, 30), Duration::from_secs(30)); + } +} + +/// Integration tests for the kind-aware retry policy and body-boundary coverage. +/// +/// These tests spin up a local HTTP server using axum and issue real HTTP requests +/// through `BuzzClient` to verify behavioural properties — not implementation details. +#[cfg(test)] +mod retry_policy_tests { + use std::net::SocketAddr; + use std::sync::atomic::{AtomicU32, Ordering}; + use std::sync::Arc; + + use axum::body::Body; + use axum::extract::State; + use axum::http::{HeaderMap, Response, StatusCode}; + use axum::routing::post; + use axum::Router; + use nostr::{EventBuilder, Keys, Kind}; + use tokio::net::TcpListener; + + use super::super::error::CliError; + use super::BuzzClient; + + /// Spawn a one-shot axum server on a random port. The handler `f` receives the + /// attempt counter (incremented before every call) and returns a `(StatusCode, + /// String)`. Returns the base URL and a join handle so the caller can assert + /// attempt counts after the test. + async fn test_server(f: F) -> (String, Arc) + where + F: Fn(u32) -> (StatusCode, String) + Send + Sync + 'static, + { + let counter = Arc::new(AtomicU32::new(0)); + let handler: Arc (StatusCode, String) + Send + Sync> = Arc::new(f); + let state = (handler, counter.clone()); + + type S = ( + Arc (StatusCode, String) + Send + Sync>, + Arc, + ); + let app = Router::new() + .route( + "/events", + post( + |State((handler, ctr)): State, _headers: HeaderMap, _body: Body| async move { + let n = ctr.fetch_add(1, Ordering::SeqCst) + 1; + let (status, body) = handler(n); + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Body::from(body)) + .unwrap() + }, + ), + ) + .with_state(state); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr: SocketAddr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + (format!("http://{addr}"), counter) + } + + fn test_client(base_url: &str) -> BuzzClient { + let keys = Keys::generate(); + BuzzClient::new(base_url.to_string(), keys, None, None).unwrap() + } + + fn make_moderation_event(keys: &Keys, kind: u16) -> nostr::Event { + EventBuilder::new(Kind::Custom(kind), "") + .sign_with_keys(keys) + .unwrap() + } + + fn make_stored_event(keys: &Keys) -> nostr::Event { + EventBuilder::new(Kind::TextNote, "hi") + .sign_with_keys(keys) + .unwrap() + } + + /// A moderation command (kind 9040) that fails the first attempt with HTTP 429 + /// carrying a plain (non-relay-ingest) body is NOT retried — surfaces as + /// `DeliveryUnknown`. + #[tokio::test] + async fn moderation_kind_non_ingest_429_returns_delivery_unknown() { + let (url, attempts) = test_server(|_n| { + ( + StatusCode::TOO_MANY_REQUESTS, + r#"{"error":"slow down"}"#.to_string(), + ) + }) + .await; + let client = test_client(&url); + let event = make_moderation_event(client.keys(), 9040); + let err = client.submit_event(event).await.unwrap_err(); + assert!( + matches!(err, CliError::DeliveryUnknown(_)), + "expected DeliveryUnknown, got {err:?}" + ); + // Non-ingest 429 must not be retried — exactly 1 attempt. + assert_eq!( + attempts.load(Ordering::SeqCst), + 1, + "must not retry non-ingest 429" + ); + } + + /// A moderation command (kind 9041) that gets a relay-ingest 429 (production JSON + /// envelope `{"error":"rate-limited: ..."}`) IS retried, and the `retry in Ns` hint + /// is honoured. + /// + /// Uses a 2s hint; jitter max for attempt 0 is 0.5s, so asserting elapsed ≥ 2s + /// cleanly distinguishes hint-honoured from jitter-fallback. + #[tokio::test] + async fn moderation_kind_ingest_429_is_retried_until_success() { + let (url, attempts) = test_server(|n| { + if n < 2 { + ( + StatusCode::TOO_MANY_REQUESTS, + // Exact production envelope: api_error() wraps every message as + // {"error":"..."}. The extracted field starts with "rate-limited:" + // so the command is retried; the hint is honoured. + r#"{"error":"rate-limited: quota exceeded; retry in 2s"}"#.to_string(), + ) + } else { + ( + StatusCode::OK, + r#"{"event_id":"abc","accepted":true,"message":""}"#.to_string(), + ) + } + }) + .await; + let client = test_client(&url); + let event = make_moderation_event(client.keys(), 9041); + let t0 = std::time::Instant::now(); + let result = client.submit_event(event).await; + let elapsed = t0.elapsed(); + assert!( + result.is_ok(), + "expected Ok after ingest-429 retry, got {result:?}" + ); + assert!( + attempts.load(Ordering::SeqCst) >= 2, + "must have retried at least once" + ); + assert!( + elapsed.as_secs_f64() >= 2.0, + "elapsed {:.2}s < 2s — hint was not honoured (fell back to jitter)", + elapsed.as_secs_f64() + ); + } + + /// A moderation command that receives the canonical pre-ingest 429 on EVERY + /// attempt exhausts the retry budget and surfaces `CliError::Relay { status: 429 }` — + /// NOT `DeliveryUnknown`. The relay provably never executed the command on any + /// attempt, so the caller must be told it is safe to retry. + #[tokio::test] + async fn exhausted_ingest_429_returns_relay_429_retryable() { + let (url, attempts) = test_server(|_n| { + ( + StatusCode::TOO_MANY_REQUESTS, + r#"{"error":"rate-limited: quota exceeded; retry in 0s"}"#.to_string(), + ) + }) + .await; + let client = test_client(&url); + let event = make_moderation_event(client.keys(), 9040); + let err = client.submit_event(event).await.unwrap_err(); + + // Must be Relay(429), not DeliveryUnknown. + assert!( + matches!(err, CliError::Relay { status: 429, .. }), + "exhausted ingest 429 must surface as Relay(429), got {err:?}" + ); + // Must NOT be retryable:false. + assert!( + crate::error::is_retryable_error(&err), + "Relay(429) must be retryable; got {err:?}" + ); + // All RETRY_MAX_ATTEMPTS must have been tried. + assert_eq!( + attempts.load(Ordering::SeqCst), + 3, + "all retry attempts must fire for exhausted ingest 429" + ); + } + + /// A moderation command (kind 9042) that gets HTTP 502 returns `DeliveryUnknown` + /// immediately — proxy errors leave relay execution state ambiguous. + #[tokio::test] + async fn moderation_kind_502_returns_delivery_unknown() { + let (url, attempts) = + test_server(|_n| (StatusCode::BAD_GATEWAY, "bad gateway".to_string())).await; + let client = test_client(&url); + let event = make_moderation_event(client.keys(), 9042); + let err = client.submit_event(event).await.unwrap_err(); + assert!( + matches!(err, CliError::DeliveryUnknown(_)), + "expected DeliveryUnknown for 502, got {err:?}" + ); + // 502 must not be retried — exactly 1 attempt. + assert_eq!( + attempts.load(Ordering::SeqCst), + 1, + "must not retry 502 for moderation kind" + ); + } + + /// When all retry attempts are connect-failures (the relay definitively never saw + /// the request), `submit_event` must return `CliError::Network` with + /// `retryable:true` — not `DeliveryUnknown`. Connect-failure is the one error + /// condition the implementation itself identifies as confirmed-unreceived. + #[tokio::test] + async fn exhausted_connect_failures_return_network_retryable() { + // Bind a port, capture the address, then drop the listener so every + // subsequent connect attempt is refused immediately. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + drop(listener); + + let base = format!("http://{addr}"); + let client = test_client(&base); + let event = make_moderation_event(client.keys(), 9040); + let err = client.submit_event(event).await.unwrap_err(); + // Must be Network (retryable), not DeliveryUnknown (retryable:false). + assert!( + matches!(err, super::super::error::CliError::Network(_)), + "exhausted connect failures must surface as Network, got {err:?}" + ); + // Confirm the error description does not suggest ambiguous delivery. + let description = format!("{err:?}"); + assert!( + !description.contains("outcome unknown"), + "connect failure must not be labeled DeliveryUnknown; got: {description}" + ); + } + + /// A stored (non-moderation) event submitted to a server that returns 502 on the + /// first attempt and then succeeds is retried under the standard policy. + #[tokio::test] + async fn stored_event_502_is_retried_under_standard_policy() { + let (url, attempts) = test_server(|n| { + if n == 1 { + (StatusCode::BAD_GATEWAY, "transient".to_string()) + } else { + ( + StatusCode::OK, + r#"{"event_id":"abc","accepted":true,"message":""}"#.to_string(), + ) + } + }) + .await; + let client = test_client(&url); + let event = make_stored_event(client.keys()); + let result = client.submit_event(event).await; + assert!( + result.is_ok(), + "expected Ok after 502 retry for stored event, got {result:?}" + ); + assert!( + attempts.load(Ordering::SeqCst) >= 2, + "must have retried at least once" + ); + } + + /// Spin up a one-shot axum server that handles `GET /info` (and any other GET). + /// Same contract as `test_server` — returns base URL and attempt counter. + async fn get_server(f: F) -> (String, Arc) + where + F: Fn(u32) -> (StatusCode, String) + Send + Sync + 'static, + { + let counter = Arc::new(AtomicU32::new(0)); + let handler: Arc (StatusCode, String) + Send + Sync> = Arc::new(f); + let state = (handler, counter.clone()); + + type S = ( + Arc (StatusCode, String) + Send + Sync>, + Arc, + ); + let app = Router::new() + .route( + "/{*path}", + axum::routing::get( + |State((handler, ctr)): State, _headers: HeaderMap| async move { + let n = ctr.fetch_add(1, Ordering::SeqCst) + 1; + let (status, body) = handler(n); + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Body::from(body)) + .unwrap() + }, + ), + ) + .with_state(state); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr: SocketAddr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + (format!("http://{addr}"), counter) + } + + /// `with_retry_body` retries transient HTTP 502 on a read path (`get_authed`) + /// and succeeds on the next attempt. + #[tokio::test] + async fn query_502_is_retried_then_succeeds() { + let (url, attempts) = get_server(|n| { + if n == 1 { + ( + StatusCode::BAD_GATEWAY, + "transient gateway error".to_string(), + ) + } else { + (StatusCode::OK, r#"{"ok":true}"#.to_string()) + } + }) + .await; + let client = test_client(&url); + let result = client.get_authed("/info").await; + assert!( + result.is_ok(), + "expected Ok after 502 retry, got {result:?}" + ); + assert!( + attempts.load(Ordering::SeqCst) >= 2, + "must have retried at least once" + ); + } + + /// `with_retry_body` retries a 429 with a `retry in Ns` hint, honours the hint + /// delay (not the shorter jitter fallback), and ultimately succeeds. + /// + /// Uses a 2s hint; jitter max for attempt 0 is 0.5s, so asserting elapsed ≥ 2s + /// cleanly distinguishes hint-honoured from jitter-fallback. + #[tokio::test] + async fn query_429_with_hint_is_retried() { + let (url, attempts) = get_server(|n| { + if n < 2 { + ( + StatusCode::TOO_MANY_REQUESTS, + // handle_response extracts the "error" field; the plain text + // "rate-limited: retry in 2s" then reaches parse_retry_hint_text. + r#"{"error":"rate-limited: retry in 2s"}"#.to_string(), + ) + } else { + (StatusCode::OK, r#"{"ok":true}"#.to_string()) + } + }) + .await; + let client = test_client(&url); + let t0 = std::time::Instant::now(); + // Measure from just before attempt 1 fires so we capture the inter-attempt wait. + let result = client.get_authed("/info").await; + // Record elapsed after attempt 1 returns (inside the future) is not possible + // directly, but the total includes the hint sleep; jitter max is 0.5s so ≥ 2s + // proves the hint was honoured. + let elapsed = t0.elapsed(); + assert!( + result.is_ok(), + "expected Ok after 429 retry, got {result:?}" + ); + assert!( + attempts.load(Ordering::SeqCst) >= 2, + "must have retried at least once" + ); + assert!( + elapsed.as_secs_f64() >= 2.0, + "elapsed {:.2}s < 2s — hint was not honoured (fell back to jitter)", + elapsed.as_secs_f64() + ); + } + + /// A definitive 4xx (403 Forbidden) is NOT retried — exactly 1 attempt. + #[tokio::test] + async fn query_403_is_not_retried() { + let (url, attempts) = get_server(|_n| { + ( + StatusCode::FORBIDDEN, + r#"{"error":"not allowed"}"#.to_string(), + ) + }) + .await; + let client = test_client(&url); + let result = client.get_authed("/info").await; + assert!( + matches!(result, Err(CliError::Relay { status: 403, .. })), + "expected Relay 403 error, got {result:?}" + ); + assert_eq!( + attempts.load(Ordering::SeqCst), + 1, + "403 must not be retried" + ); + } + + /// `with_retry_body` retries on `is_body()` network errors (F2: body transfer inside + /// the retry boundary). Verified by confirming that a call through `get_authed` + /// (which uses `with_retry_body`) retries when the server drops the connection after + /// sending headers. We simulate body loss by returning an intentionally truncated + /// chunked response that reqwest will surface as an `is_body()` error. + /// + /// This test uses a raw TCP server to write partial HTTP responses; axum cannot + /// easily simulate mid-body connection drops. + #[tokio::test] + async fn with_retry_body_retries_on_body_transfer_failure() { + use tokio::io::AsyncWriteExt; + + let counter = Arc::new(AtomicU32::new(0)); + let counter2 = counter.clone(); + + // Bind a raw TCP listener. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + break; + }; + let n = counter2.fetch_add(1, Ordering::SeqCst) + 1; + + // Consume the request (required to avoid connection reset by server). + let mut buf = vec![0u8; 4096]; + use tokio::io::AsyncReadExt; + let _ = tokio::time::timeout( + std::time::Duration::from_millis(100), + stream.read(&mut buf), + ) + .await; + + if n < 3 { + // Attempts 1 & 2: send valid headers claiming a body, then drop. + let partial = b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: 100\r\n\r\n{\"partial\":"; + let _ = stream.write_all(partial).await; + // Drop the stream without completing the body — causes is_body() on client. + } else { + // Attempt 3: complete response. + let ok = b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: 2\r\n\r\n{}"; + let _ = stream.write_all(ok).await; + } + } + }); + + let base = format!("http://{addr}"); + // get_authed internally uses with_retry_body — the body read is inside the retry loop. + let client = test_client(&base); + // Stub path: the raw TCP server ignores the URL and always responds based on attempt count. + let result = client.get_authed("/any-path").await; + assert!( + result.is_ok(), + "expected Ok after body-loss retries, got {result:?}" + ); + assert_eq!( + counter.load(Ordering::SeqCst), + 3, + "expected 3 attempts (2 body-loss + 1 success)" + ); + } + + /// `submit_event` (non-moderation kind) uses `with_retry_body` — the full + /// operation including response body read is inside the retry boundary. + /// A partial-body drop after 200 headers must be retried with the same + /// serialized event bytes (and a fresh NIP-98 auth per attempt). + #[tokio::test] + async fn stored_event_body_loss_is_retried_with_same_event_bytes() { + use tokio::io::AsyncReadExt; + use tokio::io::AsyncWriteExt; + + let counter = Arc::new(AtomicU32::new(0)); + let counter2 = counter.clone(); + let bodies: Arc>>> = + Arc::new(std::sync::Mutex::new(Vec::new())); + let bodies2 = bodies.clone(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + break; + }; + let n = counter2.fetch_add(1, Ordering::SeqCst) + 1; + + // Read the full HTTP request so we can capture the body. + let mut buf = vec![0u8; 8192]; + let _ = tokio::time::timeout( + std::time::Duration::from_millis(200), + stream.read(&mut buf), + ) + .await; + // Capture raw request bytes for assertion. + let body_end = buf + .windows(4) + .position(|w| w == b"\r\n\r\n") + .map(|i| i + 4) + .unwrap_or(0); + let payload = buf[body_end..].to_vec(); + bodies2.lock().unwrap().push(payload); + + if n < 3 { + // Partial body drop. + let partial = b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: 100\r\n\r\n{\"partial\":"; + let _ = stream.write_all(partial).await; + } else { + let ok = b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: 41\r\n\r\n{\"event_id\":\"abc\",\"accepted\":true,\"message\":\"\"}"; + let _ = stream.write_all(ok).await; + } + } + }); + + let base = format!("http://{addr}"); + let client = test_client(&base); + let event = make_stored_event(client.keys()); + let result = client.submit_event(event).await; + assert!( + result.is_ok(), + "expected Ok after body-loss retries, got {result:?}" + ); + assert_eq!( + counter.load(Ordering::SeqCst), + 3, + "expected 3 attempts (2 body-loss + 1 success)" + ); + // All three attempts must have sent the same serialized event bytes. + let captured = bodies.lock().unwrap(); + assert_eq!(captured.len(), 3, "must have captured 3 request bodies"); + // Each attempt's payload must be identical (same signed event bytes). + assert_eq!( + captured[0], captured[1], + "attempt 1 and 2 must use identical event bytes" + ); + assert_eq!( + captured[1], captured[2], + "attempt 2 and 3 must use identical event bytes" + ); + } + + /// `upload_file` uses `with_retry_body` — the full operation including response + /// body read is inside the retry boundary. A partial-body drop after 200 headers + /// must be retried with identical file bytes and a fresh Blossom auth per attempt. + #[tokio::test] + async fn upload_body_loss_is_retried_with_same_file_bytes() { + use std::io::Write; + use tokio::io::AsyncReadExt; + use tokio::io::AsyncWriteExt; + + // Write a minimal JPEG file so MIME detection works. + let mut tmp = tempfile::NamedTempFile::new().unwrap(); + // JPEG magic + JFIF app0 marker: enough for `infer` to detect image/jpeg. + let jpeg_header: &[u8] = &[ + 0xFF, 0xD8, 0xFF, 0xE0, 0x00, 0x10, 0x4A, 0x46, 0x49, 0x46, 0x00, 0x01, + ]; + tmp.write_all(jpeg_header).unwrap(); + let file_path = tmp.path().to_str().unwrap().to_string(); + + let counter = Arc::new(AtomicU32::new(0)); + let counter2 = counter.clone(); + let auth_headers: Arc>> = + Arc::new(std::sync::Mutex::new(Vec::new())); + let auth_headers2 = auth_headers.clone(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + break; + }; + let n = counter2.fetch_add(1, Ordering::SeqCst) + 1; + + // Read the request headers to extract the Authorization value. + let mut buf = vec![0u8; 8192]; + let _ = tokio::time::timeout( + std::time::Duration::from_millis(200), + stream.read(&mut buf), + ) + .await; + // Extract the Authorization header value. + let req_str = String::from_utf8_lossy(&buf); + let auth = req_str + .lines() + .find(|l| l.to_lowercase().starts_with("authorization:")) + .map(|l| l.to_string()) + .unwrap_or_default(); + auth_headers2.lock().unwrap().push(auth); + + if n < 3 { + // Partial body drop. + let partial = b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: 100\r\n\r\n{\"partial\":"; + let _ = stream.write_all(partial).await; + } else { + // Valid BlobDescriptor response. + let ok_body = r#"{"url":"https://relay.test/media/aabbcc.jpg","sha256":"aabbcc","size":12,"type":"image/jpeg","uploaded":0}"#; + let ok = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\r\n{}", + ok_body.len(), + ok_body + ); + let _ = stream.write_all(ok.as_bytes()).await; + } + } + }); + + let base = format!("http://{addr}"); + let client = test_client(&base); + let result = client.upload_file(&file_path).await; + assert!( + result.is_ok(), + "expected Ok after upload body-loss retries, got {result:?}" + ); + assert_eq!( + counter.load(Ordering::SeqCst), + 3, + "expected 3 upload attempts (2 body-loss + 1 success)" + ); + // Each attempt must carry a distinct Authorization header (fresh Blossom auth). + let auths = auth_headers.lock().unwrap(); + assert_eq!(auths.len(), 3, "must have captured 3 auth headers"); + // All three must be non-empty (auth was signed). + assert!( + auths.iter().all(|a| a.contains("Nostr ")), + "each attempt must carry Nostr auth" + ); + } + + /// When all retry attempts for a stored event end with a partial body (200 + /// headers, dropped connection), the final error must be `DeliveryUnknown` + /// (retryable:false) — the relay may have stored the event on any attempt, so + /// an outer re-sign would risk a duplicate visible write. All three attempts + /// must fire with identical serialized event bytes. + #[tokio::test] + async fn stored_event_all_body_losses_return_delivery_unknown() { + use tokio::io::AsyncWriteExt; + + let bodies: Arc>>> = + Arc::new(std::sync::Mutex::new(Vec::new())); + let bodies2 = bodies.clone(); + let counter = Arc::new(AtomicU32::new(0)); + let counter2 = counter.clone(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::spawn(async move { + use tokio::io::AsyncReadExt; + loop { + let Ok((mut stream, _)) = listener.accept().await else { + break; + }; + counter2.fetch_add(1, Ordering::SeqCst); + let mut buf = vec![0u8; 8192]; + let _ = tokio::time::timeout( + std::time::Duration::from_millis(200), + stream.read(&mut buf), + ) + .await; + // Extract the request body (after the blank line separating headers). + let raw = buf.split(|&b| b == 0).next().unwrap_or(&buf).to_vec(); + if let Some(pos) = raw.windows(4).position(|w| w == b"\r\n\r\n") { + bodies2.lock().unwrap().push(raw[pos + 4..].to_vec()); + } + // Partial body: send headers + truncated body, then drop. + let _ = stream + .write_all( + b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: 100\r\n\r\n{\"partial\":", + ) + .await; + // Drop stream — causes body-loss error on the client side. + } + }); + + let base = format!("http://{addr}"); + let client = test_client(&base); + let event = make_stored_event(client.keys()); + let err = client.submit_event(event).await.unwrap_err(); + + // Final error must be DeliveryUnknown — relay may have accepted any attempt. + assert!( + matches!(err, CliError::DeliveryUnknown(_)), + "all-body-loss exhaustion must return DeliveryUnknown, got {err:?}" + ); + // All RETRY_MAX_ATTEMPTS must have fired. + assert_eq!( + counter.load(Ordering::SeqCst), + 3, + "all 3 attempts must be made before surfacing DeliveryUnknown" + ); + // All attempts must have sent identical serialized event bytes. + let captured = bodies.lock().unwrap(); + if captured.len() >= 2 { + assert_eq!( + captured[0], captured[1], + "all attempts must use identical event bytes" + ); + } + } + + /// When all retry attempts for a stored event return HTTP 502, the final error + /// must be `DeliveryUnknown` (retryable:false) — a proxy 502 may occur after + /// the relay accepted the event. + #[tokio::test] + async fn stored_event_all_502s_return_delivery_unknown() { + let (url, attempts) = + test_server(|_n| (StatusCode::BAD_GATEWAY, "bad gateway".to_string())).await; + let client = test_client(&url); + let event = make_stored_event(client.keys()); + let err = client.submit_event(event).await.unwrap_err(); + + assert!( + matches!(err, CliError::DeliveryUnknown(_)), + "all-502 exhaustion must return DeliveryUnknown, got {err:?}" + ); + assert_eq!( + attempts.load(Ordering::SeqCst), + 3, + "all 3 attempts must fire before surfacing DeliveryUnknown" + ); + } +} + #[cfg(test)] mod tests { use super::{ diff --git a/crates/buzz-cli/src/error.rs b/crates/buzz-cli/src/error.rs index f4cdb36856..2edcd6aa9d 100644 --- a/crates/buzz-cli/src/error.rs +++ b/crates/buzz-cli/src/error.rs @@ -11,7 +11,7 @@ pub enum CliError { Relay { status: u16, body: String }, /// Network-level failure (connect, timeout, DNS) - #[error("network error: {0}")] + #[error("network error: {}", fmt_reqwest_error(.0))] Network(#[from] reqwest::Error), /// Auth missing or rejected (401/403) @@ -32,11 +32,58 @@ pub enum CliError { #[error("{0}")] NotFound(String), + /// A non-idempotent command's outcome is unknown: the request may have + /// reached the relay, but the response was lost. Never auto-retried and + /// never labeled retryable — the relay executes these commands before any + /// dedup, so a blind re-run can duplicate the mutation. + #[error("delivery unknown: {0}")] + DeliveryUnknown(String), + /// Catch-all for unexpected failures #[error("{0}")] Other(String), } +/// Walk the full `std::error::Error::source()` chain on a `reqwest::Error` +/// and render it as a colon-separated string, e.g. +/// `error sending request: dns error: failed to lookup address information: ...` +fn fmt_reqwest_error(e: &reqwest::Error) -> String { + let mut msg = e.to_string(); + let mut source: &dyn std::error::Error = e; + while let Some(cause) = source.source() { + let cause_str = cause.to_string(); + if !msg.contains(&cause_str) { + msg.push_str(": "); + msg.push_str(&cause_str); + } + source = cause; + } + msg +} + +/// Returns `true` when the error is transient and a retry may succeed. +/// +/// Transport-level network errors (connect failure, timeout, mid-request, +/// mid-body transfer, or body decode failure) and relay overload responses +/// (429 / 502 / 503 / 504) are retryable. `DeliveryUnknown` is never +/// retryable: the operation may already have executed. All other errors +/// indicate a permanent failure: auth, bad input, builder errors, or logic +/// errors. +pub fn is_retryable_error(e: &CliError) -> bool { + match e { + CliError::Network(ref net_err) => { + net_err.is_connect() + || net_err.is_timeout() + || net_err.is_request() + || net_err.is_body() + || net_err.is_decode() + } + CliError::Relay { status, .. } => matches!(status, 429 | 502 | 503 | 504), + CliError::DeliveryUnknown(_) => false, + _ => false, + } +} + /// Map CliError to process exit code. /// 0=success (not an error), 1=user/not-found, 2=network/relay, 3=auth, /// 4=other, 5=write conflict (NIP-33 dominated head). @@ -55,12 +102,13 @@ pub fn exit_code(e: &CliError) -> i32 { CliError::Key(_) => 3, CliError::Conflict(_) => 5, CliError::NotFound(_) => 1, + CliError::DeliveryUnknown(_) => 2, CliError::Other(_) => 4, } } /// Serialize error to JSON and write to stderr. -/// Format: {"error": "", "message": ""} +/// Format: {"error": "", "message": "", "retryable": } pub fn print_error(e: &CliError) { let category = match e { CliError::Usage(_) => "user_error", @@ -76,11 +124,113 @@ pub fn print_error(e: &CliError) { CliError::Key(_) => "key_error", CliError::Conflict(_) => "conflict", CliError::NotFound(_) => "not_found", + CliError::DeliveryUnknown(_) => "delivery_unknown", CliError::Other(_) => "error", }; let obj = serde_json::json!({ "error": category, "message": e.to_string(), + "retryable": is_retryable_error(e), }); eprintln!("{}", obj); } + +#[cfg(test)] +mod tests { + use super::*; + + // ---- is_retryable_error ---- + + #[test] + fn network_builder_errors_are_not_retryable() { + // A bad URL produces a builder-level reqwest::Error (is_builder() == true). + // Builder errors are not transport failures — not retryable. + // Transport errors (is_connect/timeout/request) require live I/O to construct; + // the predicate here mirrors with_retry's condition exactly. + let e = reqwest::Client::new().get("not-a-url").build().unwrap_err(); + assert!(e.is_builder(), "expected a builder error from bad URL"); + assert!(!is_retryable_error(&CliError::Network(e))); + } + + #[test] + fn relay_429_502_503_504_are_retryable() { + for status in [429u16, 502, 503, 504] { + assert!( + is_retryable_error(&CliError::Relay { + status, + body: String::new() + }), + "status {status} should be retryable" + ); + } + } + + #[test] + fn relay_400_401_403_404_422_are_not_retryable() { + for status in [400u16, 401, 403, 404, 422] { + assert!( + !is_retryable_error(&CliError::Relay { + status, + body: String::new() + }), + "status {status} should not be retryable" + ); + } + } + + #[test] + fn other_errors_are_not_retryable() { + assert!(!is_retryable_error(&CliError::Usage("bad flag".into()))); + assert!(!is_retryable_error(&CliError::Auth("missing key".into()))); + assert!(!is_retryable_error(&CliError::Key("bad key".into()))); + assert!(!is_retryable_error(&CliError::Conflict( + "superseded".into() + ))); + assert!(!is_retryable_error(&CliError::NotFound("gone".into()))); + assert!(!is_retryable_error(&CliError::Other("unexpected".into()))); + } + + // ---- print_error "retryable" field ---- + + #[test] + fn json_error_includes_retryable_field_for_network() { + // Builder errors (bad URL) are not transport-level — retryable: false. + // This test verifies the JSON shape and that the field is present. + let e = reqwest::Client::new().get("not-a-url").build().unwrap_err(); + let err = CliError::Network(e); + let v = serde_json::json!({ + "error": "network_error", + "message": err.to_string(), + "retryable": is_retryable_error(&err), + }); + assert_eq!(v["retryable"].as_bool(), Some(false)); + assert_eq!(v["error"].as_str(), Some("network_error")); + } + + #[test] + fn json_error_retryable_false_for_usage() { + let err = CliError::Usage("bad flag".into()); + let v = serde_json::json!({ + "error": "user_error", + "message": err.to_string(), + "retryable": is_retryable_error(&err), + }); + assert_eq!(v["retryable"].as_bool(), Some(false)); + } + + // ---- Display source-chain ---- + + #[test] + fn network_display_includes_detail_beyond_prefix() { + let e = reqwest::Client::new().get("not-a-url").build().unwrap_err(); + let display = CliError::Network(e).to_string(); + assert!( + display.starts_with("network error:"), + "display should start with 'network error:': {display}" + ); + assert!( + display.len() > "network error: ".len(), + "display should contain error detail: {display}" + ); + } +} diff --git a/crates/buzz-ws-client/src/connection.rs b/crates/buzz-ws-client/src/connection.rs index b7b0fe7ea8..bec5b56bb4 100644 --- a/crates/buzz-ws-client/src/connection.rs +++ b/crates/buzz-ws-client/src/connection.rs @@ -13,6 +13,15 @@ use crate::message::{build_auth_event, parse_relay_message, OkResponse, RelayMes type WsStream = WebSocketStream>; +/// Seconds to wait for the relay to send the NIP-42 AUTH challenge after connecting. +pub const AUTH_CHALLENGE_TIMEOUT_SECS: u64 = 20; + +/// Seconds to wait for the relay's OK response to the AUTH event. +pub const AUTH_OK_TIMEOUT_SECS: u64 = 20; + +/// Seconds to wait for the relay's OK response to a published event. +pub const PUBLISH_OK_TIMEOUT_SECS: u64 = 30; + /// A NIP-42-capable WebSocket connection to a Nostr relay. pub struct NostrWsConnection { ws: WsStream, @@ -63,14 +72,18 @@ impl NostrWsConnection { keys: &Keys, auth_tag: Option<&Tag>, ) -> Result<(), WsClientError> { - let challenge = self.wait_for_auth_challenge(Duration::from_secs(5)).await?; + let challenge = self + .wait_for_auth_challenge(Duration::from_secs(AUTH_CHALLENGE_TIMEOUT_SECS)) + .await?; let auth_event = build_auth_event(&challenge, &self.relay_url, keys, auth_tag)?; let event_id = auth_event.id.to_hex(); self.send_raw(&json!(["AUTH", auth_event])).await?; - let ok = self.wait_for_ok(&event_id, Duration::from_secs(5)).await?; + let ok = self + .wait_for_ok(&event_id, Duration::from_secs(AUTH_OK_TIMEOUT_SECS)) + .await?; if !ok.accepted { return Err(WsClientError::AuthFailed(ok.message)); } @@ -83,7 +96,8 @@ impl NostrWsConnection { pub async fn send_event(&mut self, event: Event) -> Result { let event_id = event.id.to_hex(); self.send_raw(&json!(["EVENT", event])).await?; - self.wait_for_ok(&event_id, Duration::from_secs(10)).await + self.wait_for_ok(&event_id, Duration::from_secs(PUBLISH_OK_TIMEOUT_SECS)) + .await } /// Receives the next relay message, waiting up to `timeout_dur`. @@ -278,3 +292,23 @@ pub async fn publish_event( .map_err(|_| WsClientError::Timeout)?; result } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn auth_challenge_timeout_meets_floor() { + const { assert!(AUTH_CHALLENGE_TIMEOUT_SECS >= 20) }; + } + + #[test] + fn auth_ok_timeout_meets_floor() { + const { assert!(AUTH_OK_TIMEOUT_SECS >= 20) }; + } + + #[test] + fn publish_ok_timeout_meets_floor() { + const { assert!(PUBLISH_OK_TIMEOUT_SECS >= 30) }; + } +}