From 104588ff207effca7955f866fd5500c4b1e57c58 Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Tue, 2 Jun 2026 19:12:28 +1000 Subject: [PATCH] fix(serverless): make agent replies resilient to dead/rate-limited relays - Agent reply retries the full relay fan-out with backoff when every relay rate-limited (transient), instead of silently dropping the reply. Hard rejections are not retried. - Prune relays that fail the WebSocket handshake (403/paid/unreachable) from the agent's working set so discovery and publish only use healthy relays. - Launch managed agents with the CURRENT workspace relay list (serverless) rather than the list frozen into the agent record at creation time, so stale/paid relays drop out without recreating the agent. - Auto-fill the recommended public relays when the serverless toggle is enabled in the Add Workspace dialog (no more tapping each chip). --- crates/sprout-acp/src/relay.rs | 117 ++++++++++++++---- .../src-tauri/src/managed_agents/runtime.rs | 33 +++-- .../workspaces/ui/AddWorkspaceDialog.tsx | 17 ++- 3 files changed, 136 insertions(+), 31 deletions(-) diff --git a/crates/sprout-acp/src/relay.rs b/crates/sprout-acp/src/relay.rs index dd156cfa87..e70852506b 100644 --- a/crates/sprout-acp/src/relay.rs +++ b/crates/sprout-acp/src/relay.rs @@ -436,31 +436,81 @@ impl RestClient { // Fan out to every relay; succeed if ANY accepts. This makes the // agent's replies resilient to a single relay rate-limiting ("noting // too much") or being down — exactly what the desktop relay pool does. + // + // Rate-limits are transient: if a full fan-out pass fails ONLY because + // every healthy relay rate-limited us (no hard rejection, no success), + // we back off and retry the whole pass. The agent's reply is the one + // write that must not silently vanish — dropping it makes the agent + // look dead even though everything upstream worked. Hard rejections + // (bad event, restricted) are NOT retried; they won't change. let event_id = event.id.to_hex(); - let mut last_err: Option = None; - for relay in &self.ws_urls { - match self.submit_event_ws_one(relay, event).await { - Ok(v) => { - let accepted = v.get("accepted").and_then(|b| b.as_bool()).unwrap_or(false); - if accepted { - return Ok(v); + // Backoffs between full fan-out passes. `None` is the final pass with no + // further retry, so the loop runs `RETRY_BACKOFFS_MS.len() + 1` times. + const RETRY_BACKOFFS_MS: [u64; 3] = [800, 2000, 4000]; + let backoffs: Vec> = RETRY_BACKOFFS_MS + .iter() + .map(|ms| Some(*ms)) + .chain(std::iter::once(None)) + .collect(); + + for next_backoff in backoffs { + let mut last_err: Option = None; + let mut all_rate_limited = true; + let mut tried_any = false; + + for relay in &self.ws_urls { + match self.submit_event_ws_one(relay, event).await { + Ok(v) => { + let accepted = v.get("accepted").and_then(|b| b.as_bool()).unwrap_or(false); + if accepted { + return Ok(v); + } + let msg = v + .get("message") + .and_then(|m| m.as_str()) + .unwrap_or("") + .to_string(); + tried_any = true; + if !msg.contains("rate-limit") { + // A hard rejection (restricted/invalid) won't change + // on retry — don't keep hammering for it. + all_rate_limited = false; + } + tracing::warn!("relay {relay} rejected event (trying next): {msg}"); + last_err = Some(RelayError::Http(format!("relay rejected: {msg}"))); + } + Err(e) => { + // Connect/IO errors (dead/paid relay) are not rate-limits; + // retrying the whole pass won't revive them, so don't loop + // forever on a list of dead relays. + all_rate_limited = false; + tracing::warn!("relay {relay} publish failed (trying next): {e}"); + last_err = Some(e); } - let msg = v - .get("message") - .and_then(|m| m.as_str()) - .unwrap_or("") - .to_string(); - tracing::warn!("relay {relay} rejected event (trying next): {msg}"); - last_err = Some(RelayError::Http(format!("relay rejected: {msg}"))); - } - Err(e) => { - tracing::warn!("relay {relay} publish failed (trying next): {e}"); - last_err = Some(e); } } + + // Retry only when the sole reason for failure was rate-limiting and + // we actually got a rate-limit response from at least one relay, and + // there is another pass left. + if tried_any && all_rate_limited { + if let Some(backoff) = next_backoff { + tracing::warn!( + "all relays rate-limited event {event_id}; retrying in {backoff}ms" + ); + tokio::time::sleep(Duration::from_millis(backoff)).await; + continue; + } + } + + return Err(last_err.unwrap_or_else(|| { + RelayError::Http(format!("no relay accepted event {event_id}")) + })); } - Err(last_err - .unwrap_or_else(|| RelayError::Http(format!("no relay accepted event {event_id}")))) + + Err(RelayError::Http(format!( + "no relay accepted event {event_id} after retries" + ))) } /// Publish a signed event to a single relay over plain WS (serverless mode) @@ -778,6 +828,13 @@ impl HarnessRelay { let mut bg_handles: Vec> = Vec::new(); let mut connected_any = false; let mut last_err: Option = None; + // Only the relays we actually connected to are usable for reads/writes. + // A relay that refuses the WebSocket handshake (403 Forbidden / paid / + // unreachable) will never serve this session — keeping it in the + // working set just makes discovery and publish waste time on it and, + // worse, can leave the agent's reply with no healthy relay to fail over + // to. Prune them up front. + let mut connected_urls: Vec = Vec::new(); for url in &relay_urls { let (ws, handshake_buffer) = @@ -790,6 +847,7 @@ impl HarnessRelay { } }; connected_any = true; + connected_urls.push(url.clone()); let (task_cmd_tx, task_cmd_rx) = mpsc::channel::(CMD_CHANNEL_CAPACITY); per_relay_cmd_tx.push(task_cmd_tx); @@ -826,6 +884,23 @@ impl HarnessRelay { return Err(last_err.unwrap_or(RelayError::ConnectionClosed)); } + // Reads/writes fan out only over relays that actually connected. + let working_urls = if connected_urls.is_empty() { + relay_urls.clone() + } else { + connected_urls + }; + // Prefer a connected relay as the primary/display label. + let primary = working_urls.first().cloned().unwrap_or(primary); + if working_urls.len() < relay_urls.len() { + tracing::info!( + "serverless: using {}/{} relays that connected: {}", + working_urls.len(), + relay_urls.len(), + working_urls.join(", ") + ); + } + // Command fan-out: clone each command to every relay's task. let fanout_handle = tokio::spawn(async move { while let Some(cmd) = cmd_rx.recv().await { @@ -850,7 +925,7 @@ impl HarnessRelay { .build() .map_err(|e| RelayError::Http(format!("failed to build HTTP client: {e}")))?, relay_url: primary, - relay_urls, + relay_urls: working_urls, keys: keys.clone(), auth_tag, serverless, diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index c5cf1a7444..0fdae90c3b 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -824,19 +824,34 @@ pub fn spawn_agent_child( } command.env("RUST_LOG", child_rust_log_filter()); command.env("SPROUT_PRIVATE_KEY", &record.private_key_nsec); - command.env("SPROUT_RELAY_URL", &record.relay_url); + // Relay URL: in serverless mode, use the CURRENT workspace relay list, not + // the list frozen into the agent record at creation time. Otherwise an + // agent created under an old workspace keeps hammering stale/paid relays + // (e.g. nostr.land/nostr.wine) that were since removed — and if the only + // healthy relay rate-limits, the agent's reply has nowhere to fail over to + // and silently vanishes. Server mode keeps the record's relay (it points at + // a specific Sprout server, not a swappable public-relay set). + use tauri::Manager; + let state = app.state::(); + let current = crate::relay::relay_ws_url_with_override(&state); + let relay_url = if state.is_serverless() && !current.trim().is_empty() { + current + } else { + record.relay_url.clone() + }; + command.env("SPROUT_RELAY_URL", &relay_url); // Serverless mode: tell the ACP harness (and the sprout CLI it spawns for // replies) to use plain-WS transport against a generic relay instead of the // Sprout HTTP bridge. Inherited from the active workspace. See // docs/SPROUT_LITE_MODE.md. - { - use tauri::Manager; - let serverless = app.state::().is_serverless(); - command.env( - "SPROUT_SERVERLESS", - if serverless { "true" } else { "false" }, - ); - } + command.env( + "SPROUT_SERVERLESS", + if state.is_serverless() { + "true" + } else { + "false" + }, + ); command.env("SPROUT_ACP_AGENT_COMMAND", &resolved_agent_command); command.env("SPROUT_ACP_AGENT_ARGS", agent_args.join(",")); match &resolved_mcp_command { diff --git a/desktop/src/features/workspaces/ui/AddWorkspaceDialog.tsx b/desktop/src/features/workspaces/ui/AddWorkspaceDialog.tsx index 477ed770db..7afcf74e26 100644 --- a/desktop/src/features/workspaces/ui/AddWorkspaceDialog.tsx +++ b/desktop/src/features/workspaces/ui/AddWorkspaceDialog.tsx @@ -87,7 +87,22 @@ export function AddWorkspaceDialog({ setServerless(e.target.checked)} + onChange={(e) => { + const on = e.target.checked; + setServerless(on); + // Turning serverless ON pre-fills the recommended relays so the + // user doesn't have to tap every chip. Only auto-fill when the + // field is empty or still holds the defaults, so we never stomp + // a custom relay list the user typed. Turning it OFF clears the + // default list (a single Sprout relay is expected instead). + if (on) { + if (relayUrl.trim() === "") { + setRelayUrl(DEFAULT_SERVERLESS_RELAY); + } + } else if (relayUrl.trim() === DEFAULT_SERVERLESS_RELAY) { + setRelayUrl(""); + } + }} type="checkbox" />