From 8465af96e191c103730cffe4439cadec786a3149 Mon Sep 17 00:00:00 2001 From: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz> Date: Thu, 27 Aug 2026 14:16:07 -0600 Subject: [PATCH 1/3] fix(desktop): bound HTTP event submission Signed-off-by: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz> --- .../src/commands/personas/snapshot/import.rs | 21 ++-- .../src-tauri/src/commands/team_snapshot.rs | 21 ++-- desktop/src-tauri/src/egress_guard_tests.rs | 107 ++++++++++++------ desktop/src-tauri/src/huddle/pipeline.rs | 17 ++- desktop/src-tauri/src/relay.rs | 33 +----- desktop/src-tauri/src/relay/submit.rs | 79 +++++++++++-- desktop/src-tauri/src/relay/tests.rs | 72 +++++++++++- 7 files changed, 245 insertions(+), 105 deletions(-) diff --git a/desktop/src-tauri/src/commands/personas/snapshot/import.rs b/desktop/src-tauri/src/commands/personas/snapshot/import.rs index 75a1edea65e..8f27f1411f5 100644 --- a/desktop/src-tauri/src/commands/personas/snapshot/import.rs +++ b/desktop/src-tauri/src/commands/personas/snapshot/import.rs @@ -819,19 +819,14 @@ pub(crate) async fn submit_engram_event( // wait produces a stale `created_at` that the relay will reject. crate::relay_admission::wait_for_rate_limit().await; let auth = build_nip98_auth_header_for_keys(agent_keys, &Method::POST, url, event_json)?; - let mut request = state - .http_client - .post(url) - .header("Authorization", auth) - .header("Content-Type", "application/json"); - if let Some(tag) = auth_tag { - request = request.header("x-auth-tag", tag); - } - let response = request - .body(event_json.to_vec()) - .send() - .await - .map_err(|e| crate::relay::classify_request_error(&e))?; + let response = crate::relay::send_event_http_request( + &state.http_client, + url, + &auth, + auth_tag, + event_json.to_vec(), + ) + .await?; if !response.status().is_success() { let msg = crate::relay::relay_error_message(response).await; diff --git a/desktop/src-tauri/src/commands/team_snapshot.rs b/desktop/src-tauri/src/commands/team_snapshot.rs index e4c08a14be0..0182bd2ea3c 100644 --- a/desktop/src-tauri/src/commands/team_snapshot.rs +++ b/desktop/src-tauri/src/commands/team_snapshot.rs @@ -914,19 +914,14 @@ pub(crate) async fn submit_engram_event( // wait produces a stale `created_at` that the relay will reject. crate::relay_admission::wait_for_rate_limit().await; let auth = build_nip98_auth_header_for_keys(agent_keys, &Method::POST, url, event_json)?; - let mut request = state - .http_client - .post(url) - .header("Authorization", auth) - .header("Content-Type", "application/json"); - if let Some(tag) = auth_tag { - request = request.header("x-auth-tag", tag); - } - let response = request - .body(event_json.to_vec()) - .send() - .await - .map_err(|e| crate::relay::classify_request_error(&e))?; + let response = crate::relay::send_event_http_request( + &state.http_client, + url, + &auth, + auth_tag, + event_json.to_vec(), + ) + .await?; if !response.status().is_success() { let msg = crate::relay::relay_error_message(response).await; diff --git a/desktop/src-tauri/src/egress_guard_tests.rs b/desktop/src-tauri/src/egress_guard_tests.rs index 0c2a9573af6..8f3e12876f7 100644 --- a/desktop/src-tauri/src/egress_guard_tests.rs +++ b/desktop/src-tauri/src/egress_guard_tests.rs @@ -249,37 +249,40 @@ fn src_rust_files() -> Vec { } /// Site-granular `/events` inventory: `(file suffix, expected non-comment -/// `/events` occurrences, expected guard call sites — full-path calls into -/// the egress-guard module)`. +/// `/events` occurrences, expected guard call sites, expected calls into the +/// bounded HTTP event-submit funnel)`. /// -/// Every entry pairs the URL-construction count with the guard-call count for -/// that file, so BOTH of these fail the scan (not just a brand-new file): +/// Every production entry pairs the URL-construction count with both safety +/// boundaries for that file, so all of these fail the scan (not just a +/// brand-new file): /// - adding an unguarded ninth `/events` site inside an already-listed file -/// (count goes up without a matching table update), and -/// - removing/refactoring away a guard call while its egress site remains. +/// (count goes up without a matching table update), +/// - removing/refactoring away a guard call while its egress site remains, +/// - bypassing `send_event_http_request` and its per-request deadline. /// -/// Updating a row here is the deliberate act that must accompany wiring the -/// guard + adding an injection test for the new site. -const EVENTS_INVENTORY: &[(&str, usize, usize)] = &[ +/// Updating a row here is the deliberate act that must accompany wiring both +/// boundaries and adding an injection test for a new production site. +const EVENTS_INVENTORY: &[(&str, usize, usize, usize)] = &[ // Production egress boundaries (see egress_guard.rs table): - ("src/relay.rs", 2, 2), // boundaries 2, 4 - ("src/relay/submit.rs", 1, 1), // boundaries 1 + 3 (shared funnel) - ("src/huddle/pipeline.rs", 1, 1), // boundary 5 - ("src/commands/team_snapshot.rs", 1, 1), // boundary 6 - ("src/commands/personas/snapshot/import.rs", 2, 1), // boundary 7 + its in-file injection-test fixture URL - ("src/native_websocket.rs", 0, 2), // boundary 8 (WS frames; no events URL) - // Test-only fixtures — no production egress, no guard: - ("src/relay_admission.rs", 1, 0), - ("src/archive/mod_tests.rs", 1, 0), - ("src/managed_agents/persona_events/tests.rs", 1, 0), - ("src/commands/team_snapshot/tests.rs", 1, 0), + ("src/relay.rs", 2, 2, 2), // boundaries 2, 4 + ("src/relay/submit.rs", 1, 1, 1), // boundaries 1 + 3 (shared funnel) + ("src/huddle/pipeline.rs", 1, 1, 1), // boundary 5 + ("src/commands/team_snapshot.rs", 1, 1, 1), // boundary 6 + ("src/commands/personas/snapshot/import.rs", 2, 1, 1), // boundary 7 + its in-file injection-test fixture URL + ("src/native_websocket.rs", 0, 2, 0), // boundary 8 (WS frames; no events URL) + // Test-only fixtures — no production egress, no guard or production submit call: + ("src/relay/tests.rs", 2, 0, 0), + ("src/relay_admission.rs", 1, 0, 0), + ("src/archive/mod_tests.rs", 1, 0, 0), + ("src/managed_agents/persona_events/tests.rs", 1, 0, 0), + ("src/commands/team_snapshot/tests.rs", 1, 0, 0), // Mock-relay route in its in-file tests; production publish goes through // the guarded boundary-1 funnel (`submit_signed_event_at_with_keys`). - ("src/commands/personas/sharing.rs", 1, 0), + ("src/commands/personas/sharing.rs", 1, 0, 0), // Loopback submit relay in `identity_archive.rs`'s in-file regen tests; // production archive/unarchive publish through the guarded boundary-1 // funnel via `submit_event`. - ("src/commands/identity_archive.rs", 1, 0), + ("src/commands/identity_archive.rs", 1, 0, 0), ]; // Needles are assembled at runtime so this scan file itself contains no @@ -290,11 +293,22 @@ fn events_needle() -> String { fn guard_needle() -> String { ["egress_guard::", "assert_no_key_backup"].concat() } +fn event_submit_call_count(content: &str) -> usize { + let needle = &["send_event_", "http_request("].concat(); + content + .lines() + .filter(|line| { + line.contains(needle) + && !line.trim_start().starts_with("//") + && !line.contains("fn send_event_") + }) + .count() +} /// Pure scan core over `(relative path, content)` pairs. Returns violations; /// empty means every file matches its inventory row exactly (files absent -/// from the table are expected to have zero `/events` sites and zero guard -/// calls). +/// from the table are expected to have zero `/events` sites, guard calls, and +/// bounded HTTP event-submit calls). fn events_inventory_violations(files: &[(String, String)]) -> Vec { let events = events_needle(); let guard = guard_needle(); @@ -303,9 +317,9 @@ fn events_inventory_violations(files: &[(String, String)]) -> Vec { for (rel, content) in files { let expected = EVENTS_INVENTORY .iter() - .find(|(suffix, _, _)| rel.ends_with(suffix)) - .map(|&(_, e, g)| (e, g)) - .unwrap_or((0, 0)); + .find(|(suffix, _, _, _)| rel.ends_with(suffix)) + .map(|&(_, e, g, s)| (e, g, s)) + .unwrap_or((0, 0, 0)); let mut event_sites = Vec::new(); for (i, line) in content.lines().enumerate() { @@ -317,15 +331,18 @@ fn events_inventory_violations(files: &[(String, String)]) -> Vec { } } let guard_count = content.matches(&guard).count(); + let event_submit_count = event_submit_call_count(content); - if (event_sites.len(), guard_count) != expected { + if (event_sites.len(), guard_count, event_submit_count) != expected { violations.push(format!( - "{rel}: found {} events-URL site(s) + {} guard call(s), inventory \ - expects {} + {}. Sites found:\n{}", + "{rel}: found {} events-URL site(s) + {} guard call(s) + {} bounded-submit \ + call(s), inventory expects {} + {} + {}. Sites found:\n{}", event_sites.len(), guard_count, + event_submit_count, expected.0, expected.1, + expected.2, if event_sites.is_empty() { " (none)".to_string() } else { @@ -355,9 +372,9 @@ fn read_src_files() -> Vec<(String, String)> { /// Inventory completeness: every `/events` URL-construction site in /// `desktop/src-tauri/src` must match the site-granular inventory above. A -/// future ninth submission path — in a NEW file or an ALREADY-LISTED one — -/// fails this test until its guard is wired, its injection test exists, and -/// its inventory row is updated. +/// future ninth production submission path — in a NEW file or an +/// ALREADY-LISTED one — fails this test until its guard and bounded HTTP funnel +/// are wired, its injection test exists, and its inventory row is updated. #[test] fn events_url_inventory_is_fully_guarded() { let violations = events_inventory_violations(&read_src_files()); @@ -406,6 +423,30 @@ fn inventory_scan_catches_removed_guard_call() { ); } +/// The timeout boundary also fires in reverse: a production event path that +/// bypasses the bounded HTTP funnel while retaining its `/events` URL and egress +/// guard is caught. +#[test] +fn inventory_scan_catches_removed_bounded_submit_call() { + let mut files = read_src_files(); + let huddle = files + .iter_mut() + .find(|(rel, _)| rel.ends_with("src/huddle/pipeline.rs")) + .expect("huddle pipeline must be in the scan set"); + huddle.1 = huddle.1.replacen( + &["send_event_", "http_request("].concat(), + "unbounded_event_submit(", + 1, + ); + let violations = events_inventory_violations(&files); + assert!( + violations + .iter() + .any(|v| v.contains("src/huddle/pipeline.rs")), + "a removed bounded event-submit call must trip the scan: {violations:?}" + ); +} + /// A brand-new file with an `/events` site (no inventory row) is caught. #[test] fn inventory_scan_catches_new_unlisted_file() { diff --git a/desktop/src-tauri/src/huddle/pipeline.rs b/desktop/src-tauri/src/huddle/pipeline.rs index 47d4aeb43d1..075504a225c 100644 --- a/desktop/src-tauri/src/huddle/pipeline.rs +++ b/desktop/src-tauri/src/huddle/pipeline.rs @@ -709,15 +709,14 @@ pub(crate) fn spawn_transcription_task( } }; - let response = { - http_client - .post(&url) - .header("Authorization", auth_header) - .header("Content-Type", "application/json") - .body(body_bytes) - .send() - .await - }; + let response = crate::relay::send_event_http_request( + &http_client, + &url, + &auth_header, + None, + body_bytes, + ) + .await; match response { Ok(resp) if resp.status().is_success() => {} diff --git a/desktop/src-tauri/src/relay.rs b/desktop/src-tauri/src/relay.rs index f408ef2afda..e18b2a2467c 100644 --- a/desktop/src-tauri/src/relay.rs +++ b/desktop/src-tauri/src/relay.rs @@ -530,19 +530,8 @@ pub async fn sync_managed_agent_profile( let url = format!("{}/events", relay_http_base_url(relay_url)); let auth = build_nip98_auth_header_for_keys(agent_keys, &Method::POST, &url, &body_bytes)?; - let mut request = state - .http_client - .post(&url) - .header("Authorization", auth) - .header("Content-Type", "application/json"); - if let Some(tag) = auth_tag { - request = request.header("x-auth-tag", tag); - } - let response = request - .body(body_bytes) - .send() - .await - .map_err(|e| classify_request_error(&e))?; + let response = + send_event_http_request(&state.http_client, &url, &auth, auth_tag, body_bytes).await?; if !response.status().is_success() { let msg = relay_error_message(response).await; @@ -611,6 +600,7 @@ mod get; pub use get::get_relay_json; mod submit; +pub(crate) use submit::send_event_http_request; pub use submit::{ submit_event, submit_event_at_created_at, submit_event_at_with_keys, submit_event_with_keys_created_at, submit_signed_event_at_with_keys, SubmitEventResponse, @@ -649,20 +639,9 @@ pub async fn submit_signed_event_with_keys( crate::egress_guard::assert_no_key_backup_bytes(&body_bytes, "signed event submit (keys)")?; let auth_header = build_nip98_auth_header_for_keys(keys, &Method::POST, &url, &body_bytes)?; - let mut request = state - .http_client - .post(&url) - .header("Authorization", auth_header) - .header("Content-Type", "application/json"); - if let Some(tag) = auth_tag { - request = request.header("x-auth-tag", tag); - } - - let response = request - .body(body_bytes) - .send() - .await - .map_err(|e| classify_request_error(&e))?; + let response = + send_event_http_request(&state.http_client, &url, &auth_header, auth_tag, body_bytes) + .await?; if !response.status().is_success() { return Err(relay_error_message(response).await); diff --git a/desktop/src-tauri/src/relay/submit.rs b/desktop/src-tauri/src/relay/submit.rs index b6a5703fd96..7a87516d262 100644 --- a/desktop/src-tauri/src/relay/submit.rs +++ b/desktop/src-tauri/src/relay/submit.rs @@ -1,5 +1,73 @@ use super::*; +/// Per-request deadline for `POST /events`, scoped to event publication so the +/// shared client can remain unbounded for long-running model and media work. +const EVENT_SUBMIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + +/// Send one authenticated event request with a bounded header/body deadline. +/// +/// Every production `POST /events` path funnels through this helper. Callers +/// retain their existing status and response-body semantics; the request-level +/// deadline follows the returned response through body consumption. +pub(crate) async fn send_event_http_request( + http_client: &reqwest::Client, + url: &str, + auth_header: &str, + auth_tag: Option<&str>, + body_bytes: Vec, +) -> Result { + send_event_http_request_with_timeout( + http_client, + url, + auth_header, + auth_tag, + body_bytes, + EVENT_SUBMIT_TIMEOUT, + ) + .await +} + +async fn send_event_http_request_with_timeout( + http_client: &reqwest::Client, + url: &str, + auth_header: &str, + auth_tag: Option<&str>, + body_bytes: Vec, + timeout: std::time::Duration, +) -> Result { + let mut request = http_client + .post(url) + .header("Authorization", auth_header) + .header("Content-Type", "application/json") + .timeout(timeout); + if let Some(tag) = auth_tag { + request = request.header("x-auth-tag", tag); + } + request + .body(body_bytes) + .send() + .await + .map_err(|error| classify_request_error(&error)) +} + +#[cfg(test)] +pub(super) async fn send_event_http_request_for_test( + http_client: &reqwest::Client, + url: &str, + body_bytes: Vec, + timeout: std::time::Duration, +) -> Result { + send_event_http_request_with_timeout( + http_client, + url, + "Nostr test-auth", + None, + body_bytes, + timeout, + ) + .await +} + /// Response from `POST /events`. #[derive(Debug, Deserialize, serde::Serialize)] pub struct SubmitEventResponse { @@ -28,15 +96,8 @@ pub async fn submit_signed_event_at_with_keys( crate::egress_guard::assert_no_key_backup_bytes(&body_bytes, "relay event submit")?; let auth_header = build_nip98_auth_header_for_keys(keys, &Method::POST, &url, &body_bytes)?; - let response = state - .http_client - .post(&url) - .header("Authorization", auth_header) - .header("Content-Type", "application/json") - .body(body_bytes) - .send() - .await - .map_err(|e| classify_request_error(&e))?; + let response = + send_event_http_request(&state.http_client, &url, &auth_header, None, body_bytes).await?; if !response.status().is_success() { return Err(relay_error_message(response).await); diff --git a/desktop/src-tauri/src/relay/tests.rs b/desktop/src-tauri/src/relay/tests.rs index 4ae39249328..b6c20395c50 100644 --- a/desktop/src-tauri/src/relay/tests.rs +++ b/desktop/src-tauri/src/relay/tests.rs @@ -3,7 +3,8 @@ use super::{ build_profile_event, classify_intercepted_response, effective_agent_relay_url, - extract_retry_in_hint, parse_command_response, relay_http_base_url, MALFORMED_RESPONSE_MESSAGE, + extract_retry_in_hint, parse_command_response, relay_http_base_url, + submit::send_event_http_request_for_test, MALFORMED_RESPONSE_MESSAGE, }; use serde::Deserialize; @@ -601,6 +602,75 @@ fn profile_event_without_auth_tag() { assert_eq!(event.kind, nostr::Kind::Metadata); } +#[tokio::test] +async fn stalled_event_submit_times_out_with_classified_error() { + use std::io::Read as _; + use std::time::Duration; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let handle = std::thread::spawn(move || { + if let Ok((mut stream, _)) = listener.accept() { + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf); + std::thread::sleep(Duration::from_secs(2)); + } + }); + + let result = tokio::time::timeout( + Duration::from_secs(5), + send_event_http_request_for_test( + &reqwest::Client::new(), + &format!("http://{addr}/events"), + b"{}".to_vec(), + Duration::from_millis(200), + ), + ) + .await + .expect("the event submit helper must honor its per-request timeout and resolve within 5s"); + + assert_eq!( + result.expect_err("a stalled event submit must fail"), + "relay unreachable: request timed out" + ); + let _ = handle.join(); +} + +#[tokio::test] +async fn event_submit_timeout_covers_response_body() { + use std::io::{Read as _, Write as _}; + use std::time::Duration; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let handle = std::thread::spawn(move || { + if let Ok((mut stream, _)) = listener.accept() { + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf); + let _ = stream.write_all( + b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 64\r\n\r\n", + ); + let _ = stream.flush(); + std::thread::sleep(Duration::from_secs(2)); + } + }); + + let response = send_event_http_request_for_test( + &reqwest::Client::new(), + &format!("http://{addr}/events"), + b"{}".to_vec(), + Duration::from_millis(200), + ) + .await + .expect("response headers should arrive before the request deadline"); + let err = super::parse_json_response::(response) + .await + .expect_err("the stalled response body must time out"); + + assert_eq!(err, "relay unreachable: request timed out"); + let _ = handle.join(); +} + #[test] fn profile_event_rejects_invalid_auth_tag() { let agent_keys = nostr::Keys::generate(); From 859dabf59d81e8f4d7cf252125c83b7981361546 Mon Sep 17 00:00:00 2001 From: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz> Date: Thu, 27 Aug 2026 14:54:49 -0600 Subject: [PATCH 2/3] fix(desktop): classify snapshot body timeouts Signed-off-by: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz> --- .../src/commands/engram_submit_response.rs | 73 +++++++++++++++++++ desktop/src-tauri/src/commands/mod.rs | 1 + .../src/commands/personas/snapshot/import.rs | 6 +- .../src-tauri/src/commands/team_snapshot.rs | 6 +- desktop/src-tauri/src/egress_guard_tests.rs | 1 + desktop/src-tauri/src/relay.rs | 9 +++ 6 files changed, 88 insertions(+), 8 deletions(-) create mode 100644 desktop/src-tauri/src/commands/engram_submit_response.rs diff --git a/desktop/src-tauri/src/commands/engram_submit_response.rs b/desktop/src-tauri/src/commands/engram_submit_response.rs new file mode 100644 index 00000000000..ec9f5c115cc --- /dev/null +++ b/desktop/src-tauri/src/commands/engram_submit_response.rs @@ -0,0 +1,73 @@ +/// Apply the shared relay-body timeout classification while preserving the +/// snapshot paths' established detail for every non-timeout read failure. +pub(crate) async fn read_engram_submit_response( + response: reqwest::Response, +) -> Result { + response.text().await.map_err(|error| { + crate::relay::classify_body_read_error(&error, || { + format!("failed to read relay response: {error}") + }) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn body_stall_surfaces_classified_timeout() { + use std::io::{Read as _, Write as _}; + use std::time::Duration; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let handle = std::thread::spawn(move || { + if let Ok((mut stream, _)) = listener.accept() { + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf); + let _ = stream.write_all( + b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 64\r\n\r\n", + ); + let _ = stream.flush(); + std::thread::sleep(Duration::from_secs(1)); + } + }); + let response = reqwest::Client::new() + .get(format!("http://{addr}/events")) + .timeout(Duration::from_millis(200)) + .send() + .await + .expect("headers must arrive before the deadline"); + let err = read_engram_submit_response(response).await.unwrap_err(); + + assert_eq!(err, "relay unreachable: request timed out"); + let _ = handle.join(); + } + + #[tokio::test] + async fn incomplete_body_keeps_existing_non_timeout_prefix() { + use std::io::{Read as _, Write as _}; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let handle = std::thread::spawn(move || { + if let Ok((mut stream, _)) = listener.accept() { + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf); + let _ = stream.write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 64\r\nConnection: close\r\n\r\nshort", + ); + let _ = stream.flush(); + } + }); + let response = reqwest::Client::new() + .get(format!("http://{addr}/events")) + .send() + .await + .expect("headers must arrive"); + let err = read_engram_submit_response(response).await.unwrap_err(); + + assert!(err.starts_with("failed to read relay response: "), "{err}"); + let _ = handle.join(); + } +} diff --git a/desktop/src-tauri/src/commands/mod.rs b/desktop/src-tauri/src/commands/mod.rs index 7cb2d8e3b83..7f3982d51d4 100644 --- a/desktop/src-tauri/src/commands/mod.rs +++ b/desktop/src-tauri/src/commands/mod.rs @@ -18,6 +18,7 @@ mod channel_window; mod channels; mod clipboard; mod dms; +mod engram_submit_response; mod engrams; mod export_util; mod global_agent_config; diff --git a/desktop/src-tauri/src/commands/personas/snapshot/import.rs b/desktop/src-tauri/src/commands/personas/snapshot/import.rs index 8f27f1411f5..1ca13c40270 100644 --- a/desktop/src-tauri/src/commands/personas/snapshot/import.rs +++ b/desktop/src-tauri/src/commands/personas/snapshot/import.rs @@ -833,10 +833,8 @@ pub(crate) async fn submit_engram_event( return Err(format!("relay rejected engram: {msg}")); } - let body = response - .text() - .await - .map_err(|e| format!("failed to read relay response: {e}"))?; + let body = + crate::commands::engram_submit_response::read_engram_submit_response(response).await?; let parsed: serde_json::Value = serde_json::from_str(&body).map_err(|e| format!("relay response not JSON: {e}"))?; let accepted = parsed diff --git a/desktop/src-tauri/src/commands/team_snapshot.rs b/desktop/src-tauri/src/commands/team_snapshot.rs index 0182bd2ea3c..a75d773c132 100644 --- a/desktop/src-tauri/src/commands/team_snapshot.rs +++ b/desktop/src-tauri/src/commands/team_snapshot.rs @@ -928,10 +928,8 @@ pub(crate) async fn submit_engram_event( return Err(format!("relay rejected engram: {msg}")); } - let body = response - .text() - .await - .map_err(|e| format!("failed to read relay response: {e}"))?; + let body = + crate::commands::engram_submit_response::read_engram_submit_response(response).await?; let parsed: serde_json::Value = serde_json::from_str(&body).map_err(|e| format!("relay response not JSON: {e}"))?; let accepted = parsed diff --git a/desktop/src-tauri/src/egress_guard_tests.rs b/desktop/src-tauri/src/egress_guard_tests.rs index 8f3e12876f7..def374d604d 100644 --- a/desktop/src-tauri/src/egress_guard_tests.rs +++ b/desktop/src-tauri/src/egress_guard_tests.rs @@ -272,6 +272,7 @@ const EVENTS_INVENTORY: &[(&str, usize, usize, usize)] = &[ ("src/native_websocket.rs", 0, 2, 0), // boundary 8 (WS frames; no events URL) // Test-only fixtures — no production egress, no guard or production submit call: ("src/relay/tests.rs", 2, 0, 0), + ("src/commands/engram_submit_response.rs", 2, 0, 0), ("src/relay_admission.rs", 1, 0, 0), ("src/archive/mod_tests.rs", 1, 0, 0), ("src/managed_agents/persona_events/tests.rs", 1, 0, 0), diff --git a/desktop/src-tauri/src/relay.rs b/desktop/src-tauri/src/relay.rs index e18b2a2467c..a626a3c6dab 100644 --- a/desktop/src-tauri/src/relay.rs +++ b/desktop/src-tauri/src/relay.rs @@ -196,6 +196,15 @@ fn classify_body_timeout(e: &reqwest::Error) -> Option { e.is_timeout().then(|| classify_request_error(e)) } +/// Preserve timeout bodies as connectivity failures while allowing each caller +/// to retain its established non-timeout error behavior. +pub(crate) fn classify_body_read_error( + e: &reqwest::Error, + fallback: impl FnOnce() -> String, +) -> String { + classify_body_timeout(e).unwrap_or_else(fallback) +} + /// Detect responses that were intercepted by a captive portal or auth proxy. /// /// Returns `Some(msg)` when the response clearly did not come from the relay: From 34591d687467cbb31f5a71847f9c95d3f2cc3b75 Mon Sep 17 00:00:00 2001 From: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz> Date: Thu, 27 Aug 2026 15:50:49 -0600 Subject: [PATCH 3/3] fix(desktop): retry ambiguous event submissions Signed-off-by: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz> --- desktop/src-tauri/src/commands/mod.rs | 2 +- .../src/commands/personas/snapshot/import.rs | 27 +- .../src-tauri/src/commands/team_snapshot.rs | 27 +- desktop/src-tauri/src/egress_guard_tests.rs | 23 +- desktop/src-tauri/src/huddle/pipeline.rs | 25 +- desktop/src-tauri/src/relay.rs | 35 +- desktop/src-tauri/src/relay/submit.rs | 356 ++++++++++++++++-- desktop/src-tauri/src/relay/tests.rs | 282 +++++++++++++- 8 files changed, 666 insertions(+), 111 deletions(-) diff --git a/desktop/src-tauri/src/commands/mod.rs b/desktop/src-tauri/src/commands/mod.rs index 7f3982d51d4..c98035d255c 100644 --- a/desktop/src-tauri/src/commands/mod.rs +++ b/desktop/src-tauri/src/commands/mod.rs @@ -18,7 +18,7 @@ mod channel_window; mod channels; mod clipboard; mod dms; -mod engram_submit_response; +pub(crate) mod engram_submit_response; mod engrams; mod export_util; mod global_agent_config; diff --git a/desktop/src-tauri/src/commands/personas/snapshot/import.rs b/desktop/src-tauri/src/commands/personas/snapshot/import.rs index 1ca13c40270..02b48042fcd 100644 --- a/desktop/src-tauri/src/commands/personas/snapshot/import.rs +++ b/desktop/src-tauri/src/commands/personas/snapshot/import.rs @@ -809,32 +809,29 @@ pub(crate) async fn submit_engram_event( url: &str, auth_tag: Option<&str>, ) -> Result<(), String> { - use crate::relay::build_nip98_auth_header_for_keys; - use reqwest::Method; - crate::egress_guard::assert_no_key_backup_bytes(event_json, "persona snapshot engram submit")?; // Wait before signing: the relay enforces NIP-98 freshness (±60s) and the // gate may hold for up to MAX_HINT_SECONDS (300s). Building auth before the // wait produces a stale `created_at` that the relay will reject. crate::relay_admission::wait_for_rate_limit().await; - let auth = build_nip98_auth_header_for_keys(agent_keys, &Method::POST, url, event_json)?; - let response = crate::relay::send_event_http_request( + let body = match crate::relay::submit_event_text_with_keys( &state.http_client, url, - &auth, + agent_keys, auth_tag, - event_json.to_vec(), + event_json, ) - .await?; - - if !response.status().is_success() { - let msg = crate::relay::relay_error_message(response).await; - return Err(format!("relay rejected engram: {msg}")); - } + .await + { + Ok(body) => body, + Err(crate::relay::EventSubmitHttpError::Auth(error)) => return Err(error), + Err(crate::relay::EventSubmitHttpError::Rejected(error)) => { + return Err(format!("relay rejected engram: {error}")); + } + Err(error) => return Err(error.into_message()), + }; - let body = - crate::commands::engram_submit_response::read_engram_submit_response(response).await?; let parsed: serde_json::Value = serde_json::from_str(&body).map_err(|e| format!("relay response not JSON: {e}"))?; let accepted = parsed diff --git a/desktop/src-tauri/src/commands/team_snapshot.rs b/desktop/src-tauri/src/commands/team_snapshot.rs index a75d773c132..32f3bcc5ae8 100644 --- a/desktop/src-tauri/src/commands/team_snapshot.rs +++ b/desktop/src-tauri/src/commands/team_snapshot.rs @@ -904,32 +904,29 @@ pub(crate) async fn submit_engram_event( url: &str, auth_tag: Option<&str>, ) -> Result<(), String> { - use crate::relay::build_nip98_auth_header_for_keys; - use reqwest::Method; - crate::egress_guard::assert_no_key_backup_bytes(event_json, "team snapshot engram submit")?; // Wait before signing: the relay enforces NIP-98 freshness (±60s) and the // gate may hold for up to MAX_HINT_SECONDS (300s). Building auth before the // wait produces a stale `created_at` that the relay will reject. crate::relay_admission::wait_for_rate_limit().await; - let auth = build_nip98_auth_header_for_keys(agent_keys, &Method::POST, url, event_json)?; - let response = crate::relay::send_event_http_request( + let body = match crate::relay::submit_event_text_with_keys( &state.http_client, url, - &auth, + agent_keys, auth_tag, - event_json.to_vec(), + event_json, ) - .await?; - - if !response.status().is_success() { - let msg = crate::relay::relay_error_message(response).await; - return Err(format!("relay rejected engram: {msg}")); - } + .await + { + Ok(body) => body, + Err(crate::relay::EventSubmitHttpError::Auth(error)) => return Err(error), + Err(crate::relay::EventSubmitHttpError::Rejected(error)) => { + return Err(format!("relay rejected engram: {error}")); + } + Err(error) => return Err(error.into_message()), + }; - let body = - crate::commands::engram_submit_response::read_engram_submit_response(response).await?; let parsed: serde_json::Value = serde_json::from_str(&body).map_err(|e| format!("relay response not JSON: {e}"))?; let accepted = parsed diff --git a/desktop/src-tauri/src/egress_guard_tests.rs b/desktop/src-tauri/src/egress_guard_tests.rs index def374d604d..1e1ff8455c6 100644 --- a/desktop/src-tauri/src/egress_guard_tests.rs +++ b/desktop/src-tauri/src/egress_guard_tests.rs @@ -250,7 +250,7 @@ fn src_rust_files() -> Vec { /// Site-granular `/events` inventory: `(file suffix, expected non-comment /// `/events` occurrences, expected guard call sites, expected calls into the -/// bounded HTTP event-submit funnel)`. +/// bounded, idempotent HTTP event-submit funnel)`. /// /// Every production entry pairs the URL-construction count with both safety /// boundaries for that file, so all of these fail the scan (not just a @@ -258,7 +258,7 @@ fn src_rust_files() -> Vec { /// - adding an unguarded ninth `/events` site inside an already-listed file /// (count goes up without a matching table update), /// - removing/refactoring away a guard call while its egress site remains, -/// - bypassing `send_event_http_request` and its per-request deadline. +/// - bypassing the retrying event-submit funnel and its per-request deadline. /// /// Updating a row here is the deliberate act that must accompany wiring both /// boundaries and adding an injection test for a new production site. @@ -271,7 +271,7 @@ const EVENTS_INVENTORY: &[(&str, usize, usize, usize)] = &[ ("src/commands/personas/snapshot/import.rs", 2, 1, 1), // boundary 7 + its in-file injection-test fixture URL ("src/native_websocket.rs", 0, 2, 0), // boundary 8 (WS frames; no events URL) // Test-only fixtures — no production egress, no guard or production submit call: - ("src/relay/tests.rs", 2, 0, 0), + ("src/relay/tests.rs", 5, 0, 0), ("src/commands/engram_submit_response.rs", 2, 0, 0), ("src/relay_admission.rs", 1, 0, 0), ("src/archive/mod_tests.rs", 1, 0, 0), @@ -295,13 +295,18 @@ fn guard_needle() -> String { ["egress_guard::", "assert_no_key_backup"].concat() } fn event_submit_call_count(content: &str) -> usize { - let needle = &["send_event_", "http_request("].concat(); + let needles = [ + ["send_event_", "http_request_with_keys("].concat(), + ["submit_event_", "json_with_keys("].concat(), + ["submit_event_", "text_with_keys("].concat(), + ]; content .lines() .filter(|line| { - line.contains(needle) + needles.iter().any(|needle| line.contains(needle)) && !line.trim_start().starts_with("//") && !line.contains("fn send_event_") + && !line.contains("fn submit_event_") }) .count() } @@ -424,9 +429,9 @@ fn inventory_scan_catches_removed_guard_call() { ); } -/// The timeout boundary also fires in reverse: a production event path that -/// bypasses the bounded HTTP funnel while retaining its `/events` URL and egress -/// guard is caught. +/// The timeout/retry boundary also fires in reverse: a production event path +/// that bypasses the bounded idempotent funnel while retaining its `/events` +/// URL and egress guard is caught. #[test] fn inventory_scan_catches_removed_bounded_submit_call() { let mut files = read_src_files(); @@ -435,7 +440,7 @@ fn inventory_scan_catches_removed_bounded_submit_call() { .find(|(rel, _)| rel.ends_with("src/huddle/pipeline.rs")) .expect("huddle pipeline must be in the scan set"); huddle.1 = huddle.1.replacen( - &["send_event_", "http_request("].concat(), + &["send_event_", "http_request_with_keys("].concat(), "unbounded_event_submit(", 1, ); diff --git a/desktop/src-tauri/src/huddle/pipeline.rs b/desktop/src-tauri/src/huddle/pipeline.rs index 075504a225c..e9cc99354d4 100644 --- a/desktop/src-tauri/src/huddle/pipeline.rs +++ b/desktop/src-tauri/src/huddle/pipeline.rs @@ -696,25 +696,12 @@ pub(crate) fn spawn_transcription_task( } }; let url = format!("{relay_base_url}/events"); - let auth_header = match crate::relay::build_nip98_auth_header_for_keys( - &keys, - &reqwest::Method::POST, - &url, - &body_bytes, - ) { - Ok(h) => h, - Err(e) => { - eprintln!("buzz-desktop: STT NIP-98 auth: {e}"); - continue; - } - }; - - let response = crate::relay::send_event_http_request( + let response = crate::relay::send_event_http_request_with_keys( &http_client, &url, - &auth_header, + &keys, None, - body_bytes, + &body_bytes, ) .await; @@ -726,8 +713,12 @@ pub(crate) fn spawn_transcription_task( let msg = crate::relay::relay_error_message(resp).await; eprintln!("buzz-desktop: STT kind:9 post failed: {msg}"); } + Err(crate::relay::EventSubmitHttpError::Auth(e)) => { + eprintln!("buzz-desktop: STT NIP-98 auth: {e}"); + continue; + } Err(e) => { - eprintln!("buzz-desktop: STT kind:9 post failed: {e}"); + eprintln!("buzz-desktop: STT kind:9 post failed: {}", e.into_message()); } } } diff --git a/desktop/src-tauri/src/relay.rs b/desktop/src-tauri/src/relay.rs index a626a3c6dab..320cc8c9dd3 100644 --- a/desktop/src-tauri/src/relay.rs +++ b/desktop/src-tauri/src/relay.rs @@ -537,10 +537,19 @@ pub async fn sync_managed_agent_profile( crate::egress_guard::assert_no_key_backup_bytes(&body_bytes, "agent profile sync")?; let url = format!("{}/events", relay_http_base_url(relay_url)); - let auth = build_nip98_auth_header_for_keys(agent_keys, &Method::POST, &url, &body_bytes)?; - - let response = - send_event_http_request(&state.http_client, &url, &auth, auth_tag, body_bytes).await?; + let response = match send_event_http_request_with_keys( + &state.http_client, + &url, + agent_keys, + auth_tag, + &body_bytes, + ) + .await + { + Ok(response) => response, + Err(EventSubmitHttpError::Auth(error)) => return Err(error), + Err(error) => return Err(error.into_message()), + }; if !response.status().is_success() { let msg = relay_error_message(response).await; @@ -609,7 +618,10 @@ mod get; pub use get::get_relay_json; mod submit; -pub(crate) use submit::send_event_http_request; +pub(crate) use submit::{ + send_event_http_request_with_keys, submit_event_json_with_keys, submit_event_text_with_keys, + EventSubmitHttpError, +}; pub use submit::{ submit_event, submit_event_at_created_at, submit_event_at_with_keys, submit_event_with_keys_created_at, submit_signed_event_at_with_keys, SubmitEventResponse, @@ -646,17 +658,8 @@ pub async fn submit_signed_event_with_keys( let url = format!("{}/events", relay_api_base_url_with_override(state)); let body_bytes = event.as_json().into_bytes(); crate::egress_guard::assert_no_key_backup_bytes(&body_bytes, "signed event submit (keys)")?; - let auth_header = build_nip98_auth_header_for_keys(keys, &Method::POST, &url, &body_bytes)?; - - let response = - send_event_http_request(&state.http_client, &url, &auth_header, auth_tag, body_bytes) - .await?; - - if !response.status().is_success() { - return Err(relay_error_message(response).await); - } - - let result: SubmitEventResponse = parse_json_response(response).await?; + let result: SubmitEventResponse = + submit_event_json_with_keys(&state.http_client, &url, keys, auth_tag, &body_bytes).await?; if !result.accepted { return Err(format!("relay rejected event: {}", result.message)); diff --git a/desktop/src-tauri/src/relay/submit.rs b/desktop/src-tauri/src/relay/submit.rs index 7a87516d262..9cb718bd279 100644 --- a/desktop/src-tauri/src/relay/submit.rs +++ b/desktop/src-tauri/src/relay/submit.rs @@ -1,25 +1,57 @@ use super::*; -/// Per-request deadline for `POST /events`, scoped to event publication so the -/// shared client can remain unbounded for long-running model and media work. +/// Every attempt gets a 30-second request deadline; an ambiguous timeout may +/// therefore take up to 60 seconds before failure. The shared client remains +/// unbounded for long-running model and media work. const EVENT_SUBMIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); -/// Send one authenticated event request with a bounded header/body deadline. -/// -/// Every production `POST /events` path funnels through this helper. Callers -/// retain their existing status and response-body semantics; the request-level -/// deadline follows the returned response through body consumption. -pub(crate) async fn send_event_http_request( +async fn send_event_http_request_once( http_client: &reqwest::Client, url: &str, auth_header: &str, auth_tag: Option<&str>, - body_bytes: Vec, + body_bytes: &[u8], + timeout: std::time::Duration, ) -> Result { - send_event_http_request_with_timeout( + let mut request = http_client + .post(url) + .header("Authorization", auth_header) + .header("Content-Type", "application/json") + .timeout(timeout); + if let Some(tag) = auth_tag { + request = request.header("x-auth-tag", tag); + } + request + .body(body_bytes.to_vec()) + .send() + .await + .map_err(|error| classify_request_error(&error)) +} + +const EVENT_SUBMIT_MAX_ATTEMPTS: usize = 2; + +fn is_event_submit_timeout(error: &str) -> bool { + error == "relay unreachable: request timed out" +} + +/// Send an already-signed event, retrying the identical bytes once if the +/// request times out before response headers arrive. +/// +/// This headers-only variant preserves callers that intentionally treat any +/// successful HTTP status as completion. NIP-98 auth is rebuilt for each +/// attempt because each auth event is single-use. A non-success status body is +/// deliberately left to the caller so rate-limit handling stays unchanged. +pub(crate) async fn send_event_http_request_with_keys( + http_client: &reqwest::Client, + url: &str, + keys: &Keys, + auth_tag: Option<&str>, + body_bytes: &[u8], +) -> Result { + send_event_http_request_with_keys_and_timeout( http_client, url, - auth_header, + keys, auth_tag, body_bytes, EVENT_SUBMIT_TIMEOUT, @@ -27,27 +59,227 @@ pub(crate) async fn send_event_http_request( .await } -async fn send_event_http_request_with_timeout( +/// Fully consume one event-submit response, retrying the exact signed event on +/// an ambiguous timeout. Keeping this generic lets JSON and legacy text callers +/// share the retry boundary while preserving their established parse errors. +async fn consume_event_http_response_with_keys( http_client: &reqwest::Client, url: &str, - auth_header: &str, + keys: &Keys, auth_tag: Option<&str>, - body_bytes: Vec, + body_bytes: &[u8], + consume_response: Consume, + classify_response_error: ClassifyResponse, +) -> Result +where + Consume: FnMut(reqwest::Response) -> ResponseFuture, + ResponseFuture: std::future::Future>, + ClassifyResponse: Fn(String) -> EventSubmitHttpError, +{ + submit_event_response_with_keys_and_timeout( + http_client, + url, + keys, + auth_tag, + body_bytes, + EVENT_SUBMIT_TIMEOUT, + consume_response, + classify_response_error, + ) + .await +} + +async fn send_event_http_request_with_keys_and_timeout( + http_client: &reqwest::Client, + url: &str, + keys: &Keys, + auth_tag: Option<&str>, + body_bytes: &[u8], timeout: std::time::Duration, -) -> Result { - let mut request = http_client - .post(url) - .header("Authorization", auth_header) - .header("Content-Type", "application/json") - .timeout(timeout); - if let Some(tag) = auth_tag { - request = request.header("x-auth-tag", tag); +) -> Result { + let mut last_timeout = None; + + for _ in 0..EVENT_SUBMIT_MAX_ATTEMPTS { + let auth_header = build_nip98_auth_header_for_keys(keys, &Method::POST, url, body_bytes) + .map_err(EventSubmitHttpError::Auth)?; + match send_event_http_request_once( + http_client, + url, + &auth_header, + auth_tag, + body_bytes, + timeout, + ) + .await + { + Ok(response) => return Ok(response), + Err(error) if is_event_submit_timeout(&error) => { + last_timeout = Some(EventSubmitHttpError::Request(error)); + } + Err(error) => return Err(EventSubmitHttpError::Request(error)), + } } - request - .body(body_bytes) - .send() + + Err(last_timeout.unwrap_or_else(|| { + EventSubmitHttpError::Request("relay unreachable: request timed out".to_string()) + })) +} + +/// Which phase of a fully-consumed event submission failed. +/// +/// Snapshot callers retain distinct prefixes for relay rejections while +/// leaving transport and successful-response decode failures unwrapped. +#[derive(Debug)] +pub(crate) enum EventSubmitHttpError { + Auth(String), + Request(String), + Rejected(String), + Response(String), +} + +impl EventSubmitHttpError { + fn is_timeout(&self) -> bool { + is_event_submit_timeout(self.message()) + } + + fn message(&self) -> &str { + match self { + Self::Auth(message) + | Self::Request(message) + | Self::Rejected(message) + | Self::Response(message) => message, + } + } + + pub(crate) fn into_message(self) -> String { + match self { + Self::Auth(message) + | Self::Request(message) + | Self::Rejected(message) + | Self::Response(message) => message, + } + } +} + +/// Submit one already-signed event and fully consume the relay response. +/// +/// A timeout is ambiguous because relay ingest may complete before either the +/// headers or body reach Desktop. Rebuild NIP-98 request auth and retry the +/// exact same serialized event once. Ordinary event ingest treats a repeated +/// event ID as accepted without another insert or dispatch, so this reconciles +/// an accepted-first-attempt/body-stall without re-signing the user action +/// under a second ID. +async fn submit_event_response_with_keys_and_timeout( + http_client: &reqwest::Client, + url: &str, + keys: &Keys, + auth_tag: Option<&str>, + body_bytes: &[u8], + timeout: std::time::Duration, + mut consume_response: Consume, + classify_response_error: ClassifyResponse, +) -> Result +where + Consume: FnMut(reqwest::Response) -> ResponseFuture, + ResponseFuture: std::future::Future>, + ClassifyResponse: Fn(String) -> EventSubmitHttpError, +{ + let mut last_timeout = None; + + for _ in 0..EVENT_SUBMIT_MAX_ATTEMPTS { + let auth_header = build_nip98_auth_header_for_keys(keys, &Method::POST, url, body_bytes) + .map_err(EventSubmitHttpError::Auth)?; + let response = match send_event_http_request_once( + http_client, + url, + &auth_header, + auth_tag, + body_bytes, + timeout, + ) .await - .map_err(|error| classify_request_error(&error)) + { + Ok(response) => response, + Err(error) => { + let error = EventSubmitHttpError::Request(error); + if error.is_timeout() { + last_timeout = Some(error); + continue; + } + return Err(error); + } + }; + + if !response.status().is_success() { + let error = EventSubmitHttpError::Rejected(relay_error_message(response).await); + if error.is_timeout() { + last_timeout = Some(error); + continue; + } + return Err(error); + } + + match consume_response(response).await { + Ok(result) => return Ok(result), + Err(error) => { + let error = classify_response_error(error); + if error.is_timeout() { + last_timeout = Some(error); + continue; + } + return Err(error); + } + } + } + + Err(last_timeout.unwrap_or_else(|| { + EventSubmitHttpError::Request("relay unreachable: request timed out".to_string()) + })) +} + +pub(crate) async fn submit_event_json_with_keys( + http_client: &reqwest::Client, + url: &str, + keys: &Keys, + auth_tag: Option<&str>, + body_bytes: &[u8], +) -> Result { + consume_event_http_response_with_keys( + http_client, + url, + keys, + auth_tag, + body_bytes, + parse_json_response::, + EventSubmitHttpError::Response, + ) + .await + .map_err(EventSubmitHttpError::into_message) +} + +/// Submit an event and return its successful response body as text. +/// +/// This variant preserves the snapshot import paths' established non-timeout +/// body-read detail while keeping the body inside the idempotent retry boundary. +pub(crate) async fn submit_event_text_with_keys( + http_client: &reqwest::Client, + url: &str, + keys: &Keys, + auth_tag: Option<&str>, + body_bytes: &[u8], +) -> Result { + consume_event_http_response_with_keys( + http_client, + url, + keys, + auth_tag, + body_bytes, + |response| async move { + crate::commands::engram_submit_response::read_engram_submit_response(response).await + }, + EventSubmitHttpError::Response, + ) + .await } #[cfg(test)] @@ -57,13 +289,71 @@ pub(super) async fn send_event_http_request_for_test( body_bytes: Vec, timeout: std::time::Duration, ) -> Result { - send_event_http_request_with_timeout( + send_event_http_request_once( http_client, url, "Nostr test-auth", None, + &body_bytes, + timeout, + ) + .await +} + +#[cfg(test)] +pub(super) async fn send_event_http_request_with_keys_for_test( + http_client: &reqwest::Client, + url: &str, + keys: &Keys, + body_bytes: &[u8], + timeout: std::time::Duration, +) -> Result { + send_event_http_request_with_keys_and_timeout(http_client, url, keys, None, body_bytes, timeout) + .await + .map_err(EventSubmitHttpError::into_message) +} + +#[cfg(test)] +pub(super) async fn submit_event_json_with_keys_for_test( + http_client: &reqwest::Client, + url: &str, + keys: &Keys, + body_bytes: &[u8], + timeout: std::time::Duration, +) -> Result { + submit_event_response_with_keys_and_timeout( + http_client, + url, + keys, + None, body_bytes, timeout, + parse_json_response::, + EventSubmitHttpError::Response, + ) + .await + .map_err(EventSubmitHttpError::into_message) +} + +#[cfg(test)] +pub(super) async fn submit_event_text_with_keys_for_test( + http_client: &reqwest::Client, + url: &str, + keys: &Keys, + body_bytes: &[u8], + timeout: std::time::Duration, +) -> Result { + submit_event_response_with_keys_and_timeout( + http_client, + url, + keys, + None, + body_bytes, + timeout, + |response| async move { + crate::commands::engram_submit_response::read_engram_submit_response(response).await + }, + EventSubmitHttpError::Response, ) .await } @@ -94,16 +384,8 @@ pub async fn submit_signed_event_at_with_keys( let url = format!("{}/events", api_base_url.trim_end_matches('/')); let body_bytes = event.as_json().into_bytes(); crate::egress_guard::assert_no_key_backup_bytes(&body_bytes, "relay event submit")?; - let auth_header = build_nip98_auth_header_for_keys(keys, &Method::POST, &url, &body_bytes)?; - - let response = - send_event_http_request(&state.http_client, &url, &auth_header, None, body_bytes).await?; - - if !response.status().is_success() { - return Err(relay_error_message(response).await); - } - - let result: SubmitEventResponse = parse_json_response(response).await?; + let result: SubmitEventResponse = + submit_event_json_with_keys(&state.http_client, &url, keys, None, &body_bytes).await?; if !result.accepted { return Err(format!("relay rejected event: {}", result.message)); } diff --git a/desktop/src-tauri/src/relay/tests.rs b/desktop/src-tauri/src/relay/tests.rs index b6c20395c50..9ae70412d30 100644 --- a/desktop/src-tauri/src/relay/tests.rs +++ b/desktop/src-tauri/src/relay/tests.rs @@ -4,8 +4,14 @@ use super::{ build_profile_event, classify_intercepted_response, effective_agent_relay_url, extract_retry_in_hint, parse_command_response, relay_http_base_url, - submit::send_event_http_request_for_test, MALFORMED_RESPONSE_MESSAGE, + submit::{ + send_event_http_request_for_test, send_event_http_request_with_keys_for_test, + submit_event_json_with_keys_for_test, submit_event_text_with_keys_for_test, + EventSubmitHttpError, SubmitEventResponse, + }, + MALFORMED_RESPONSE_MESSAGE, }; +use nostr::JsonUtil as _; use serde::Deserialize; // ── extract_retry_in_hint ──────────────────────────────────────────────── @@ -636,6 +642,51 @@ async fn stalled_event_submit_times_out_with_classified_error() { let _ = handle.join(); } +#[tokio::test] +async fn pre_header_timeout_retries_same_event_once() { + use std::io::Write as _; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + use std::time::Duration; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let attempts = Arc::new(AtomicUsize::new(0)); + let server_attempts = Arc::clone(&attempts); + let handle = std::thread::spawn(move || { + let (first, _) = listener.accept().expect("first submit"); + server_attempts.fetch_add(1, Ordering::SeqCst); + std::thread::sleep(Duration::from_millis(300)); + drop(first); + + let (mut second, _) = listener.accept().expect("retry submit"); + server_attempts.fetch_add(1, Ordering::SeqCst); + second + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .unwrap(); + second.flush().unwrap(); + }); + + let keys = nostr::Keys::generate(); + let result = tokio::time::timeout( + Duration::from_secs(5), + send_event_http_request_with_keys_for_test( + &reqwest::Client::new(), + &format!("http://{addr}/events"), + &keys, + b"{}", + Duration::from_millis(200), + ), + ) + .await + .expect("headers-only submit must resolve within 5s") + .expect("the retry receives a success response"); + + assert!(result.status().is_success()); + handle.join().unwrap(); + assert_eq!(attempts.load(Ordering::SeqCst), 2); +} + #[tokio::test] async fn event_submit_timeout_covers_response_body() { use std::io::{Read as _, Write as _}; @@ -671,6 +722,235 @@ async fn event_submit_timeout_covers_response_body() { let _ = handle.join(); } +#[tokio::test] +async fn accepted_event_body_timeout_retries_same_event_with_fresh_auth() { + use base64::Engine as _; + use std::io::{Read as _, Write as _}; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + fn read_request(stream: &mut std::net::TcpStream) -> (Vec, String) { + let mut request = Vec::new(); + let mut buf = [0u8; 4096]; + let header_end = loop { + let read = stream.read(&mut buf).expect("read request"); + assert!(read > 0, "request closed before headers completed"); + request.extend_from_slice(&buf[..read]); + if let Some(index) = request.windows(4).position(|window| window == b"\r\n\r\n") { + break index + 4; + } + }; + let headers = std::str::from_utf8(&request[..header_end]) + .expect("request headers are UTF-8") + .to_string(); + let content_length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().expect("valid content length")) + }) + .expect("content length header"); + while request.len() < header_end + content_length { + let read = stream.read(&mut buf).expect("read request body"); + assert!(read > 0, "request closed before body completed"); + request.extend_from_slice(&buf[..read]); + } + let auth = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("authorization") + .then(|| value.trim().to_string()) + }) + .expect("authorization header"); + ( + request[header_end..header_end + content_length].to_vec(), + auth, + ) + } + + fn auth_nonce(auth: &str) -> String { + let encoded = auth.strip_prefix("Nostr ").expect("Nostr auth scheme"); + let bytes = base64::engine::general_purpose::STANDARD + .decode(encoded) + .expect("base64 NIP-98 event"); + let event: nostr::Event = serde_json::from_slice(&bytes).expect("NIP-98 event JSON"); + event + .tags + .iter() + .find_map(|tag| { + let values = tag.as_slice(); + (values.first().map(String::as_str) == Some("nonce")) + .then(|| values.get(1).cloned()) + .flatten() + }) + .expect("NIP-98 nonce tag") + } + + let keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "one user action") + .sign_with_keys(&keys) + .unwrap(); + let event_id = event.id.to_hex(); + let body = event.as_json().into_bytes(); + let attempts = Arc::new(Mutex::new(Vec::<(Vec, String)>::new())); + let durable_event_ids = Arc::new(Mutex::new(std::collections::HashSet::::new())); + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let server_attempts = Arc::clone(&attempts); + let server_durable_event_ids = Arc::clone(&durable_event_ids); + let server_event_id = event_id.clone(); + let handle = std::thread::spawn(move || { + let (mut first, _) = listener.accept().expect("first submit"); + let first_request = read_request(&mut first); + let first_event: nostr::Event = + serde_json::from_slice(&first_request.0).expect("first signed event"); + assert!( + server_durable_event_ids + .lock() + .unwrap() + .insert(first_event.id.to_hex()), + "the first attempt models the relay commit" + ); + server_attempts.lock().unwrap().push(first_request); + first + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 128\r\n\r\n", + ) + .unwrap(); + first.flush().unwrap(); + + let (mut second, _) = listener.accept().expect("retry submit"); + let second_request = read_request(&mut second); + let second_event: nostr::Event = + serde_json::from_slice(&second_request.0).expect("retried signed event"); + assert!( + !server_durable_event_ids + .lock() + .unwrap() + .insert(second_event.id.to_hex()), + "the retry models the relay's duplicate-ID no-op" + ); + server_attempts.lock().unwrap().push(second_request); + let response_body = serde_json::json!({ + "event_id": server_event_id, + "accepted": true, + "message": "duplicate:" + }) + .to_string(); + write!( + second, + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + response_body.len(), + response_body + ) + .unwrap(); + second.flush().unwrap(); + }); + + let result: SubmitEventResponse = tokio::time::timeout( + Duration::from_secs(5), + submit_event_json_with_keys_for_test( + &reqwest::Client::new(), + &format!("http://{addr}/events"), + &keys, + &body, + Duration::from_millis(200), + ), + ) + .await + .expect("the retrying submit helper must resolve within 5s") + .expect("the second identical submission reconciles the ambiguous timeout"); + + handle.join().unwrap(); + let attempts = attempts.lock().unwrap(); + assert_eq!(attempts.len(), 2, "one timeout gets exactly one retry"); + assert_eq!(attempts[0].0, body, "first attempt uses the signed event"); + assert_eq!(attempts[1].0, body, "retry preserves the exact event bytes"); + assert_eq!(result.event_id, event_id, "one durable event ID"); + assert_eq!( + durable_event_ids.lock().unwrap().len(), + 1, + "accepted attempt plus retry remains one durable event" + ); + assert!(result.accepted); + assert_eq!(result.message, "duplicate:"); + assert_ne!(attempts[0].1, attempts[1].1, "NIP-98 auth must be fresh"); + assert_ne!( + auth_nonce(&attempts[0].1), + auth_nonce(&attempts[1].1), + "the relay replay guard must see distinct auth nonces" + ); +} + +#[tokio::test] +async fn snapshot_event_submit_preserves_error_prefix_contracts() { + use std::io::{Read as _, Write as _}; + use std::time::Duration; + + let keys = nostr::Keys::generate(); + let body = nostr::EventBuilder::new(nostr::Kind::Custom(9), "engram") + .sign_with_keys(&keys) + .unwrap() + .as_json() + .into_bytes(); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let handle = std::thread::spawn(move || { + for fixture in [ + "HTTP/1.1 400 Bad Request\r\nContent-Type: application/json\r\nContent-Length: 18\r\nConnection: close\r\n\r\n{\"message\":\"nope\"}", + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 64\r\nConnection: close\r\n\r\nshort", + ] { + let (mut stream, _) = listener.accept().expect("snapshot submit"); + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf); + stream.write_all(fixture.as_bytes()).unwrap(); + stream.flush().unwrap(); + } + }); + let url = format!("http://{addr}/events"); + + let rejected = submit_event_text_with_keys_for_test( + &reqwest::Client::new(), + &url, + &keys, + &body, + Duration::from_secs(1), + ) + .await + .expect_err("400 is a relay rejection"); + assert!( + matches!( + rejected, + EventSubmitHttpError::Rejected(ref error) + if error == "relay returned 400 Bad Request: nope" + ), + "{rejected:?}" + ); + + let body_error = submit_event_text_with_keys_for_test( + &reqwest::Client::new(), + &url, + &keys, + &body, + Duration::from_secs(1), + ) + .await + .expect_err("an incomplete 2xx body is a response read failure"); + assert!( + matches!( + body_error, + EventSubmitHttpError::Response(ref error) + if error.starts_with("failed to read relay response: ") + ), + "{body_error:?}" + ); + + handle.join().unwrap(); +} + #[test] fn profile_event_rejects_invalid_auth_tag() { let agent_keys = nostr::Keys::generate();