diff --git a/AGENTS.md b/AGENTS.md
index be1f1cc4a3..c94ba3881e 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -462,34 +462,41 @@ new arbitrary text-size literal — px **or** rem/em. Genuinely decorative glyph
(e.g. the `text-[6rem]` avatar emoji) are allowlisted by `path:line` in that
script.
-### Workspace Switching
+### Community Switching
-The desktop app supports multiple workspaces (each backed by a different relay).
-Switching workspaces does **not** reload the page — it uses React key-based
-remounting. `` in `App.tsx` forces the entire
-workspace-scoped subtree to unmount and remount with fresh state.
+The desktop app supports multiple communities (each backed by a different relay).
+Switching communities does **not** reload the page — it uses React key-based
+remounting. `` in `App.tsx` forces the entire
+community-scoped subtree to unmount and remount with fresh state.
**Module-level singletons must be explicitly reset.** React remounting only
clears React state (useState, useRef, context). Module-level variables (Maps,
-class instances, cached promises) survive across remounts. Every workspace-scoped
-singleton needs a reset function wired into `resetWorkspaceState()` in
-`desktop/src/features/workspaces/useWorkspaceInit.ts`.
+class instances, cached promises) survive across remounts. Every community-scoped
+singleton needs a reset function wired into `resetCommunityState()` in
+`desktop/src/features/communities/useCommunityInit.ts`.
-Current singletons that are reset on workspace switch:
+Current singletons that are reset on community switch:
- `relayClient.disconnect()` — WebSocket teardown + promise rejection
+- `resetRateLimitGate()` — clears any active rate-limit window from the old relay
+- `clearAllDrafts()` — message draft cache
+- `resetAgentObserverStore()` — agent observer relay store
+- `resetActiveAgentTurnsStore()` — active agent turn timers
+- `resetAgentWorkingSignal()` — agent working indicator signal
+- `resetSidebarRelayConnectionCardState()` — sidebar relay card dismiss state
- `resetMediaCaches()` — proxy port and relay origin caches
+- `resetVideoPlayerState()` — video player singleton
+- `resetRenderScopedReactionHydration()` — reaction hydration cache
- `clearSearchHitEventCache()` — search result event cache
-- `clearAllDrafts()` — message draft cache
+- `clearMarkdownNodeCache()` — markdown parse-node cache
**If you add a new module-level cache, Map, or class instance that holds
-workspace-scoped data, you must add its reset to `resetWorkspaceState()`.**
-Failure to do so causes data from the old workspace to leak into the new one.
+community-scoped data, you must add its reset to `resetCommunityState()`.**
+Failure to do so causes data from the old community to leak into the new one.
Key files:
-- `desktop/src/app/App.tsx` — workspace key, init gate, remount boundary
-- `desktop/src/features/workspaces/useWorkspaceInit.ts` — `resetWorkspaceState()`, applies config to Tauri backend
-- `desktop/src/features/workspaces/useWorkspaces.tsx` — `WorkspacesProvider` context (shared state for App + AppShell)
-- `desktop/src/main.tsx` — provider hierarchy (`QueryClientProvider` > `WorkspacesProvider` > `App`)
+- `desktop/src/app/App.tsx` — community key, init gate, remount boundary
+- `desktop/src/features/communities/useCommunityInit.ts` — `resetCommunityState()`, applies config to Tauri backend
+- `desktop/src/main.tsx` — provider hierarchy (`QueryClientProvider` > `App`)
---
diff --git a/desktop/scripts/check-file-sizes.mjs b/desktop/scripts/check-file-sizes.mjs
index 1963acc3bc..845f991623 100644
--- a/desktop/scripts/check-file-sizes.mjs
+++ b/desktop/scripts/check-file-sizes.mjs
@@ -316,7 +316,16 @@ const overrides = new Map([
// migration/materialize.rs; ratchet held at 1110.
["src-tauri/src/migration_tests.rs", 1110],
["src-tauri/src/nostr_convert.rs", 1126],
- ["src/shared/api/relayClientSession.ts", 1022],
+ // degraded-network resilience: relay.rs grew past 1000 with the addition of
+ // relay_error_message hint-capping (oversized-hint test via loopback TCP) and
+ // the relay_admission freshness-verification test. The loopback mock was
+ // hardened (std::net + request-read-before-write) adding ~10 lines.
+ // Queued to split test helpers to relay/tests.rs.
+ ["src-tauri/src/relay.rs", 1047],
+ // degraded-network resilience: visibleChannelId field + getter/setter, NOTICE
+ // handler for relay back-pressure, and rate-limit gate imports add ~74 lines
+ // of load-bearing degraded-network recovery code. Queued to split.
+ ["src/shared/api/relayClientSession.ts", 1096],
// Boot-time event sync (persona/team/agent event reconcile) was split out
// to event_sync.rs, ratcheting this limit 1575 → 1310. Remaining content is
// the pre-identity data migrations; still queued to split further.
diff --git a/desktop/src-tauri/Cargo.toml b/desktop/src-tauri/Cargo.toml
index 65be49e60b..8a9feb562a 100644
--- a/desktop/src-tauri/Cargo.toml
+++ b/desktop/src-tauri/Cargo.toml
@@ -123,3 +123,6 @@ tempfile = "3"
strip-ansi-escapes = "0.2"
[dev-dependencies]
+# `test-util` enables tokio's paused-clock (`start_paused`) so the relay
+# admission gate tests can assert exact wait durations without real sleeps.
+tokio = { version = "1", features = ["test-util"] }
diff --git a/desktop/src-tauri/src/commands/personas/snapshot/import.rs b/desktop/src-tauri/src/commands/personas/snapshot/import.rs
index 9e2b8b3921..ac5c0eace6 100644
--- a/desktop/src-tauri/src/commands/personas/snapshot/import.rs
+++ b/desktop/src-tauri/src/commands/personas/snapshot/import.rs
@@ -640,6 +640,10 @@ async fn submit_engram_event(
use crate::relay::build_nip98_auth_header_for_keys;
use reqwest::Method;
+ // 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 mut request = state
.http_client
diff --git a/desktop/src-tauri/src/commands/team_snapshot.rs b/desktop/src-tauri/src/commands/team_snapshot.rs
index dc376ca064..0476be79a9 100644
--- a/desktop/src-tauri/src/commands/team_snapshot.rs
+++ b/desktop/src-tauri/src/commands/team_snapshot.rs
@@ -901,6 +901,10 @@ async fn submit_engram_event(
use crate::relay::build_nip98_auth_header_for_keys;
use reqwest::Method;
+ // 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 mut request = state
.http_client
diff --git a/desktop/src-tauri/src/commands/workspace.rs b/desktop/src-tauri/src/commands/workspace.rs
index cb17ad608e..561e901998 100644
--- a/desktop/src-tauri/src/commands/workspace.rs
+++ b/desktop/src-tauri/src/commands/workspace.rs
@@ -143,6 +143,9 @@ pub async fn apply_workspace(
let mut override_guard = state.relay_url_override.lock().map_err(|e| e.to_string())?;
*override_guard = Some(relay_url);
}
+ // Reset the Rust-side admission gate when switching workspace/community,
+ // matching `resetRateLimitGate()` on the TS side (useCommunityInit.ts:38).
+ crate::relay_admission::reset_gate_for_workspace_change();
if let Some(keys) = parsed_keys {
let mut keys_guard = state.keys.lock().map_err(|e| e.to_string())?;
diff --git a/desktop/src-tauri/src/huddle/pipeline.rs b/desktop/src-tauri/src/huddle/pipeline.rs
index bc9a1abd41..ceccedd8b6 100644
--- a/desktop/src-tauri/src/huddle/pipeline.rs
+++ b/desktop/src-tauri/src/huddle/pipeline.rs
@@ -305,6 +305,11 @@ pub(crate) fn spawn_transcription_task(
continue;
}
};
+ // Wait before signing: the relay enforces NIP-98 freshness (±60s)
+ // and the gate may hold for up to MAX_HINT_SECONDS (300s). Sign
+ // the kind event and build NIP-98 auth after the wait so both
+ // timestamps are fresh — single clean order: wait → sign → auth → send.
+ crate::relay_admission::wait_for_rate_limit().await;
let event = match builder.sign_with_keys(&keys) {
Ok(e) => e,
Err(e) => {
@@ -327,21 +332,23 @@ 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 = {
+ http_client
+ .post(&url)
+ .header("Authorization", auth_header)
+ .header("Content-Type", "application/json")
+ .body(body_bytes)
+ .send()
+ .await
+ };
match response {
Ok(resp) if resp.status().is_success() => {}
Ok(resp) => {
- eprintln!(
- "buzz-desktop: STT kind:9 post failed: HTTP {}",
- resp.status()
- );
+ // Route through relay_error_message so a 429 arms the
+ // admission gate for subsequent relay sends.
+ let msg = crate::relay::relay_error_message(resp).await;
+ eprintln!("buzz-desktop: STT kind:9 post failed: {msg}");
}
Err(e) => {
eprintln!("buzz-desktop: STT kind:9 post failed: {e}");
diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs
index 9d6424aca2..c22de0205f 100644
--- a/desktop/src-tauri/src/lib.rs
+++ b/desktop/src-tauri/src/lib.rs
@@ -23,6 +23,7 @@ pub mod nostr_convert;
mod prevent_sleep;
mod ptt_shortcut;
mod relay;
+mod relay_admission;
mod reset;
mod secret_store;
mod shutdown;
diff --git a/desktop/src-tauri/src/relay.rs b/desktop/src-tauri/src/relay.rs
index 0c253ed0ae..99ad702c1c 100644
--- a/desktop/src-tauri/src/relay.rs
+++ b/desktop/src-tauri/src/relay.rs
@@ -231,6 +231,17 @@ pub(crate) async fn parse_json_response(
.map_err(|_| MALFORMED_RESPONSE_MESSAGE.to_string())
}
+/// Extract the `retry in Ns` hint from a rate-limit error string.
+///
+/// Matches the canonical format emitted by the relay in both HTTP 429 bodies
+/// and CLOSED/NOTICE messages: `quota exceeded; retry in 4s`.
+fn extract_retry_in_hint(body: &str) -> Option {
+ let re_match = body.find("retry in ")?;
+ let after = &body[re_match + "retry in ".len()..];
+ let digits: String = after.chars().take_while(|c| c.is_ascii_digit()).collect();
+ digits.parse::().ok()
+}
+
pub async fn relay_error_message(response: reqwest::Response) -> String {
let status = response.status();
@@ -250,6 +261,27 @@ pub async fn relay_error_message(response: reqwest::Response) -> String {
// Real relay error: extract the structured message field if available.
let body = response.text().await.unwrap_or_default();
+ // 429 Too Many Requests → typed `relay rate-limited:` prefix so the TS
+ // client can activate the rate-limit gate without confusing it with a
+ // connectivity failure (`relay unreachable:`). Also arm the Rust-side
+ // admission gate here — the one place every relay HTTP error funnels
+ // through — so the next relay-backed command waits out the quota window
+ // instead of burning it (see `relay_admission`).
+ if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
+ let hint = extract_retry_in_hint(&body);
+ // Clamp the hint to MAX_HINT_SECONDS before arming the Rust gate AND
+ // before embedding it in the returned string. Every consumer (Rust gate
+ // via `activate_rate_limit` and TS gate via `applyTauriRateLimitIfNeeded`)
+ // must see the same capped value — a single policy point prevents the TS
+ // gate from receiving an uncapped hint from an untrusted relay.
+ let capped_hint = hint.map(|s| s.min(crate::relay_admission::MAX_HINT_SECONDS));
+ crate::relay_admission::activate_rate_limit(capped_hint);
+ if let Some(secs) = capped_hint {
+ return format!("relay rate-limited: retry in {secs}s");
+ }
+ return "relay rate-limited: quota exceeded".to_string();
+ }
+
if let Ok(value) = serde_json::from_str::(&body) {
if let Some(message) = value.get("message").and_then(serde_json::Value::as_str) {
return format!("relay returned {status}: {message}");
@@ -286,6 +318,7 @@ pub async fn query_relay_at(
api_base_url: &str,
filters: &[serde_json::Value],
) -> Result, String> {
+ crate::relay_admission::wait_for_rate_limit().await;
let url = format!("{}/query", api_base_url);
let body_bytes =
serde_json::to_vec(filters).map_err(|e| format!("filter serialization failed: {e}"))?;
@@ -382,6 +415,7 @@ pub async fn sync_managed_agent_profile(
avatar_url: Option<&str>,
auth_tag: Option<&str>, // NIP-OA auth tag JSON
) -> Result<(), String> {
+ crate::relay_admission::wait_for_rate_limit().await;
// Build a signed kind:0 profile event (with optional NIP-OA auth tag).
let event = build_profile_event(agent_keys, display_name, avatar_url, auth_tag)?;
let event_json = event.as_json();
@@ -482,6 +516,7 @@ pub async fn submit_event(
builder: nostr::EventBuilder,
state: &AppState,
) -> Result {
+ crate::relay_admission::wait_for_rate_limit().await;
// All synchronous work (signing) must complete before any .await
// so the MutexGuard is dropped and the future remains Send.
let url = format!("{}/events", relay_api_base_url_with_override(state));
@@ -529,6 +564,7 @@ pub async fn submit_signed_event(
event: &nostr::Event,
state: &AppState,
) -> Result {
+ crate::relay_admission::wait_for_rate_limit().await;
let url = format!("{}/events", relay_api_base_url_with_override(state));
let body_bytes = event.as_json().into_bytes();
let auth_header = {
@@ -586,6 +622,7 @@ pub async fn submit_signed_event_with_keys(
if event.pubkey != keys.public_key() {
return Err("signed event does not match the publishing identity".to_string());
}
+ crate::relay_admission::wait_for_rate_limit().await;
let url = format!("{}/events", relay_api_base_url_with_override(state));
let body_bytes = event.as_json().into_bytes();
let auth_header = build_nip98_auth_header_for_keys(keys, &Method::POST, &url, &body_bytes)?;
@@ -624,10 +661,106 @@ pub async fn submit_signed_event_with_keys(
mod tests {
use super::{
build_profile_event, classify_intercepted_response, effective_agent_relay_url,
- parse_command_response, relay_http_base_url, MALFORMED_RESPONSE_MESSAGE,
+ extract_retry_in_hint, parse_command_response, relay_http_base_url,
+ MALFORMED_RESPONSE_MESSAGE,
};
use serde::Deserialize;
+ // ── extract_retry_in_hint ────────────────────────────────────────────────
+
+ #[test]
+ fn extracts_hint_from_429_body() {
+ assert_eq!(
+ extract_retry_in_hint(r#"{"error":"rate-limited: quota exceeded; retry in 4s"}"#),
+ Some(4)
+ );
+ }
+
+ #[test]
+ fn extracts_hint_when_no_json_wrapper() {
+ assert_eq!(extract_retry_in_hint("retry in 30s"), Some(30));
+ }
+
+ #[test]
+ fn returns_none_when_no_hint_present() {
+ assert_eq!(
+ extract_retry_in_hint(r#"{"error":"rate-limited: quota exceeded"}"#),
+ None
+ );
+ assert_eq!(extract_retry_in_hint(""), None);
+ }
+
+ #[test]
+ fn overlong_digit_string_returns_none() {
+ // A digit sequence that exceeds u64::MAX cannot be parsed; the function
+ // must return None (→ caller uses the default) rather than panicking.
+ assert_eq!(
+ extract_retry_in_hint("retry in 99999999999999999999999s"),
+ None
+ );
+ }
+
+ // ── relay_error_message: hint capping ────────────────────────────────────
+ //
+ // Verify that an oversized relay hint is capped in the returned message
+ // string, not just inside `activate_rate_limit()`. This guarantees every
+ // consumer — including the TS gate via `applyTauriRateLimitIfNeeded` —
+ // receives the capped value rather than the raw untrusted relay value.
+
+ #[tokio::test]
+ async fn oversized_hint_is_capped_in_relay_error_message_string() {
+ use crate::relay_admission::MAX_HINT_SECONDS;
+ use std::io::{Read as _, Write as _};
+
+ // Use a std::net listener on a std::thread — the same pattern as the
+ // relay_admission loopback tests. This avoids two races that cause CI
+ // failures with tokio::net + into_std():
+ // 1. No request read: the client is still sending when the response
+ // arrives → hyper `UnexpectedMessage`/`Canceled` under load.
+ // 2. into_std() leaves the socket in nonblocking mode → write_all
+ // may return WouldBlock and silently drop the response.
+ let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+ let addr = listener.local_addr().unwrap();
+
+ // Serve a 429 with a hint far exceeding MAX_HINT_SECONDS (300).
+ let oversized = 1_000_000u64;
+ let body = format!(r#"{{"error":"rate-limited: quota exceeded; retry in {oversized}s"}}"#);
+ let body_len = body.len();
+ std::thread::spawn(move || {
+ if let Ok((mut stream, _)) = listener.accept() {
+ // Read the request first so the client finishes sending before
+ // we write the response — mirrors relay_admission.rs pattern.
+ let mut buf = [0u8; 4096];
+ let _ = stream.read(&mut buf);
+ let response = format!(
+ "HTTP/1.1 429 Too Many Requests\r\nContent-Type: application/json\r\nContent-Length: {body_len}\r\nConnection: close\r\n\r\n{body}"
+ );
+ let _ = stream.write_all(response.as_bytes());
+ let _ = stream.flush();
+ }
+ });
+
+ let client = reqwest::Client::new();
+ let response = client
+ .get(format!("http://{addr}/"))
+ .send()
+ .await
+ .expect("request must succeed");
+
+ let msg = super::relay_error_message(response).await;
+
+ // The message must embed the CAPPED hint, not the raw 1 000 000.
+ assert_eq!(
+ msg,
+ format!("relay rate-limited: retry in {MAX_HINT_SECONDS}s"),
+ "relay_error_message must embed the capped hint, not the raw untrusted value"
+ );
+ assert!(
+ !msg.contains(&oversized.to_string()),
+ "raw oversized hint must not appear in the message string"
+ );
+ }
+
// ── effective_agent_relay_url: per-agent override precedence ─────────────
#[test]
diff --git a/desktop/src-tauri/src/relay_admission.rs b/desktop/src-tauri/src/relay_admission.rs
new file mode 100644
index 0000000000..15222f8590
--- /dev/null
+++ b/desktop/src-tauri/src/relay_admission.rs
@@ -0,0 +1,504 @@
+//! Admission gate for relay HTTP bridge requests.
+//!
+//! When the relay answers 429, every relay-backed HTTP request must hold new
+//! sends until the quota window clears — matching the TS-side gate in
+//! `relayRateLimitGate.ts` that already governs WebSocket operations.
+//!
+//! **Coverage:** all entry points in `relay.rs` (`query_relay_at`,
+//! `submit_event`, `submit_signed_event`, `submit_signed_event_with_keys`,
+//! `sync_managed_agent_profile`) and the three previously-direct senders
+//! (`submit_engram_event` in snapshot import + team_snapshot, huddle STT)
+//! all call `wait_for_rate_limit()` before `.send()`.
+//!
+//! **Media upload/download and `/info`** call `relay_error_message()` on
+//! non-200 responses, so their 429s arm the shared gate as conservative
+//! back-off (any relay overload signal is worth honouring across domains).
+//! They do not call `wait_for_rate_limit()` themselves — their operations
+//! are driven by user-initiated file transfers rather than bridge event flow,
+//! and they have independent retry logic.
+//!
+//! **Community scope:** the gate is reset on every `apply_workspace` call,
+//! mirroring the TS gate's `resetRateLimitGate()` on community switch in
+//! `useCommunityInit.ts`. A 429 from community A cannot stall community B.
+//!
+//! Mirrors the TS gate's semantics: overlapping hints never shrink the window,
+//! and a hint-less 429 arms the same 10-second default.
+
+use std::sync::Mutex;
+use tokio::time::{sleep_until, Duration, Instant};
+
+/// Minimum gate duration when the relay provides no `retry in Ns` hint.
+/// Deliberately equal to `DEFAULT_RATE_LIMIT_SECONDS` in `relayRateLimitGate.ts`
+/// so both halves of the client back off for the same window.
+const DEFAULT_RATE_LIMIT_SECONDS: u64 = 10;
+
+/// Maximum hint the gate will honour from a relay 429 response.
+/// Prevents an untrusted relay from pinning traffic for an unreasonable window
+/// or overflowing `Instant` arithmetic.
+/// Exposed `pub` so `relay.rs` can clamp the hint before embedding it in the
+/// returned error string — ensuring every consumer (Rust gate and TS gate via
+/// `applyTauriRateLimitIfNeeded`) sees the same capped value.
+pub const MAX_HINT_SECONDS: u64 = 300;
+
+static GATE_EXPIRY: Mutex> = Mutex::new(None);
+
+/// Arm (or extend) the admission gate from a relay 429.
+///
+/// `retry_in_seconds` is the parsed `retry in Ns` hint, if the relay provided
+/// one. Hints are capped at `MAX_HINT_SECONDS`; values of zero or `None` use
+/// `DEFAULT_RATE_LIMIT_SECONDS`. The expiry only ever moves forward: a shorter
+/// hint arriving under a longer active window is ignored, so overlapping 429s
+/// never schedule a premature retry.
+pub fn activate_rate_limit(retry_in_seconds: Option) {
+ let secs = match retry_in_seconds {
+ Some(s) if s > 0 => s.min(MAX_HINT_SECONDS),
+ _ => DEFAULT_RATE_LIMIT_SECONDS,
+ };
+ let new_expiry = Instant::now()
+ .checked_add(Duration::from_secs(secs))
+ .unwrap_or_else(|| Instant::now() + Duration::from_secs(DEFAULT_RATE_LIMIT_SECONDS));
+ let mut guard = GATE_EXPIRY.lock().unwrap_or_else(|e| e.into_inner());
+ match *guard {
+ Some(current) if new_expiry <= current => {}
+ _ => *guard = Some(new_expiry),
+ }
+}
+
+/// Wait until the admission gate is clear.
+///
+/// Returns immediately when no gate is active. Loops after sleeping because a
+/// concurrent 429 may extend the expiry while this caller is parked.
+pub async fn wait_for_rate_limit() {
+ loop {
+ let expiry = {
+ let guard = GATE_EXPIRY.lock().unwrap_or_else(|e| e.into_inner());
+ match *guard {
+ Some(expiry) if expiry > Instant::now() => Some(expiry),
+ _ => None,
+ }
+ };
+ match expiry {
+ Some(expiry) => sleep_until(expiry).await,
+ None => return,
+ }
+ }
+}
+
+/// Reset the gate on a workspace/community change.
+///
+/// Called by `apply_workspace` to ensure a 429 from community A does not stall
+/// requests to community B. Mirrors `resetRateLimitGate()` in
+/// `useCommunityInit.ts`.
+pub fn reset_gate_for_workspace_change() {
+ *GATE_EXPIRY.lock().unwrap_or_else(|e| e.into_inner()) = None;
+}
+
+/// Reset the gate. Test-only: production never clears an armed window early
+/// except via `reset_gate_for_workspace_change`.
+#[cfg(test)]
+pub fn reset_rate_limit_gate() {
+ *GATE_EXPIRY.lock().unwrap_or_else(|e| e.into_inner()) = None;
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ // The gate is a process-wide static shared by every test in this binary,
+ // so all gate tests serialize on one async lock to keep armed expiries
+ // from bleeding between parallel test threads.
+ pub(crate) static TEST_SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+ #[tokio::test(start_paused = true)]
+ async fn wait_returns_immediately_when_gate_is_inactive() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+ let start = Instant::now();
+ wait_for_rate_limit().await;
+ assert_eq!(
+ Instant::now(),
+ start,
+ "inactive gate must not consume any (paused) time"
+ );
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn hintless_429_arms_the_ten_second_default() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+ activate_rate_limit(None);
+ let start = Instant::now();
+ wait_for_rate_limit().await;
+ assert_eq!(Instant::now() - start, Duration::from_secs(10));
+ reset_rate_limit_gate();
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn shorter_hint_never_shrinks_an_active_window() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+ activate_rate_limit(Some(8));
+ activate_rate_limit(Some(1));
+ let start = Instant::now();
+ wait_for_rate_limit().await;
+ assert_eq!(
+ Instant::now() - start,
+ Duration::from_secs(8),
+ "the 1s hint must not shorten the active 8s window"
+ );
+ reset_rate_limit_gate();
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn concurrent_429_extends_the_window_for_parked_waiters() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+ activate_rate_limit(Some(2));
+ let start = Instant::now();
+ let waiter = tokio::spawn(async {
+ wait_for_rate_limit().await;
+ });
+ // Extend while the waiter is parked on the first expiry.
+ tokio::time::sleep(Duration::from_secs(1)).await;
+ activate_rate_limit(Some(4));
+ waiter.await.unwrap();
+ assert_eq!(
+ Instant::now() - start,
+ Duration::from_secs(5),
+ "waiter must respect the extension armed mid-sleep (1s + 4s)"
+ );
+ reset_rate_limit_gate();
+ }
+
+ // ── hint capping and overflow safety ─────────────────────────────────────
+
+ #[tokio::test(start_paused = true)]
+ async fn hint_zero_uses_default() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+ activate_rate_limit(Some(0));
+ let start = Instant::now();
+ wait_for_rate_limit().await;
+ assert_eq!(
+ Instant::now() - start,
+ Duration::from_secs(DEFAULT_RATE_LIMIT_SECONDS),
+ "hint=0 must use the default"
+ );
+ reset_rate_limit_gate();
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn hint_at_max_is_honoured() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+ activate_rate_limit(Some(MAX_HINT_SECONDS));
+ let start = Instant::now();
+ wait_for_rate_limit().await;
+ assert_eq!(
+ Instant::now() - start,
+ Duration::from_secs(MAX_HINT_SECONDS),
+ "hint at the cap must be honoured in full"
+ );
+ reset_rate_limit_gate();
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn oversize_hint_is_clamped_to_max() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+ // An oversize hint (including u64::MAX) must clamp rather than panic.
+ activate_rate_limit(Some(u64::MAX));
+ let start = Instant::now();
+ wait_for_rate_limit().await;
+ assert_eq!(
+ Instant::now() - start,
+ Duration::from_secs(MAX_HINT_SECONDS),
+ "u64::MAX hint must clamp to MAX_HINT_SECONDS"
+ );
+ reset_rate_limit_gate();
+ }
+
+ // ── community / workspace boundary ───────────────────────────────────────
+
+ #[tokio::test(start_paused = true)]
+ async fn workspace_change_clears_armed_gate() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+ activate_rate_limit(Some(60));
+ // Switch workspace — gate for community A must not stall community B.
+ reset_gate_for_workspace_change();
+ let start = Instant::now();
+ wait_for_rate_limit().await;
+ assert_eq!(
+ Instant::now(),
+ start,
+ "gate must be clear immediately after workspace change"
+ );
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn community_a_gate_does_not_block_community_b() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+ // Community A gets a 429 with a 30s window.
+ activate_rate_limit(Some(30));
+ // Community switch.
+ reset_gate_for_workspace_change();
+ // Community B's first request must not wait.
+ let start = Instant::now();
+ wait_for_rate_limit().await;
+ assert_eq!(
+ Instant::now(),
+ start,
+ "community A's armed gate must not delay community B"
+ );
+ }
+
+ /// A 429 on one admission-gated path withholds sends on a different path
+ /// until the hinted window expires.
+ ///
+ /// Arms the gate directly (as `relay_error_message` would on a 429 response),
+ /// then drives a second distinct gated send through the loopback acceptance
+ /// server and asserts it waited out the window before succeeding.
+ #[tokio::test]
+ async fn gate_armed_by_one_path_withholds_another_path() {
+ use std::io::{Read, Write};
+
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+
+ // The loopback server answers every request with 200 [].
+ let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+ let addr = listener.local_addr().unwrap();
+ let server = std::thread::spawn(move || loop {
+ let Ok((mut stream, _)) = listener.accept() else {
+ break;
+ };
+ 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: 2\r\nConnection: close\r\n\r\n[]",
+ );
+ let _ = stream.flush();
+ });
+
+ // Capture t0 before arming so elapsed is measured from before the gate
+ // expiry is set (expiry = now + 1s), guaranteeing elapsed ≥ 1s.
+ let t0 = std::time::Instant::now();
+
+ // Arm the gate for 1s — simulates what relay_error_message does on any
+ // gated path that receives a 429, e.g. submit_engram_event.
+ activate_rate_limit(Some(1));
+
+ let state = crate::app_state::build_app_state();
+ *state.relay_url_override.lock().unwrap() = Some(format!("http://{addr}"));
+ let filters = [serde_json::json!({ "kinds": [1], "limit": 1 })];
+
+ // query_relay is a different gated path — it must wait out the 1s window
+ // even though it was not the source of the 429.
+ let events = crate::relay::query_relay(&state, &filters)
+ .await
+ .expect("query must succeed after admission wait");
+ let elapsed = t0.elapsed();
+
+ assert!(
+ events.is_empty(),
+ "server returns empty; unexpected events: {events:?}"
+ );
+ assert!(
+ elapsed >= Duration::from_secs(1),
+ "cross-path send ran {}ms — it must wait out the 1s window armed by another path",
+ elapsed.as_millis()
+ );
+
+ drop(server);
+ reset_rate_limit_gate();
+ }
+
+ /// A waiter parked on community A's gate must NOT wake early when the gate
+ /// is reset by a workspace change — it sleeps until the original expiry,
+ /// then rechecks, finds the gate clear, and proceeds.
+ ///
+ /// This documents the contract: `sleep_until` is already scheduled against
+ /// A's expiry; the reset clears `GATE_EXPIRY` but cannot cancel an in-flight
+ /// sleep. The recheck loop in `wait_for_rate_limit` then sees `None` and
+ /// returns. Net effect: the waiter observes at most one full window, which
+ /// is the same bound as if the workspace had not changed.
+ #[tokio::test(start_paused = true)]
+ async fn parked_waiter_does_not_wake_early_after_workspace_reset() {
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+
+ // Arm a 5s gate for community A.
+ activate_rate_limit(Some(5));
+
+ let start = Instant::now();
+
+ // Spawn a waiter that parks on the gate.
+ let waiter = tokio::spawn(async {
+ wait_for_rate_limit().await;
+ Instant::now()
+ });
+
+ // After 2s (well before the 5s expiry), simulate a workspace change
+ // that resets the gate.
+ tokio::time::sleep(Duration::from_secs(2)).await;
+ reset_gate_for_workspace_change();
+
+ let woke_at = waiter.await.unwrap();
+ let elapsed = woke_at - start;
+
+ // The waiter must have waited at least until A's original expiry (5s).
+ // It cannot be woken early by the reset — only the recheck loop exit
+ // can terminate it, and the recheck fires after sleep_until(expiry).
+ assert!(
+ elapsed >= Duration::from_secs(5),
+ "waiter woke at {}ms — must not wake before A's 5s expiry even after reset",
+ elapsed.as_millis()
+ );
+
+ reset_rate_limit_gate();
+ }
+
+ /// Wait-then-sign ensures NIP-98 auth is fresh after an admission wait.
+ ///
+ /// The relay enforces NIP-98 freshness within ±60s (`TIMESTAMP_TOLERANCE_SECS`
+ /// in `buzz-auth`), while the gate honours hints up to `MAX_HINT_SECONDS`
+ /// (300s). Signing BEFORE the wait produces a stale `created_at` that would
+ /// be rejected on any active window >60s.
+ ///
+ /// This test arms a 1s gate, records the wall-clock time immediately AFTER
+ /// the wait returns, then builds the NIP-98 header and asserts its
+ /// `created_at` is ≥ the wake time — proving the sign happened post-wait.
+ #[tokio::test]
+ async fn nip98_auth_created_at_is_fresh_after_admission_wait() {
+ use base64::{engine::general_purpose::STANDARD as BASE64, Engine as _};
+ use serde::Deserialize;
+
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+
+ // Arm the gate for 1s (real time — NIP-98 uses SystemTime, not Tokio clock).
+ activate_rate_limit(Some(1));
+
+ // Wait out the gate — this is the critical ordering: wait THEN sign.
+ wait_for_rate_limit().await;
+
+ // Capture the wake time as a Unix timestamp AFTER the wait returns.
+ let wake_unix: u64 = std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .unwrap()
+ .as_secs();
+
+ // Build NIP-98 auth header (contains a signed event whose `created_at`
+ // is set at sign time — i.e., now, after the wait).
+ let keys = nostr::Keys::generate();
+ let header = crate::relay::build_nip98_auth_header_for_keys(
+ &keys,
+ &reqwest::Method::POST,
+ "https://relay.example.com/events",
+ b"{}",
+ )
+ .expect("header build must succeed");
+
+ // Decode the base64-encoded Nostr event from "Nostr ".
+ let b64 = header
+ .strip_prefix("Nostr ")
+ .expect("header must start with 'Nostr '");
+ let json_bytes = BASE64.decode(b64).expect("valid base64");
+
+ #[derive(Deserialize)]
+ struct EventShell {
+ created_at: u64,
+ }
+ let shell: EventShell = serde_json::from_slice(&json_bytes).expect("valid event JSON");
+
+ assert!(
+ shell.created_at >= wake_unix,
+ "NIP-98 created_at ({}) must be >= wake time ({}); \
+ signing before the wait produces stale auth",
+ shell.created_at,
+ wake_unix
+ );
+
+ reset_rate_limit_gate();
+ }
+
+ /// Acceptance: a 429 from one relay-backed command withholds the next
+ /// relay-backed command until the hinted window expires, then it resumes.
+ ///
+ /// Drives the production `query_relay` path end-to-end against a loopback
+ /// HTTP server — NIP-98 signing, `relay_error_message` classification and
+ /// gate arming, and the admission wait all execute for real. Real time is
+ /// required (the request crosses actual TCP), so the hint is kept at 1s.
+ #[tokio::test]
+ async fn http_429_withholds_next_relay_command_until_expiry_then_resumes() {
+ use std::io::{Read, Write};
+
+ let _serial = TEST_SERIAL.lock().await;
+ reset_rate_limit_gate();
+
+ let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+ let addr = listener.local_addr().unwrap();
+
+ // First request → 429 with a 1s retry hint; every later request → 200 [].
+ let server = std::thread::spawn(move || {
+ let responses = [
+ "HTTP/1.1 429 Too Many Requests\r\n\
+ Content-Type: application/json\r\n\
+ Content-Length: 53\r\n\
+ Connection: close\r\n\r\n\
+ {\"error\":\"rate-limited: quota exceeded; retry in 1s\"}",
+ "HTTP/1.1 200 OK\r\n\
+ Content-Type: application/json\r\n\
+ Content-Length: 2\r\n\
+ Connection: close\r\n\r\n\
+ []",
+ ];
+ for i in 0..2 {
+ let Ok((mut stream, _)) = listener.accept() else {
+ return;
+ };
+ let mut buf = [0u8; 4096];
+ let _ = stream.read(&mut buf);
+ let _ = stream.write_all(responses[i.min(1)].as_bytes());
+ let _ = stream.flush();
+ }
+ });
+
+ let state = crate::app_state::build_app_state();
+ *state.relay_url_override.lock().unwrap() = Some(format!("http://{addr}"));
+ let filters = [serde_json::json!({ "kinds": [1], "limit": 1 })];
+
+ // Command 1: the relay answers 429 — the caller sees the typed error
+ // and the admission gate arms for the hinted 1s window.
+ let err = crate::relay::query_relay(&state, &filters)
+ .await
+ .expect_err("first command must surface the 429");
+ assert!(
+ err.starts_with("relay rate-limited: retry in 1s"),
+ "429 must map to the typed rate-limited error, got: {err}"
+ );
+
+ // Measure from after command 1 returns so the timer only covers the
+ // admission wait (not command 1's own network time).
+ let after_first_429 = std::time::Instant::now();
+
+ // Command 2: must be withheld until the window expires, then resume
+ // and succeed against the now-healthy relay.
+ let events = crate::relay::query_relay(&state, &filters)
+ .await
+ .expect("second command must resume and succeed after expiry");
+ assert!(events.is_empty());
+
+ let wait_elapsed = after_first_429.elapsed();
+ assert!(
+ wait_elapsed >= Duration::from_secs(1),
+ "second command ran {}ms after the 429 — it must wait out the full 1s window",
+ wait_elapsed.as_millis()
+ );
+
+ server.join().unwrap();
+ reset_rate_limit_gate();
+ }
+}
diff --git a/desktop/src/features/communities/useCommunityInit.ts b/desktop/src/features/communities/useCommunityInit.ts
index 06ba8ff4f1..a50a8665c7 100644
--- a/desktop/src/features/communities/useCommunityInit.ts
+++ b/desktop/src/features/communities/useCommunityInit.ts
@@ -1,6 +1,7 @@
import { useEffect, useRef, useState } from "react";
import { relayClient } from "@/shared/api/relayClient";
+import { resetRateLimitGate } from "@/shared/api/relayRateLimitGate";
import { applyCommunity, getDefaultRelayUrl } from "@/shared/api/tauri";
import { getIdentity } from "@/shared/api/tauriIdentity";
import { getOverrides } from "@/shared/features";
@@ -34,6 +35,7 @@ import type { Community } from "./types";
*/
function resetCommunityState(): void {
relayClient.disconnect();
+ resetRateLimitGate();
clearAllDrafts();
resetAgentObserverStore();
resetActiveAgentTurnsStore();
diff --git a/desktop/src/features/home/hooks.ts b/desktop/src/features/home/hooks.ts
index 33d1bca204..7d18fb80d8 100644
--- a/desktop/src/features/home/hooks.ts
+++ b/desktop/src/features/home/hooks.ts
@@ -1,8 +1,12 @@
import { useQuery } from "@tanstack/react-query";
import { getHomeFeed } from "@/shared/api/tauri";
+import { useRelayConnection } from "@/shared/api/useRelayConnection";
export function useHomeFeedQuery() {
+ const connectionState = useRelayConnection();
+ const connected = connectionState === "connected";
+
return useQuery({
queryKey: ["home-feed"],
queryFn: () =>
@@ -12,6 +16,9 @@ export function useHomeFeedQuery() {
}),
staleTime: 15_000,
gcTime: 5 * 60 * 1_000,
- refetchInterval: 30_000,
+ // Pause background polling on degraded/stalled/disconnected connections.
+ // The relay can't serve the request anyway, and the spurious failures
+ // consume quota that the recovery path needs.
+ refetchInterval: connected ? 30_000 : false,
});
}
diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts
index 3d34d07d1f..062b0ee40b 100644
--- a/desktop/src/features/messages/hooks.ts
+++ b/desktop/src/features/messages/hooks.ts
@@ -31,7 +31,7 @@ import {
clearTimeoutState,
recordTimeoutFromRejection,
} from "@/features/moderation/lib/timeoutStore";
-import { relayClient } from "@/shared/api/relayClient";
+import { relayClient, setVisibleChannel } from "@/shared/api/relayClient";
import { customEmojiQueryKey } from "@/features/custom-emoji/hooks";
import { channelsQueryKey } from "@/features/channels/hooks";
import { reactionEmojiUrl } from "@/shared/api/customEmoji";
@@ -330,6 +330,17 @@ export function useChannelSubscription(channel: Channel | null) {
}
});
+ // Notify the relay client which channel is currently visible so its live
+ // subscriptions are replayed first on reconnect, reducing latency on
+ // degraded networks.
+ useEffect(() => {
+ if (!channelId || channelType === "forum") return;
+ setVisibleChannel(channelId);
+ return () => {
+ setVisibleChannel(null);
+ };
+ }, [channelId, channelType]);
+
useEffect(() => {
if (!channelId || channelType === "forum") {
return;
diff --git a/desktop/src/features/presence/hooks.ts b/desktop/src/features/presence/hooks.ts
index 2778c6d910..5ba8683693 100644
--- a/desktop/src/features/presence/hooks.ts
+++ b/desktop/src/features/presence/hooks.ts
@@ -2,6 +2,8 @@ import * as React from "react";
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import { relayClient } from "@/shared/api/relayClient";
+import { isRateLimited } from "@/shared/api/relayRateLimitGate";
+import { useRelayConnection } from "@/shared/api/useRelayConnection";
import { getOsIdleSeconds } from "@/shared/api/osIdle";
import { getPresence } from "@/shared/api/tauri";
import { normalizePubkey } from "@/shared/lib/pubkey";
@@ -79,6 +81,8 @@ export function usePresenceQuery(
) {
const normalizedPubkeys = normalizePubkeys(pubkeys);
const enabled = (options?.enabled ?? true) && normalizedPubkeys.length > 0;
+ const connectionState = useRelayConnection();
+ const connected = connectionState === "connected";
return useQuery({
enabled,
@@ -86,8 +90,9 @@ export function usePresenceQuery(
queryFn: () => getPresence(normalizedPubkeys),
staleTime: 30_000,
// Backstop poll: catches REST-only writers (ACP agents) and TTL expiry
- // (crashed clients). WS events handle the fast path.
- refetchInterval: 60_000,
+ // (crashed clients). WS events handle the fast path. Pause on degraded
+ // connections — HTTP presence calls fail anyway and consume relay quota.
+ refetchInterval: connected ? 60_000 : false,
});
}
@@ -370,6 +375,11 @@ export function usePresenceSession(pubkey?: string) {
}
const intervalId = window.setInterval(() => {
+ // Skip heartbeat ticks while the relay is unavailable or rate-limited —
+ // the publish would fail anyway and consumes quota the recovery needs.
+ if (relayClient.getConnectionState() !== "connected" || isRateLimited()) {
+ return;
+ }
syncPresence(currentStatus);
}, PRESENCE_HEARTBEAT_INTERVAL_MS);
diff --git a/desktop/src/shared/api/readOnlyRelayClient.ts b/desktop/src/shared/api/readOnlyRelayClient.ts
index 1c24acc8fc..c7446f4e70 100644
--- a/desktop/src/shared/api/readOnlyRelayClient.ts
+++ b/desktop/src/shared/api/readOnlyRelayClient.ts
@@ -8,10 +8,11 @@ import {
type RelaySubscriptionFilter,
} from "@/shared/api/relayClientShared";
import { closeWebSocket } from "@/shared/api/relayWebSocketClose";
-
-const AUTH_TIMEOUT_MS = 8_000;
-const HISTORY_TIMEOUT_MS = 8_000;
-const PUBLISH_TIMEOUT_MS = 8_000;
+import {
+ AUTH_TIMEOUT_MS,
+ HISTORY_TIMEOUT_MS,
+ PUBLISH_TIMEOUT_MS,
+} from "@/shared/api/relayClientSession";
type PendingHistory = {
events: RelayEvent[];
diff --git a/desktop/src/shared/api/relayClient.ts b/desktop/src/shared/api/relayClient.ts
index a558f93f50..9cf4c13447 100644
--- a/desktop/src/shared/api/relayClient.ts
+++ b/desktop/src/shared/api/relayClient.ts
@@ -1,3 +1,16 @@
import { RelayClient } from "@/shared/api/relayClientSession";
export const relayClient = new RelayClient();
+
+/**
+ * Notify the relay client which channel is currently visible in the UI.
+ *
+ * On reconnect, subscriptions for the visible channel are sent in the first
+ * replay batch so the user sees their active channel recover before others
+ * on degraded networks.
+ *
+ * Call with `null` when the user navigates away from a channel view.
+ */
+export function setVisibleChannel(id: string | null): void {
+ relayClient.setVisibleChannelId(id);
+}
diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts
index 7a151e5d28..6dcc857156 100644
--- a/desktop/src/shared/api/relayClientSession.ts
+++ b/desktop/src/shared/api/relayClientSession.ts
@@ -35,6 +35,12 @@ import {
prepareSubscriptionEvent,
} from "@/shared/api/relayClosedRecovery";
import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay";
+import {
+ activateRateLimit,
+ parseRateLimitHint,
+ waitForRateLimit,
+} from "@/shared/api/relayRateLimitGate";
+import { requestHistoryGated } from "@/shared/api/relayGateBoundary";
import { RelayConnectionStateEmitter } from "@/shared/api/relayConnectionStateEmitter";
import {
shouldRefuseConnect,
@@ -48,6 +54,22 @@ const RECONNECT_BASE_DELAY_MS = 1_000,
EVENT_BATCH_MS = 16,
AUX_BACKFILL_CONCURRENCY = 4;
+/**
+ * Op-level timeout constants. Raised from 8 s to 25 s to survive degraded
+ * networks where TLS handshakes and DNS resolution can take 3–10 s.
+ */
+export const AUTH_TIMEOUT_MS = 25_000;
+export const HISTORY_TIMEOUT_MS = 25_000;
+export const PUBLISH_TIMEOUT_MS = 25_000;
+
+/**
+ * The connection must remain stable for this long after a successful AUTH
+ * before the reconnect backoff delay resets to its base value. Stability-
+ * gated reset prevents repeated fast reconnects (flapping) from erasing the
+ * backoff that throttles them.
+ */
+export const BACKOFF_RESET_STABLE_MS = 60_000;
+
/**
* Passive liveness check. The relay sends heartbeat pings every 30s; if no
* inbound frame arrives for two heartbeat windows, treat the socket as stalled.
@@ -77,6 +99,8 @@ export class RelayClient {
private notifyReconnectListeners = false;
private onMessageChannel: Channel | null = null;
private connectionGeneration = 0;
+ private stabilityTimer: number | null = null;
+ private visibleChannelId: string | null = null;
/**
* Sticky terminal flag. Set when `resetConnection` is called with
@@ -100,6 +124,16 @@ export class RelayClient {
},
});
+ /**
+ * Track which channel the user is currently viewing so its subscriptions
+ * are sent first during reconnect replay — reducing visible latency on
+ * degraded networks where the relay REQ storm would otherwise delay all
+ * channels equally.
+ */
+ setVisibleChannelId(id: string | null) {
+ this.visibleChannelId = id;
+ }
+
/**
* Cleanly tear down the connection without scheduling a reconnect.
* Used during community switches to reset the singleton before the
@@ -112,6 +146,10 @@ export class RelayClient {
window.clearTimeout(this.reconnectTimeout);
this.reconnectTimeout = null;
}
+ if (this.stabilityTimer !== null) {
+ window.clearTimeout(this.stabilityTimer);
+ this.stabilityTimer = null;
+ }
this.stallWatchdog.stop();
this.connectionGeneration++;
this.keepAliveRequested = false;
@@ -119,6 +157,7 @@ export class RelayClient {
this.hasConnectedOnce = false;
this.notifyReconnectListeners = false;
this.terminal = false;
+ this.visibleChannelId = null;
this.connectionStateEmitter.set("idle");
if (this.wsId !== null) {
@@ -238,33 +277,16 @@ export class RelayClient {
return this.requestHistory(filter);
}
- private requestHistory(filter: RelaySubscriptionFilter) {
- return new Promise((resolve, reject) => {
- const subId = `history-${crypto.randomUUID()}`;
- const timeout = window.setTimeout(() => {
- this.subscriptions.delete(subId);
- void this.closeSubscription(subId);
- reject(new Error("Timed out while loading channel history."));
- }, 8_000);
-
- this.subscriptions.set(subId, {
- mode: "history",
- events: [],
- resolve,
- reject,
- timeout,
- });
-
- void this.sendRaw(["REQ", subId, filter]).catch((error) => {
- window.clearTimeout(timeout);
- this.subscriptions.delete(subId);
- reject(
- error instanceof Error
- ? error
- : new Error("Failed to request channel history."),
- );
- });
- });
+ private requestHistory(
+ filter: RelaySubscriptionFilter,
+ ): Promise {
+ return requestHistoryGated(
+ this.subscriptions,
+ (payload) => this.sendRaw(payload),
+ (subId) => this.closeSubscription(subId),
+ filter,
+ HISTORY_TIMEOUT_MS,
+ );
}
async sendMessage(
@@ -512,6 +534,13 @@ export class RelayClient {
}
private async connect() {
+ // Clear any pending stability timer from a previous connection — a new
+ // connect attempt resets the clock and must re-arm the timer on success.
+ if (this.stabilityTimer !== null) {
+ window.clearTimeout(this.stabilityTimer);
+ this.stabilityTimer = null;
+ }
+
this.connectionStateEmitter.set(
this.hasConnectedOnce ? "reconnecting" : "connecting",
);
@@ -538,7 +567,7 @@ export class RelayClient {
new Error("Timed out while waiting for relay authentication."),
);
reject(new Error("Timed out while waiting for relay authentication."));
- }, 8_000);
+ }, AUTH_TIMEOUT_MS);
this.authRequest = {
pendingEventId: "",
@@ -548,7 +577,15 @@ export class RelayClient {
};
});
- this.reconnectDelayMs = RECONNECT_BASE_DELAY_MS;
+ // Start a stability timer instead of resetting backoff immediately.
+ // The backoff resets to its base value only after BACKOFF_RESET_STABLE_MS
+ // of uninterrupted uptime, preventing fast reconnect loops from erasing
+ // the exponential backoff that throttles them.
+ this.stabilityTimer = window.setTimeout(() => {
+ this.stabilityTimer = null;
+ this.reconnectDelayMs = RECONNECT_BASE_DELAY_MS;
+ }, BACKOFF_RESET_STABLE_MS);
+
await this.replayLiveSubscriptions();
this.connectionStateEmitter.set("connected");
this.stallWatchdog.start();
@@ -665,16 +702,19 @@ export class RelayClient {
await this.sendRaw(["CLOSE", subId]);
}
- publishEvent(
+ async publishEvent(
event: RelayEvent,
timeoutMessage: string,
sendErrorMessage: string,
) {
+ // Await the gate before sending EVENT; op timeout starts after the wait.
+ await waitForRateLimit();
+
return new Promise((resolve, reject) => {
const timeout = window.setTimeout(() => {
this.pendingEvents.delete(event.id);
reject(new Error(timeoutMessage));
- }, 8_000);
+ }, PUBLISH_TIMEOUT_MS);
this.pendingEvents.set(event.id, {
event,
@@ -789,6 +829,16 @@ export class RelayClient {
"Failed to restore relay subscription after CLOSED.",
),
});
+ return;
+ }
+
+ if (type === "NOTICE" && typeof rest[0] === "string") {
+ const notice: string = rest[0];
+ // Relay back-pressure signal — activate the gate so pending operations
+ // back off until the window expires.
+ if (notice.startsWith("rate-limited:")) {
+ activateRateLimit(parseRateLimitHint(notice));
+ }
}
}
@@ -889,11 +939,14 @@ export class RelayClient {
}
private async replayLiveSubscriptions() {
+ const generation = this.connectionGeneration;
try {
await replayLiveSubscriptions({
subscriptions: this.subscriptions,
sendRaw: (payload) => this.sendRaw(payload),
requestHistory: (filter) => this.requestHistory(filter),
+ visibleChannelId: this.visibleChannelId,
+ isActive: () => this.connectionGeneration === generation,
});
} catch (error) {
const reconnectError =
@@ -918,7 +971,11 @@ export class RelayClient {
return;
}
- const delay = this.reconnectDelayMs;
+ // Apply ±25% jitter so a fleet of clients reconnecting simultaneously
+ // spreads their AUTH storms across a 50% window instead of all hitting
+ // the relay at the same instant.
+ const jitter = this.reconnectDelayMs * (0.75 + Math.random() * 0.5);
+ const delay = Math.min(jitter, RECONNECT_MAX_DELAY_MS);
this.reconnectDelayMs = Math.min(
this.reconnectDelayMs * 2,
RECONNECT_MAX_DELAY_MS,
@@ -961,6 +1018,10 @@ export class RelayClient {
this.onMessageChannel = null;
this.stallWatchdog.stop();
this.connectionGeneration++;
+ if (this.stabilityTimer !== null) {
+ window.clearTimeout(this.stabilityTimer);
+ this.stabilityTimer = null;
+ }
if (this.flushTimeout !== null) window.clearTimeout(this.flushTimeout);
this.flushTimeout = null;
this.eventBuffer = [];
diff --git a/desktop/src/shared/api/relayClosedPolicy.test.mjs b/desktop/src/shared/api/relayClosedPolicy.test.mjs
index e75cf1861e..715df77f6a 100644
--- a/desktop/src/shared/api/relayClosedPolicy.test.mjs
+++ b/desktop/src/shared/api/relayClosedPolicy.test.mjs
@@ -1,20 +1,21 @@
import assert from "node:assert/strict";
import test from "node:test";
-import { isRetryableRelayClosed } from "./relayClosedPolicy.ts";
+import { classifyRelayClosed } from "./relayClosedPolicy.ts";
-test("retries transient CLOSED responses", () => {
+// ── classifyRelayClosed ───────────────────────────────────────────────────────
+
+test("classifyRelayClosed: rate-limited messages return rate-limited", () => {
for (const message of [
+ "rate-limited: quota exceeded; retry in 4s",
"rate-limited: slow down",
- "error: database error",
- "server shutting down",
- "",
+ "rate-limited:",
]) {
- assert.equal(isRetryableRelayClosed(message), true, message);
+ assert.equal(classifyRelayClosed(message), "rate-limited", message);
}
});
-test("does not retry permanent CLOSED responses", () => {
+test("classifyRelayClosed: terminal messages return terminal", () => {
for (const message of [
"restricted: not a channel member",
"restricted: channel access revoked",
@@ -27,6 +28,46 @@ test("does not retry permanent CLOSED responses", () => {
"error: mixed search and non-search filters not supported",
"error: too many subscriptions",
]) {
- assert.equal(isRetryableRelayClosed(message), false, message);
+ assert.equal(classifyRelayClosed(message), "terminal", message);
+ }
+});
+
+test("classifyRelayClosed: transient errors return retryable", () => {
+ for (const message of ["error: database error", "server shutting down", ""]) {
+ assert.equal(classifyRelayClosed(message), "retryable", message);
+ }
+});
+
+// ── Subscription-survival semantics ──────────────────────────────────────────
+// These replace the removed isRetryableRelayClosed wrapper tests.
+// rate-limited must not delete the subscription; terminal must.
+
+test("classifyRelayClosed: rate-limited class survives (subscription must not be deleted)", () => {
+ // Subscription deletion is gated on === "terminal"; rate-limited must survive.
+ assert.notEqual(
+ classifyRelayClosed("rate-limited: quota exceeded; retry in 4s"),
+ "terminal",
+ );
+});
+
+test("classifyRelayClosed: retryable class survives (subscription must not be deleted)", () => {
+ for (const message of ["error: database error", "server shutting down", ""]) {
+ assert.notEqual(classifyRelayClosed(message), "terminal", message);
+ }
+});
+
+test("classifyRelayClosed: terminal class triggers deletion (no retry)", () => {
+ for (const message of [
+ "restricted: not a channel member",
+ "auth-required: not authenticated",
+ "blocked: banned",
+ "invalid: malformed filter",
+ "pow: difficulty too low",
+ "duplicate: subscription exists",
+ "unsupported: filter",
+ "error: mixed search and non-search filters not supported",
+ "error: too many subscriptions",
+ ]) {
+ assert.equal(classifyRelayClosed(message), "terminal", message);
}
});
diff --git a/desktop/src/shared/api/relayClosedPolicy.ts b/desktop/src/shared/api/relayClosedPolicy.ts
index f4257ed7a5..8a39e8a4d5 100644
--- a/desktop/src/shared/api/relayClosedPolicy.ts
+++ b/desktop/src/shared/api/relayClosedPolicy.ts
@@ -3,9 +3,23 @@
* changing the request; authorization and malformed-filter failures require a
* caller/state change and would otherwise loop forever.
*/
-export function isRetryableRelayClosed(message: string) {
+
+/**
+ * Three-way classification of a relay CLOSED message.
+ *
+ * - `"retryable"` — transient failure; re-send the REQ after a backoff.
+ * - `"rate-limited"` — relay back-pressure; activate the rate-limit gate and
+ * retry when the gate expires (subscription survives).
+ * - `"terminal"` — auth, access, or filter error; delete the subscription.
+ */
+export type RelayClosedClass = "retryable" | "rate-limited" | "terminal";
+
+export function classifyRelayClosed(message: string): RelayClosedClass {
const normalized = message.trim().toLowerCase();
- return !(
+ if (normalized.startsWith("rate-limited:")) {
+ return "rate-limited";
+ }
+ if (
normalized.startsWith("restricted:") ||
normalized.startsWith("auth-required:") ||
normalized.startsWith("blocked:") ||
@@ -15,5 +29,8 @@ export function isRetryableRelayClosed(message: string) {
normalized.startsWith("unsupported:") ||
normalized.startsWith("error: mixed search") ||
normalized.startsWith("error: too many subscriptions")
- );
+ ) {
+ return "terminal";
+ }
+ return "retryable";
}
diff --git a/desktop/src/shared/api/relayClosedRecovery.test.mjs b/desktop/src/shared/api/relayClosedRecovery.test.mjs
index 9c9b59440f..90acd6cb2f 100644
--- a/desktop/src/shared/api/relayClosedRecovery.test.mjs
+++ b/desktop/src/shared/api/relayClosedRecovery.test.mjs
@@ -2,11 +2,65 @@ import assert from "node:assert/strict";
import test from "node:test";
import { handleRelayClosed } from "./relayClosedRecovery.ts";
+import { requestHistoryGated } from "./relayGateBoundary.ts";
+
+// ── Fake-timer setup ──────────────────────────────────────────────────────────
+// The rate-limit gate and closed-retry logic use window.setTimeout/clearTimeout.
+
+let fakeNow = 0;
+const pendingTimers = new Map();
+let nextTimerId = 1;
+
+function fakeSetTimeout(fn, ms) {
+ const id = nextTimerId++;
+ pendingTimers.set(id, { fn, fireAt: fakeNow + ms });
+ return id;
+}
+
+function fakeClearTimeout(id) {
+ pendingTimers.delete(id);
+}
+
+function tickTo(ms) {
+ fakeNow = ms;
+ for (const [id, { fn, fireAt }] of Array.from(pendingTimers.entries())) {
+ if (fireAt <= fakeNow) {
+ pendingTimers.delete(id);
+ fn();
+ }
+ }
+}
+
+const origDateNow = Date.now;
+function setFakeNow(ms) {
+ fakeNow = ms;
+ Date.now = () => fakeNow;
+}
+
+globalThis.window = {
+ setTimeout: fakeSetTimeout,
+ clearTimeout: fakeClearTimeout,
+};
+
+// Import gate after window shim is installed.
+const { activateRateLimit, isRateLimited, resetRateLimitGate } = await import(
+ "./relayRateLimitGate.ts"
+);
+
+function resetAll(startMs = 0) {
+ pendingTimers.clear();
+ nextTimerId = 1;
+ setFakeNow(startMs);
+ resetRateLimitGate();
+}
test("production CLOSED handler rejects history once and clears its timeout", () => {
const originalWindow = globalThis.window;
const clearedTimeouts = [];
globalThis.window = {
+ // Provide both setTimeout (needed by activateRateLimit in the F1 fix) and
+ // clearTimeout (the existing assertion target).
+ setTimeout: (_fn, _ms) => 0,
clearTimeout: (timeout) => clearedTimeouts.push(timeout),
};
try {
@@ -45,6 +99,134 @@ test("production CLOSED handler rejects history once and clears its timeout", ()
}
});
+test("rate-limited history CLOSED arms the shared gate for concurrent ops", () => {
+ resetAll(0);
+ const subscriptions = new Map([
+ [
+ "history-1",
+ {
+ mode: "history",
+ events: [],
+ resolve: () => {},
+ reject: () => {},
+ timeout: 0,
+ },
+ ],
+ ]);
+ handleRelayClosed({
+ subscriptions,
+ subId: "history-1",
+ message: "rate-limited: quota exceeded; retry in 5s",
+ sendReq: () => Promise.resolve(),
+ });
+ assert.equal(
+ isRateLimited(),
+ true,
+ "gate must be active after rate-limited history CLOSED",
+ );
+ // Gate expires after the hint duration.
+ tickTo(5_001);
+ assert.equal(isRateLimited(), false, "gate must clear after hint duration");
+});
+
+test("non-rate-limited history CLOSED does not arm the gate", () => {
+ resetAll(0);
+ const subscriptions = new Map([
+ [
+ "history-2",
+ {
+ mode: "history",
+ events: [],
+ resolve: () => {},
+ reject: () => {},
+ timeout: 0,
+ },
+ ],
+ ]);
+ handleRelayClosed({
+ subscriptions,
+ subId: "history-2",
+ message: "error: database unavailable",
+ sendReq: () => Promise.resolve(),
+ });
+ assert.equal(
+ isRateLimited(),
+ false,
+ "gate must remain inactive for non-rate-limited history CLOSED",
+ );
+});
+
+test("gate armed by rate-limited history CLOSED defers the next REQ until expiry then resumes", async () => {
+ // Simulate: rate-limited CLOSED arrives on a history sub → gate arms for 5s.
+ // A concurrent requestHistoryGated call must not issue the REQ before 5s,
+ // and must issue it (and resolve) once the gate clears.
+ resetAll(0);
+
+ const subscriptions = new Map([
+ [
+ "history-gate",
+ {
+ mode: "history",
+ events: [],
+ resolve: () => {},
+ reject: () => {},
+ timeout: 0,
+ },
+ ],
+ ]);
+
+ // Arm the gate via a rate-limited history CLOSED.
+ handleRelayClosed({
+ subscriptions,
+ subId: "history-gate",
+ message: "rate-limited: quota exceeded; retry in 5s",
+ sendReq: () => Promise.resolve(),
+ });
+
+ assert.equal(isRateLimited(), true, "gate must be armed before the test");
+
+ const sentAt = [];
+ const reqSubscriptions = new Map();
+
+ // requestHistoryGated will await the gate, so the REQ must not fire at t=0.
+ const historyPromise = requestHistoryGated(
+ reqSubscriptions,
+ async (payload) => {
+ // Record when the REQ fires. The test harness sets up the EOSE path by
+ // adding a history subscription to reqSubscriptions immediately after
+ // the REQ is recorded, then resolving it.
+ sentAt.push(fakeNow);
+ // Resolve the returned promise by completing the sub synchronously.
+ const subId = payload[1];
+ const sub = reqSubscriptions.get(subId);
+ if (sub) {
+ window.clearTimeout(sub.timeout);
+ reqSubscriptions.delete(subId);
+ sub.resolve([]);
+ }
+ },
+ async () => {},
+ { kinds: [9], "#h": ["ch-test"], limit: 50 },
+ 25_000,
+ );
+
+ // REQ must not have fired yet — gate is still active at t=0.
+ await Promise.resolve();
+ assert.equal(sentAt.length, 0, "REQ must not fire while gate is active");
+
+ // Expire the gate — the deferred REQ should fire.
+ tickTo(5_001);
+
+ await historyPromise;
+
+ assert.equal(
+ sentAt.length,
+ 1,
+ "REQ must fire exactly once after gate clears",
+ );
+ assert.ok(sentAt[0] >= 5_001, "REQ must fire only after gate expiry");
+});
+
test("production CLOSED handler removes terminal live subscriptions", () => {
let readyCalls = 0;
const subscriptions = new Map([
@@ -69,3 +251,173 @@ test("production CLOSED handler removes terminal live subscriptions", () => {
assert.equal(subscriptions.has("live-1"), false);
assert.equal(readyCalls, 1);
});
+
+// ── Rate-limited CLOSED core behaviour (F5) ───────────────────────────────────
+
+test("rate-limited CLOSED keeps live subscription in the map", () => {
+ resetAll(0);
+ const subscriptions = new Map([
+ [
+ "live-1",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-1"], limit: 50 },
+ onEvent: () => {},
+ resolveReady: () => {},
+ },
+ ],
+ ]);
+ handleRelayClosed({
+ subscriptions,
+ subId: "live-1",
+ message: "rate-limited: quota exceeded; retry in 5s",
+ sendReq: () => Promise.resolve(),
+ });
+ assert.equal(
+ subscriptions.has("live-1"),
+ true,
+ "subscription must survive rate-limited CLOSED",
+ );
+});
+
+test("rate-limited CLOSED activates the rate-limit gate with the parsed hint", () => {
+ resetAll(0);
+ const subscriptions = new Map([
+ [
+ "live-1",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-1"], limit: 50 },
+ onEvent: () => {},
+ resolveReady: () => {},
+ },
+ ],
+ ]);
+ handleRelayClosed({
+ subscriptions,
+ subId: "live-1",
+ message: "rate-limited: quota exceeded; retry in 5s",
+ sendReq: () => Promise.resolve(),
+ });
+ assert.equal(
+ isRateLimited(),
+ true,
+ "gate must be active after rate-limited CLOSED",
+ );
+ // Gate should expire at 5s.
+ tickTo(5_001);
+ assert.equal(isRateLimited(), false);
+});
+
+test("rate-limited CLOSED retry delay is max(backoff, gate remaining), not just hint", () => {
+ resetAll(0);
+ // Activate a long gate first (20s), then send a shorter-hint CLOSED (5s).
+ // The retry delay must use the gate remaining time (20s), not the hint (5s).
+ activateRateLimit(20); // gate expires at 20_000 ms
+
+ const firedAt = [];
+ const subscriptions = new Map([
+ [
+ "live-1",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-1"], limit: 50 },
+ onEvent: () => {},
+ resolveReady: () => {},
+ },
+ ],
+ ]);
+ handleRelayClosed({
+ subscriptions,
+ subId: "live-1",
+ message: "rate-limited: quota exceeded; retry in 5s",
+ sendReq: () => {
+ firedAt.push(fakeNow);
+ return Promise.resolve();
+ },
+ });
+
+ // Retry should NOT fire at 5s (hint) — the gate remaining is 20s.
+ tickTo(5_001);
+ assert.equal(
+ firedAt.length,
+ 0,
+ "retry must not fire before gate remaining time",
+ );
+
+ // Should fire at 20s.
+ tickTo(20_001);
+ assert.equal(firedAt.length, 1, "retry must fire after gate remaining time");
+});
+
+test("non-rate-limited retryable CLOSED still schedules a retry", () => {
+ resetAll(0);
+ const firedAt = [];
+ const subscriptions = new Map([
+ [
+ "live-1",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-1"], limit: 50 },
+ onEvent: () => {},
+ resolveReady: () => {},
+ },
+ ],
+ ]);
+ handleRelayClosed({
+ subscriptions,
+ subId: "live-1",
+ message: "error: database error",
+ sendReq: () => {
+ firedAt.push(fakeNow);
+ return Promise.resolve();
+ },
+ });
+ // Base delay is 1s for first attempt.
+ tickTo(1_001);
+ assert.equal(firedAt.length, 1, "retryable CLOSED must schedule a retry");
+ assert.equal(
+ subscriptions.has("live-1"),
+ true,
+ "subscription must survive retryable CLOSED",
+ );
+});
+
+test("terminal CLOSED deletes subscription and does not retry", () => {
+ resetAll(0);
+ const firedAt = [];
+ const subscriptions = new Map([
+ [
+ "live-1",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-1"], limit: 50 },
+ onEvent: () => {},
+ resolveReady: () => {},
+ },
+ ],
+ ]);
+ handleRelayClosed({
+ subscriptions,
+ subId: "live-1",
+ message: "restricted: not a member",
+ sendReq: () => {
+ firedAt.push(fakeNow);
+ return Promise.resolve();
+ },
+ });
+ assert.equal(
+ subscriptions.has("live-1"),
+ false,
+ "terminal CLOSED must delete subscription",
+ );
+ tickTo(10_000);
+ assert.equal(firedAt.length, 0, "terminal CLOSED must not retry");
+});
+
+// ── Teardown ──────────────────────────────────────────────────────────────────
+
+test("teardown — restore Date.now", () => {
+ Date.now = origDateNow;
+ assert.ok(true);
+});
diff --git a/desktop/src/shared/api/relayClosedRecovery.ts b/desktop/src/shared/api/relayClosedRecovery.ts
index e8038bb71c..7206014eef 100644
--- a/desktop/src/shared/api/relayClosedRecovery.ts
+++ b/desktop/src/shared/api/relayClosedRecovery.ts
@@ -1,4 +1,9 @@
-import { isRetryableRelayClosed } from "@/shared/api/relayClosedPolicy";
+import { classifyRelayClosed } from "@/shared/api/relayClosedPolicy";
+import {
+ activateRateLimit,
+ parseRateLimitHint,
+ rateLimitRemainingMs,
+} from "@/shared/api/relayRateLimitGate";
import {
sortEvents,
type RelaySubscription,
@@ -31,6 +36,14 @@ export function handleRelayClosed({
const subscription = subscriptions.get(subId);
if (!subscription) return;
if (subscription.mode === "history") {
+ // Classify before rejecting so a `rate-limited:` history CLOSED arms the
+ // gate for concurrent ops. A history sub can't be retried (the caller holds
+ // the promise), so we still reject immediately after arming.
+ const closedClass = classifyRelayClosed(message);
+ if (closedClass === "rate-limited") {
+ const hintSeconds = parseRateLimitHint(message);
+ activateRateLimit(hintSeconds);
+ }
window.clearTimeout(subscription.timeout);
subscriptions.delete(subId);
subscription.reject(
@@ -62,17 +75,38 @@ function recoverLiveSubscriptionFromClosed({
}) {
subscription.resolveReady?.();
subscription.resolveReady = undefined;
- if (!isRetryableRelayClosed(message)) {
+
+ const closedClass = classifyRelayClosed(message);
+
+ if (closedClass === "terminal") {
+ // Auth/access/filter failure — permanently remove the subscription so it
+ // doesn't silently loop.
subscriptions.delete(subId);
return;
}
+
if (subscription.closedRetryTimeout !== undefined) return;
const attempt = subscription.closedRetryAttempt ?? 0;
- const delayMs = Math.min(
+ const backoffMs = Math.min(
RETRY_BASE_DELAY_MS * 2 ** attempt,
RETRY_MAX_DELAY_MS,
);
+
+ let delayMs = backoffMs;
+
+ if (closedClass === "rate-limited") {
+ // Activate the gate so concurrent operations back off too.
+ const hintSeconds = parseRateLimitHint(message);
+ activateRateLimit(hintSeconds);
+ // Use the gate's actual remaining time so a shorter hint arriving under a
+ // longer active gate does not schedule a premature retry that just gets
+ // another CLOSED. The fallback covers the gate-inactive edge case
+ // (hint * 1000, or 10s default when no hint).
+ const fallbackMs = (hintSeconds ?? 10) * 1_000;
+ delayMs = Math.max(backoffMs, rateLimitRemainingMs() || fallbackMs);
+ }
+
subscription.closedRetryAttempt = attempt + 1;
subscription.closedRetryTimeout = window.setTimeout(() => {
subscription.closedRetryTimeout = undefined;
diff --git a/desktop/src/shared/api/relayGateBoundary.ts b/desktop/src/shared/api/relayGateBoundary.ts
new file mode 100644
index 0000000000..99782358f2
--- /dev/null
+++ b/desktop/src/shared/api/relayGateBoundary.ts
@@ -0,0 +1,56 @@
+/**
+ * Gate-aware relay operation boundaries.
+ *
+ * History REQ operations are issued here rather than inline in RelayClient so
+ * the gate-await + op-timeout pattern lives in one place and the op timeout
+ * budget starts only after the rate-limit window has cleared.
+ */
+import { waitForRateLimit } from "@/shared/api/relayRateLimitGate";
+import type {
+ RelaySubscription,
+ RelaySubscriptionFilter,
+} from "@/shared/api/relayClientShared";
+import type { RelayEvent } from "@/shared/api/types";
+
+/**
+ * Issue a history REQ on `filter`, waiting for any active rate-limit gate
+ * before starting the subscription so the op timeout begins only after
+ * back-pressure has cleared.
+ */
+export async function requestHistoryGated(
+ subscriptions: Map,
+ sendRaw: (payload: unknown[]) => Promise,
+ closeSubscription: (subId: string) => Promise,
+ filter: RelaySubscriptionFilter,
+ historyTimeoutMs: number,
+): Promise {
+ // Await the gate before issuing REQ; op timeout starts after the wait.
+ await waitForRateLimit();
+
+ return new Promise((resolve, reject) => {
+ const subId = `history-${crypto.randomUUID()}`;
+ const timeout = window.setTimeout(() => {
+ subscriptions.delete(subId);
+ void closeSubscription(subId);
+ reject(new Error("Timed out while loading channel history."));
+ }, historyTimeoutMs);
+
+ subscriptions.set(subId, {
+ mode: "history",
+ events: [],
+ resolve,
+ reject,
+ timeout,
+ });
+
+ void sendRaw(["REQ", subId, filter]).catch((error) => {
+ window.clearTimeout(timeout);
+ subscriptions.delete(subId);
+ reject(
+ error instanceof Error
+ ? error
+ : new Error("Failed to request channel history."),
+ );
+ });
+ });
+}
diff --git a/desktop/src/shared/api/relayRateLimitGate.test.mjs b/desktop/src/shared/api/relayRateLimitGate.test.mjs
new file mode 100644
index 0000000000..a458e87b48
--- /dev/null
+++ b/desktop/src/shared/api/relayRateLimitGate.test.mjs
@@ -0,0 +1,351 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+// ── Fake-timer setup ──────────────────────────────────────────────────────────
+
+// The gate uses window.setTimeout/clearTimeout. We need a controllable fake
+// before the module is loaded. Set up globalThis.window with a fake timer
+// implementation and then dynamically import the module under test.
+
+let fakeNow = 0;
+const pendingTimers = new Map(); // id → { fn, fireAt }
+let nextTimerId = 1;
+
+function fakeSetTimeout(fn, ms) {
+ const id = nextTimerId++;
+ pendingTimers.set(id, { fn, fireAt: fakeNow + ms });
+ return id;
+}
+
+function fakeClearTimeout(id) {
+ pendingTimers.delete(id);
+}
+
+function tickTo(ms) {
+ fakeNow = ms;
+ for (const [id, { fn, fireAt }] of Array.from(pendingTimers.entries())) {
+ if (fireAt <= fakeNow) {
+ pendingTimers.delete(id);
+ fn();
+ }
+ }
+}
+
+// Install window shim before any imports use it.
+globalThis.window = {
+ setTimeout: fakeSetTimeout,
+ clearTimeout: fakeClearTimeout,
+};
+
+// Patch Date.now for the gate.
+const origDateNow = Date.now;
+function setFakeNow(ms) {
+ fakeNow = ms;
+ Date.now = () => fakeNow;
+}
+
+// Import after window is set up.
+const {
+ activateRateLimit,
+ isRateLimited,
+ waitForRateLimit,
+ resetRateLimitGate,
+ rateLimitRemainingMs,
+ parseRateLimitHint,
+ MAX_HINT_SECONDS,
+} = await import("./relayRateLimitGate.ts");
+
+// Helper to reset between tests.
+function reset(startMs = 0) {
+ pendingTimers.clear();
+ nextTimerId = 1;
+ setFakeNow(startMs);
+ resetRateLimitGate();
+}
+
+// ── parseRateLimitHint ────────────────────────────────────────────────────────
+
+test("parseRateLimitHint extracts seconds from CLOSED message", () => {
+ assert.equal(
+ parseRateLimitHint("rate-limited: quota exceeded; retry in 4s"),
+ 4,
+ );
+});
+
+test("parseRateLimitHint extracts seconds from HTTP 429 prefix", () => {
+ assert.equal(parseRateLimitHint("relay rate-limited: retry in 30s"), 30);
+});
+
+test("parseRateLimitHint returns null when no hint present", () => {
+ assert.equal(parseRateLimitHint("rate-limited: quota exceeded"), null);
+ assert.equal(parseRateLimitHint(""), null);
+ assert.equal(parseRateLimitHint("some other message"), null);
+});
+
+// ── isRateLimited / activate ──────────────────────────────────────────────────
+
+test("not rate-limited before any activation", () => {
+ reset();
+ assert.equal(isRateLimited(), false);
+});
+
+test("rate-limited immediately after activation", () => {
+ reset(0);
+ activateRateLimit(10);
+ assert.equal(isRateLimited(), true);
+});
+
+test("rate-limit expires when timer fires", () => {
+ reset(0);
+ activateRateLimit(10);
+ tickTo(10_001);
+ assert.equal(isRateLimited(), false);
+});
+
+test("activation extends expiry when new hint is longer than existing", () => {
+ reset(0);
+ activateRateLimit(5); // expires at 5000
+ activateRateLimit(20); // should extend to 20000
+ tickTo(5_001);
+ // Gate should still be active because the longer window was applied.
+ assert.equal(isRateLimited(), true);
+ tickTo(20_001);
+ assert.equal(isRateLimited(), false);
+});
+
+test("shorter hint does not shrink an existing longer window", () => {
+ reset(0);
+ activateRateLimit(20); // expires at 20000
+ activateRateLimit(5); // shorter — should NOT shrink
+ tickTo(5_001);
+ assert.equal(isRateLimited(), true);
+ tickTo(20_001);
+ assert.equal(isRateLimited(), false);
+});
+
+test("null hint uses 10s default", () => {
+ reset(0);
+ activateRateLimit(null);
+ tickTo(9_999);
+ assert.equal(isRateLimited(), true);
+ tickTo(10_001);
+ assert.equal(isRateLimited(), false);
+});
+
+test("zero hint uses 10s default (0s gate would be swallowed)", () => {
+ reset(0);
+ activateRateLimit(0);
+ tickTo(9_999);
+ assert.equal(isRateLimited(), true);
+ tickTo(10_001);
+ assert.equal(isRateLimited(), false);
+});
+
+test("negative hint uses 10s default", () => {
+ reset(0);
+ activateRateLimit(-5);
+ tickTo(9_999);
+ assert.equal(isRateLimited(), true);
+ tickTo(10_001);
+ assert.equal(isRateLimited(), false);
+});
+
+// ── MAX_HINT_SECONDS cap ──────────────────────────────────────────────────────
+
+test("oversized hint is clamped to MAX_HINT_SECONDS", () => {
+ reset(0);
+ // A relay sending 1 000 000s must not pin the gate beyond 300s.
+ activateRateLimit(1_000_000);
+ // Gate must still be active at MAX_HINT_SECONDS - 1 ms.
+ tickTo(MAX_HINT_SECONDS * 1_000 - 1);
+ assert.equal(
+ isRateLimited(),
+ true,
+ "gate must be active just before cap expiry",
+ );
+ // Gate must expire at MAX_HINT_SECONDS.
+ tickTo(MAX_HINT_SECONDS * 1_000 + 1);
+ assert.equal(isRateLimited(), false, "gate must expire at MAX_HINT_SECONDS");
+});
+
+test("hint exactly at MAX_HINT_SECONDS is honoured without clamping", () => {
+ reset(0);
+ activateRateLimit(MAX_HINT_SECONDS);
+ tickTo(MAX_HINT_SECONDS * 1_000 - 1);
+ assert.equal(isRateLimited(), true);
+ tickTo(MAX_HINT_SECONDS * 1_000 + 1);
+ assert.equal(isRateLimited(), false);
+});
+
+test("applyTauriRateLimitIfNeeded with oversized hint clamps to MAX_HINT_SECONDS", async () => {
+ // This test imports applyTauriRateLimitIfNeeded separately and verifies that
+ // the TS cap applies even when the message string contains a large hint value
+ // (the Rust layer clamps in practice, but TS must be independently safe).
+ reset(0);
+ const { applyTauriRateLimitIfNeeded } = await import("./tauri.ts");
+ // Simulate a message that somehow escaped the Rust cap (defence-in-depth).
+ applyTauriRateLimitIfNeeded("relay rate-limited: retry in 1000000s");
+ // Gate should cap at MAX_HINT_SECONDS * 1000 ms.
+ tickTo(MAX_HINT_SECONDS * 1_000 - 1);
+ assert.equal(
+ isRateLimited(),
+ true,
+ "gate must still be active just before cap",
+ );
+ tickTo(MAX_HINT_SECONDS * 1_000 + 1);
+ assert.equal(
+ isRateLimited(),
+ false,
+ "gate must expire at MAX_HINT_SECONDS, not 1 000 000s",
+ );
+});
+
+// ── waitForRateLimit ──────────────────────────────────────────────────────────
+
+test("waitForRateLimit resolves immediately when not rate-limited", async () => {
+ reset();
+ let resolved = false;
+ const p = waitForRateLimit().then(() => {
+ resolved = true;
+ });
+ await p;
+ assert.equal(resolved, true);
+});
+
+test("waitForRateLimit resolves after timer fires", async () => {
+ reset(0);
+ activateRateLimit(5);
+
+ let resolved = false;
+ const p = waitForRateLimit().then(() => {
+ resolved = true;
+ });
+
+ // Should not have resolved yet.
+ await Promise.resolve();
+ assert.equal(resolved, false);
+
+ // Fire the timer.
+ tickTo(5_001);
+
+ await p;
+ assert.equal(resolved, true);
+});
+
+test("multiple waiters all resolve when gate expires", async () => {
+ reset(0);
+ activateRateLimit(5);
+
+ const results = [];
+ const promises = [1, 2, 3].map((id) =>
+ waitForRateLimit().then(() => results.push(id)),
+ );
+
+ await Promise.resolve();
+ assert.equal(results.length, 0);
+
+ tickTo(5_001);
+ await Promise.all(promises);
+ assert.deepEqual(results, [1, 2, 3]);
+});
+
+// ── resetRateLimitGate ────────────────────────────────────────────────────────
+
+test("resetRateLimitGate clears an active gate immediately", () => {
+ reset(0);
+ activateRateLimit(30);
+ assert.equal(isRateLimited(), true);
+
+ resetRateLimitGate();
+ assert.equal(isRateLimited(), false);
+});
+
+test("resetRateLimitGate clears the timer so it does not fire later", () => {
+ reset(0);
+ activateRateLimit(10);
+ const timerCountBefore = pendingTimers.size;
+ assert.equal(timerCountBefore, 1);
+
+ resetRateLimitGate();
+ assert.equal(pendingTimers.size, 0);
+});
+
+test("in-flight waitForRateLimit resolves when resetRateLimitGate is called", async () => {
+ reset(0);
+ activateRateLimit(30);
+
+ let inFlightResolved = false;
+ const inFlight = waitForRateLimit().then(() => {
+ inFlightResolved = true;
+ });
+
+ // Not yet resolved before reset.
+ await Promise.resolve();
+ assert.equal(inFlightResolved, false);
+
+ // reset resolves the in-flight awaiter immediately.
+ resetRateLimitGate();
+ await inFlight;
+ assert.equal(inFlightResolved, true);
+
+ // Gate is now clear.
+ assert.equal(isRateLimited(), false);
+
+ // A new wait should also resolve immediately.
+ await waitForRateLimit();
+});
+
+// ── rateLimitRemainingMs ──────────────────────────────────────────────────────
+
+test("rateLimitRemainingMs returns 0 when gate is inactive", () => {
+ reset(0);
+ assert.equal(rateLimitRemainingMs(), 0);
+});
+
+test("rateLimitRemainingMs returns remaining ms when gate is active", () => {
+ reset(0);
+ activateRateLimit(10); // expires at 10_000 ms
+ setFakeNow(3_000);
+ assert.equal(rateLimitRemainingMs(), 7_000);
+});
+
+test("rateLimitRemainingMs returns 0 after gate expires", () => {
+ reset(0);
+ activateRateLimit(5);
+ tickTo(5_001);
+ assert.equal(rateLimitRemainingMs(), 0);
+});
+
+// ── NOTICE → gate integration ─────────────────────────────────────────────────
+// Verifies that the session NOTICE handler logic — a case-sensitive
+// `startsWith("rate-limited:")` check followed by activateRateLimit — correctly
+// arms the gate. The relay always emits lowercase prefixes.
+
+test("rate-limited: NOTICE prefix (case-sensitive) activates the gate", () => {
+ reset(0);
+ const notice = "rate-limited: quota exceeded; retry in 8s";
+ if (notice.startsWith("rate-limited:")) {
+ activateRateLimit(parseRateLimitHint(notice));
+ }
+ assert.equal(isRateLimited(), true);
+ tickTo(8_001);
+ assert.equal(isRateLimited(), false);
+});
+
+test("Rate-limited: NOTICE with uppercase prefix does NOT activate the gate (relay always emits lowercase)", () => {
+ reset(0);
+ const notice = "Rate-limited: quota exceeded; retry in 8s";
+ if (notice.startsWith("rate-limited:")) {
+ activateRateLimit(parseRateLimitHint(notice));
+ }
+ // Uppercase is not emitted by the relay — gate should remain inactive.
+ assert.equal(isRateLimited(), false);
+});
+
+// ── Cleanup ───────────────────────────────────────────────────────────────────
+
+// Restore Date.now after all tests to avoid polluting subsequent test files.
+test("teardown — restore Date.now", () => {
+ Date.now = origDateNow;
+ assert.ok(true);
+});
diff --git a/desktop/src/shared/api/relayRateLimitGate.ts b/desktop/src/shared/api/relayRateLimitGate.ts
new file mode 100644
index 0000000000..0af3eed7d9
--- /dev/null
+++ b/desktop/src/shared/api/relayRateLimitGate.ts
@@ -0,0 +1,137 @@
+/**
+ * Module-level rate-limit gate for the relay WebSocket and HTTP bridge.
+ *
+ * When the relay signals back-pressure via a CLOSED `rate-limited:` message or
+ * an HTTP 429 response, callers activate the gate. Operations that must not run
+ * while rate-limited call `isRateLimited()` or await `waitForRateLimit()`.
+ *
+ * The gate is a singleton: one shared expiry covers all concurrent callers so
+ * overlapping hints (multiple CLOSED frames) extend to the latest expiry without
+ * stacking. This module must be reset on community switch via
+ * `resetRateLimitGate()`.
+ */
+
+/** Minimum gate duration when the relay provides no `retry in Ns` hint. */
+const DEFAULT_RATE_LIMIT_SECONDS = 10;
+
+/**
+ * Maximum hint the TS gate will honour from a relay 429 response.
+ *
+ * Mirrors `MAX_HINT_SECONDS` in `relay_admission.rs` (Rust). The Rust relay
+ * layer clamps the hint before embedding it in the error string, so in practice
+ * this TS cap is a defence-in-depth guard against any future Rust path that
+ * forgets to clamp, keeping both gates on the same documented bound.
+ */
+export const MAX_HINT_SECONDS = 300;
+
+let expiresAt: number | null = null;
+let gateTimer: number | null = null;
+let gateResolve: (() => void) | null = null;
+let gatePromise: Promise | null = null;
+
+/**
+ * Parse `retry in Ns` from a relay message string.
+ *
+ * Matches the relay's canonical hint format embedded in both CLOSED messages
+ * (`rate-limited: quota exceeded; retry in 4s`) and HTTP 429 bodies
+ * (`relay rate-limited: retry in 4s`).
+ */
+export function parseRateLimitHint(msg: string): number | null {
+ const match = /retry in (\d+)s/i.exec(msg);
+ return match ? parseInt(match[1], 10) : null;
+}
+
+/**
+ * Activate (or extend) the rate-limit gate.
+ *
+ * If the gate is already active, the expiry is pushed forward to the maximum of
+ * the existing expiry and the new hint — overlapping hints never shrink the
+ * window. Non-positive or absent hints use the 10-second default; a 0s gate
+ * would resolve immediately and swallow the signal.
+ *
+ * Note: buzz-acp uses a 5s no-hint default; desktop deliberately uses 10s here
+ * for a wider back-off window on degraded connections.
+ */
+export function activateRateLimit(retryInSeconds: number | null): void {
+ const durationMs =
+ (retryInSeconds != null && retryInSeconds > 0
+ ? Math.min(retryInSeconds, MAX_HINT_SECONDS)
+ : DEFAULT_RATE_LIMIT_SECONDS) * 1_000;
+ const newExpiry = Date.now() + durationMs;
+
+ if (expiresAt !== null && newExpiry <= expiresAt) {
+ // Existing window already covers this hint — nothing to do.
+ return;
+ }
+
+ expiresAt = newExpiry;
+
+ if (gateTimer !== null) {
+ window.clearTimeout(gateTimer);
+ }
+
+ if (gatePromise === null) {
+ // First activation: create the shared promise that waiters will await.
+ gatePromise = new Promise((resolve) => {
+ gateResolve = resolve;
+ });
+ }
+
+ gateTimer = window.setTimeout(() => {
+ gateTimer = null;
+ expiresAt = null;
+ const resolve = gateResolve;
+ gateResolve = null;
+ gatePromise = null;
+ resolve?.();
+ }, durationMs);
+}
+
+/** Returns `true` when the relay has signalled back-pressure and the gate is active. */
+export function isRateLimited(): boolean {
+ return expiresAt !== null && Date.now() < expiresAt;
+}
+
+/**
+ * Resolves when the rate-limit gate expires.
+ *
+ * Resolves immediately if the gate is not active. Callers that need to wait
+ * before issuing a new relay request should await this before proceeding.
+ */
+export function waitForRateLimit(): Promise {
+ if (!isRateLimited() || gatePromise === null) {
+ return Promise.resolve();
+ }
+ return gatePromise;
+}
+
+/**
+ * Returns the milliseconds remaining on the active gate, or 0 when inactive.
+ *
+ * Use this instead of re-deriving the hint from the message so that a shorter
+ * relay hint arriving under a longer active gate never schedules a premature
+ * retry.
+ */
+export function rateLimitRemainingMs(): number {
+ if (expiresAt === null) return 0;
+ return Math.max(0, expiresAt - Date.now());
+}
+
+/**
+ * Reset all gate state. Must be called on community switch so a rate-limit hint
+ * from the old relay does not bleed into the new community's session.
+ *
+ * Any in-flight `waitForRateLimit()` awaiters are resolved immediately so they
+ * do not leak into the new session.
+ */
+export function resetRateLimitGate(): void {
+ if (gateTimer !== null) {
+ window.clearTimeout(gateTimer);
+ gateTimer = null;
+ }
+ expiresAt = null;
+ const resolve = gateResolve;
+ gateResolve = null;
+ gatePromise = null;
+ resolve?.();
+}
diff --git a/desktop/src/shared/api/relayReconnectReplay.test.mjs b/desktop/src/shared/api/relayReconnectReplay.test.mjs
index 63f6402cda..54254e89de 100644
--- a/desktop/src/shared/api/relayReconnectReplay.test.mjs
+++ b/desktop/src/shared/api/relayReconnectReplay.test.mjs
@@ -4,13 +4,61 @@ import test from "node:test";
import {
buildReconnectReplayFilter,
replayLiveSubscriptions,
+ REPLAY_BATCH_SIZE,
} from "./relayReconnectReplay.ts";
import { buildChannelFilter } from "./relayChannelFilters.ts";
-function replayFilter(filter, since, until) {
- return buildReconnectReplayFilter(filter, since, until);
+// ── Fake-timer + Date.now setup for gate tests ────────────────────────────────
+
+let fakeNow = 0;
+const pendingTimers = new Map();
+let nextTimerId = 1;
+
+function fakeSetTimeout(fn, ms) {
+ const id = nextTimerId++;
+ pendingTimers.set(id, { fn, fireAt: fakeNow + ms });
+ return id;
+}
+
+function fakeClearTimeout(id) {
+ pendingTimers.delete(id);
+}
+
+function tickTo(ms) {
+ fakeNow = ms;
+ for (const [id, { fn, fireAt }] of Array.from(pendingTimers.entries())) {
+ if (fireAt <= fakeNow) {
+ pendingTimers.delete(id);
+ fn();
+ }
+ }
}
+globalThis.window = {
+ setTimeout: fakeSetTimeout,
+ clearTimeout: fakeClearTimeout,
+};
+
+const origDateNow = Date.now;
+function setFakeNow(ms) {
+ fakeNow = ms;
+ Date.now = () => fakeNow;
+}
+
+// Import gate module AFTER window shim so it picks up the fake timers.
+const { activateRateLimit, resetRateLimitGate } = await import(
+ "./relayRateLimitGate.ts"
+);
+
+function resetGate(startMs = 0) {
+ pendingTimers.clear();
+ nextTimerId = 1;
+ setFakeNow(startMs);
+ resetRateLimitGate();
+}
+
+// ── Helpers ───────────────────────────────────────────────────────────────────
+
function event(id, createdAt) {
return {
id,
@@ -29,6 +77,12 @@ function eventRange(prefix, start, count) {
);
}
+function replayFilter(filter, since, until) {
+ return buildReconnectReplayFilter(filter, since, until);
+}
+
+// ── buildReconnectReplayFilter ────────────────────────────────────────────────
+
test("reconnect replay preserves small steady-state limits when adding since", () => {
const filter = {
kinds: [9, 40002],
@@ -102,7 +156,211 @@ test("initial subscription replay preserves the original filter", () => {
assert.equal(replayFilter(filter, undefined), filter);
});
+// ── Batching: REPLAY_BATCH_SIZE cap ──────────────────────────────────────────
+
+test("replay sends all subs in one batch when count equals REPLAY_BATCH_SIZE", async () => {
+ resetGate();
+ let delayCount = 0;
+ const sentIds = [];
+
+ const subscriptions = new Map(
+ Array.from({ length: REPLAY_BATCH_SIZE }, (_, i) => [
+ `sub-${i}`,
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": [`ch-${i}`], limit: 50 },
+ onEvent: () => {},
+ lastSeenCreatedAt: undefined,
+ },
+ ]),
+ );
+
+ await replayLiveSubscriptions({
+ subscriptions,
+ sendRaw: async (payload) => {
+ sentIds.push(payload[1]);
+ },
+ requestHistory: async () => [],
+ setTimeoutFn: (fn, _ms) => {
+ delayCount++;
+ fn();
+ return 0;
+ },
+ });
+
+ assert.equal(sentIds.length, REPLAY_BATCH_SIZE);
+ assert.equal(delayCount, 0, "no inter-batch delay for exactly one batch");
+});
+
+test("replay splits subscriptions into batches of REPLAY_BATCH_SIZE", async () => {
+ resetGate();
+ let delayCount = 0;
+ const sentIds = [];
+ const batchBreakpoints = []; // indices where a delay fired
+
+ const subCount = REPLAY_BATCH_SIZE + 3;
+ const subscriptions = new Map(
+ Array.from({ length: subCount }, (_, i) => [
+ `sub-${i}`,
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": [`ch-${i}`], limit: 50 },
+ onEvent: () => {},
+ lastSeenCreatedAt: undefined,
+ },
+ ]),
+ );
+
+ await replayLiveSubscriptions({
+ subscriptions,
+ sendRaw: async (payload) => {
+ sentIds.push(payload[1]);
+ },
+ requestHistory: async () => [],
+ setTimeoutFn: (fn, _ms) => {
+ delayCount++;
+ batchBreakpoints.push(sentIds.length);
+ fn();
+ return 0;
+ },
+ });
+
+ assert.equal(delayCount, 1, "one inter-batch delay between two batches");
+ assert.equal(sentIds.length, subCount, "all subs sent");
+ // The delay fired after the first batch (REPLAY_BATCH_SIZE subs sent).
+ assert.equal(batchBreakpoints[0], REPLAY_BATCH_SIZE);
+});
+
+// ── Visible-channel priority ──────────────────────────────────────────────────
+
+test("visible channel subscription is sent first", async () => {
+ resetGate();
+ const sentOrder = [];
+
+ const subscriptions = new Map([
+ [
+ "other-1",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-other"], limit: 50 },
+ onEvent: () => {},
+ lastSeenCreatedAt: undefined,
+ },
+ ],
+ [
+ "visible-sub",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-visible"], limit: 50 },
+ onEvent: () => {},
+ lastSeenCreatedAt: undefined,
+ },
+ ],
+ [
+ "other-2",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-other2"], limit: 50 },
+ onEvent: () => {},
+ lastSeenCreatedAt: undefined,
+ },
+ ],
+ ]);
+
+ await replayLiveSubscriptions({
+ subscriptions,
+ sendRaw: async (payload) => {
+ sentOrder.push(payload[1]);
+ },
+ requestHistory: async () => [],
+ visibleChannelId: "ch-visible",
+ });
+
+ assert.equal(sentOrder[0], "visible-sub", "visible sub sent first");
+ assert.equal(sentOrder.length, 3);
+});
+
+// ── Rate-limit gate deferral ──────────────────────────────────────────────────
+
+test("replay waits for rate-limit gate before sending REQs", async () => {
+ resetGate(0);
+ activateRateLimit(5); // gate active for 5 seconds
+
+ const sentIds = [];
+
+ const replayPromise = replayLiveSubscriptions({
+ subscriptions: new Map([
+ [
+ "sub-1",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-1"], limit: 50 },
+ onEvent: () => {},
+ lastSeenCreatedAt: undefined,
+ },
+ ],
+ ]),
+ sendRaw: async (payload) => {
+ sentIds.push(payload[1]);
+ },
+ requestHistory: async () => [],
+ setTimeoutFn: (fn, _ms) => {
+ fn();
+ return 0;
+ },
+ });
+
+ // Gate expires — replay should proceed now.
+ tickTo(5_001);
+
+ await replayPromise;
+
+ assert.equal(sentIds.length, 1, "REQ sent after gate expired");
+});
+
+// ── Connection-generation guard ───────────────────────────────────────────────
+
+test("stale replay sends no REQs when generation advances while gate was active", async () => {
+ resetGate(0);
+ activateRateLimit(5); // gate active for 5 seconds
+
+ let generationActive = true; // true = current, false = stale
+ const sentIds = [];
+
+ const replayPromise = replayLiveSubscriptions({
+ subscriptions: new Map([
+ [
+ "sub-1",
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": ["ch-1"], limit: 50 },
+ onEvent: () => {},
+ lastSeenCreatedAt: undefined,
+ },
+ ],
+ ]),
+ sendRaw: async (payload) => {
+ sentIds.push(payload[1]);
+ },
+ requestHistory: async () => [],
+ isActive: () => generationActive,
+ });
+
+ // Advance the generation (simulate new connection) before the gate resolves.
+ generationActive = false;
+
+ // Fire the gate timer.
+ tickTo(5_001);
+
+ await replayPromise;
+
+ assert.equal(sentIds.length, 0, "no REQs sent for a stale replay");
+});
+
+// ── Paged replay (existing behaviour) ────────────────────────────────────────
+
test("channel reconnect replay pages the missed window until a short page", async () => {
+ resetGate();
const delivered = [];
const historyFilters = [];
const sentPayloads = [];
@@ -174,6 +432,7 @@ test("channel reconnect replay pages the missed window until a short page", asyn
});
test("reconnect replay starts live REQs in parallel and preserves per-sub page order", async () => {
+ resetGate();
const sentPayloads = [];
const sendResolvers = [];
const historyFiltersByChannel = {
@@ -250,3 +509,74 @@ test("reconnect replay starts live REQs in parallel and preserves per-sub page o
"channel-2": [2000, 1701],
});
});
+
+// ── Per-batch gate re-check (F2 fix) ─────────────────────────────────────────
+
+test("batch-1 arms gate mid-replay: batch-2 is withheld until gate expires", async () => {
+ // Gate is inactive at the start of replay. Batch 1 fires and (simulating the
+ // relay's admission control) activates the gate. Batch 2 must wait until the
+ // gate clears before its REQs are sent.
+ resetGate(0);
+
+ const BATCH = REPLAY_BATCH_SIZE;
+ const sentAtMs = []; // record the fakeNow when each REQ fires
+
+ // Build BATCH+1 subscriptions so there are exactly two batches.
+ const subscriptions = new Map(
+ Array.from({ length: BATCH + 1 }, (_, i) => [
+ `sub-${i}`,
+ {
+ mode: "live",
+ filter: { kinds: [9], "#h": [`ch-${i}`], limit: 50 },
+ onEvent: () => {},
+ lastSeenCreatedAt: undefined,
+ },
+ ]),
+ );
+
+ let _batchCount = 0;
+ const replayPromise = replayLiveSubscriptions({
+ subscriptions,
+ sendRaw: async (payload) => {
+ sentAtMs.push({ id: payload[1], ms: fakeNow });
+ // After the first full batch is sent, arm the gate for 5 s.
+ // This simulates the relay responding to batch-1 traffic with back-pressure.
+ if (sentAtMs.length === BATCH) {
+ _batchCount += 1;
+ activateRateLimit(5);
+ }
+ },
+ requestHistory: async () => [],
+ setTimeoutFn: (fn, _ms) => {
+ fn();
+ return 0;
+ },
+ });
+
+ // Advance time to expire the gate while the replay is suspended in the
+ // per-batch gate await. This unblocks the second batch.
+ tickTo(5_001);
+
+ await replayPromise;
+
+ const batch1Ids = sentAtMs.filter((r) => r.ms < 5_001).map((r) => r.id);
+ const batch2Ids = sentAtMs.filter((r) => r.ms >= 5_001).map((r) => r.id);
+
+ assert.equal(
+ batch1Ids.length,
+ BATCH,
+ "batch 1 must send exactly REPLAY_BATCH_SIZE REQs",
+ );
+ assert.equal(
+ batch2Ids.length,
+ 1,
+ "batch 2 must send the remaining sub after the gate expires",
+ );
+});
+
+// ── Teardown ──────────────────────────────────────────────────────────────────
+
+test("teardown — restore Date.now", () => {
+ Date.now = origDateNow;
+ assert.ok(true);
+});
diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts
index 65709d351a..74b752f666 100644
--- a/desktop/src/shared/api/relayReconnectReplay.ts
+++ b/desktop/src/shared/api/relayReconnectReplay.ts
@@ -4,11 +4,31 @@ import type {
RelaySubscriptionFilter,
} from "@/shared/api/relayClientShared";
import type { RelayEvent } from "@/shared/api/types";
+import {
+ isRateLimited,
+ waitForRateLimit,
+} from "@/shared/api/relayRateLimitGate";
const RECONNECT_REPLAY_SKEW_SECS = 5;
export const RECONNECT_REPLAY_PAGE_LIMIT = 500;
export const RECONNECT_REPLAY_PAGE_CONCURRENCY = 4;
+/**
+ * Maximum live subscriptions sent per relay REQ burst during reconnect.
+ *
+ * Capping the initial blast prevents admission-control bursts on degraded
+ * networks where the relay is already near its per-pubkey quota.
+ */
+export const REPLAY_BATCH_SIZE = 8;
+
+/**
+ * Delay between consecutive replay batches (milliseconds).
+ *
+ * Spreads the REQ storm across time so the relay's sliding quota window
+ * can absorb each batch without triggering rate-limiting on the next.
+ */
+export const REPLAY_INTER_BATCH_DELAY_MS = 50;
+
async function runWithConcurrency(
items: T[],
concurrency: number,
@@ -104,13 +124,41 @@ export async function replayLiveSubscriptions({
requestHistory,
now = Math.floor(Date.now() / 1_000),
pageReplayConcurrency = RECONNECT_REPLAY_PAGE_CONCURRENCY,
+ visibleChannelId = null,
+ replayBatchSize = REPLAY_BATCH_SIZE,
+ interBatchDelayMs = REPLAY_INTER_BATCH_DELAY_MS,
+ setTimeoutFn = (fn: () => void, ms: number) =>
+ window.setTimeout(fn, ms) as unknown as number,
+ isActive = () => true,
}: {
subscriptions: Map;
sendRaw: (payload: unknown[]) => Promise;
requestHistory: (filter: RelaySubscriptionFilter) => Promise;
now?: number;
pageReplayConcurrency?: number;
+ /** Channel currently visible in the UI — its subscriptions go in the first batch. */
+ visibleChannelId?: string | null;
+ /** Max subscriptions per REQ burst (injectable for tests). */
+ replayBatchSize?: number;
+ /** Milliseconds between bursts (injectable for tests). */
+ interBatchDelayMs?: number;
+ /** setTimeout implementation (injectable for tests). */
+ setTimeoutFn?: (fn: () => void, ms: number) => number;
+ /**
+ * Returns false when the connection that initiated this replay has been
+ * superseded by a newer one. After the gate await resumes, a stale replay
+ * must not double-send REQs on the live socket.
+ */
+ isActive?: () => boolean;
}) {
+ // If the relay has signalled back-pressure, wait for the gate to clear
+ // before blasting a full set of REQs that would immediately be rate-limited.
+ if (isRateLimited()) await waitForRateLimit();
+
+ // A newer connection may have replayed while this one was suspended at the
+ // gate — abort silently to avoid double-sending every REQ on the live socket.
+ if (!isActive()) return;
+
const replayRequests = Array.from(subscriptions.entries())
.filter(
(
@@ -133,9 +181,35 @@ export async function replayLiveSubscriptions({
return { subId, subscription, replaySince, shouldPageReplay };
});
- await Promise.all(
- replayRequests.map(
- ({ subId, subscription, replaySince, shouldPageReplay }) =>
+ // Sort the visible channel's subscriptions first so the user sees their
+ // active channel recover before others on degraded networks.
+ if (visibleChannelId !== null) {
+ replayRequests.sort((a, b) => {
+ const aVis =
+ (a.subscription.filter["#h"] as string[] | undefined)?.includes(
+ visibleChannelId,
+ ) ?? false;
+ const bVis =
+ (b.subscription.filter["#h"] as string[] | undefined)?.includes(
+ visibleChannelId,
+ ) ?? false;
+ if (aVis === bVis) return 0;
+ return aVis ? -1 : 1;
+ });
+ }
+
+ // Send live REQs in capped batches with inter-batch delays to avoid
+ // triggering per-pubkey admission control on degraded/recovering connections.
+ for (let i = 0; i < replayRequests.length; i += replayBatchSize) {
+ // Re-check the gate before every batch: a previous batch may have triggered
+ // admission control and armed the gate mid-replay. Wait for it to clear,
+ // then verify the connection is still current — a newer connection may have
+ // replayed while we were suspended.
+ if (isRateLimited()) await waitForRateLimit();
+ if (!isActive()) return;
+ const batch = replayRequests.slice(i, i + replayBatchSize);
+ await Promise.all(
+ batch.map(({ subId, subscription, replaySince, shouldPageReplay }) =>
sendRaw([
"REQ",
subId,
@@ -143,8 +217,14 @@ export async function replayLiveSubscriptions({
? subscription.filter
: buildReconnectReplayFilter(subscription.filter, replaySince),
]),
- ),
- );
+ ),
+ );
+ if (i + replayBatchSize < replayRequests.length) {
+ await new Promise((resolve) =>
+ setTimeoutFn(resolve, interBatchDelayMs),
+ );
+ }
+ }
await runWithConcurrency(
replayRequests.filter(
diff --git a/desktop/src/shared/api/tauri.test.mjs b/desktop/src/shared/api/tauri.test.mjs
new file mode 100644
index 0000000000..68205d7298
--- /dev/null
+++ b/desktop/src/shared/api/tauri.test.mjs
@@ -0,0 +1,120 @@
+/**
+ * Unit tests for tauri.ts — focused on `applyTauriRateLimitIfNeeded`, the
+ * extracted `relay rate-limited:` classifier that activates the shared
+ * rate-limit gate when Rust emits an HTTP 429 error prefix.
+ *
+ * Testing the exported production function (not a local copy) ensures any
+ * change to the classifier logic is immediately covered here.
+ */
+import assert from "node:assert/strict";
+import test from "node:test";
+
+// ── Fake-timer + gate setup ───────────────────────────────────────────────────
+
+let fakeNow = 0;
+const pendingTimers = new Map();
+let nextTimerId = 1;
+
+function fakeSetTimeout(fn, ms) {
+ const id = nextTimerId++;
+ pendingTimers.set(id, { fn, fireAt: fakeNow + ms });
+ return id;
+}
+
+function fakeClearTimeout(id) {
+ pendingTimers.delete(id);
+}
+
+function tickTo(ms) {
+ fakeNow = ms;
+ for (const [id, { fn, fireAt }] of Array.from(pendingTimers.entries())) {
+ if (fireAt <= fakeNow) {
+ pendingTimers.delete(id);
+ fn();
+ }
+ }
+}
+
+const origDateNow = Date.now;
+function setFakeNow(ms) {
+ fakeNow = ms;
+ Date.now = () => fakeNow;
+}
+
+globalThis.window = {
+ setTimeout: fakeSetTimeout,
+ clearTimeout: fakeClearTimeout,
+};
+
+setFakeNow(0);
+
+const { isRateLimited, resetRateLimitGate } = await import(
+ "./relayRateLimitGate.ts"
+);
+
+// Import the production classifier from tauri.ts — tests must exercise the
+// real function, not a local copy, so a logic change is always caught here.
+const { applyTauriRateLimitIfNeeded } = await import("./tauri.ts");
+
+function resetGate(startMs = 0) {
+ pendingTimers.clear();
+ nextTimerId = 1;
+ setFakeNow(startMs);
+ resetRateLimitGate();
+}
+
+// ── applyTauriRateLimitIfNeeded: relay rate-limited: prefix ───────────────────
+
+test("relay rate-limited: prefix activates the rate-limit gate", () => {
+ resetGate(0);
+ applyTauriRateLimitIfNeeded("relay rate-limited: retry in 10s");
+ assert.equal(isRateLimited(), true, "gate must be active after 429 error");
+});
+
+test("relay rate-limited: prefix parses the retry hint and arms the gate duration", () => {
+ resetGate(0);
+ applyTauriRateLimitIfNeeded("relay rate-limited: retry in 7s");
+ // Gate should be active at 6s.
+ setFakeNow(6_000);
+ assert.equal(isRateLimited(), true);
+ // Gate should expire after 7s.
+ tickTo(7_001);
+ assert.equal(isRateLimited(), false);
+});
+
+test("relay rate-limited: with no hint uses the 10s default", () => {
+ resetGate(0);
+ applyTauriRateLimitIfNeeded("relay rate-limited: quota exceeded");
+ tickTo(9_999);
+ assert.equal(isRateLimited(), true);
+ tickTo(10_001);
+ assert.equal(isRateLimited(), false);
+});
+
+test("non-rate-limited error does not activate the gate", () => {
+ resetGate(0);
+ applyTauriRateLimitIfNeeded("relay returned 404 Not Found");
+ assert.equal(
+ isRateLimited(),
+ false,
+ "gate must remain inactive for unrelated errors",
+ );
+});
+
+test("relay rate-limited: prefix check is case-sensitive (Rust always emits lowercase)", () => {
+ resetGate(0);
+ // The prefix from Rust is always lowercase; mixed-case must not trigger it.
+ applyTauriRateLimitIfNeeded("Relay rate-limited: retry in 5s");
+ assert.equal(
+ isRateLimited(),
+ false,
+ "uppercase prefix must not activate gate (relay emits lowercase only)",
+ );
+});
+
+// ── Teardown ──────────────────────────────────────────────────────────────────
+
+test("teardown — restore Date.now", () => {
+ Date.now = origDateNow;
+ assert.ok(true);
+});
diff --git a/desktop/src/shared/api/tauri.ts b/desktop/src/shared/api/tauri.ts
index 46819bb906..add41deb24 100644
--- a/desktop/src/shared/api/tauri.ts
+++ b/desktop/src/shared/api/tauri.ts
@@ -1,4 +1,8 @@
import { invoke as tauriInvoke } from "@tauri-apps/api/core";
+import {
+ activateRateLimit,
+ parseRateLimitHint,
+} from "@/shared/api/relayRateLimitGate";
import type {
AddChannelMembersInput,
AddChannelMembersResult,
@@ -280,6 +284,18 @@ function toTauriError(error: unknown): Error {
}
}
+/**
+ * Inspect a Tauri error message and activate the shared rate-limit gate when
+ * the Rust relay layer emitted an HTTP 429 response (`relay rate-limited:` prefix).
+ *
+ * Extracted so it can be unit-tested without mocking the Tauri invoke bridge.
+ */
+export function applyTauriRateLimitIfNeeded(message: string): void {
+ if (message.startsWith("relay rate-limited:")) {
+ activateRateLimit(parseRateLimitHint(message));
+ }
+}
+
export async function invokeTauri(
command: string,
args?: Record,
@@ -287,7 +303,11 @@ export async function invokeTauri(
try {
return await tauriInvoke(command, args);
} catch (error) {
- throw toTauriError(error);
+ const err = toTauriError(error);
+ // Rust emits `relay rate-limited:` for HTTP 429 responses. Activate the
+ // shared gate so the TS relay client backs off for the same window.
+ applyTauriRateLimitIfNeeded(err.message);
+ throw err;
}
}
diff --git a/desktop/src/shared/api/useRelayAutoHeal.ts b/desktop/src/shared/api/useRelayAutoHeal.ts
index af0cbe55a7..11909228b8 100644
--- a/desktop/src/shared/api/useRelayAutoHeal.ts
+++ b/desktop/src/shared/api/useRelayAutoHeal.ts
@@ -8,6 +8,10 @@ import {
isRelayConnectionDegraded,
useRelayConnection,
} from "@/shared/api/useRelayConnection";
+import {
+ isRateLimited,
+ waitForRateLimit,
+} from "@/shared/api/relayRateLimitGate";
export const AUTO_HEAL_MIN_INTERVAL_MS = 15_000;
@@ -102,10 +106,22 @@ export function useRelayAutoHeal(): void {
if (schedulerRef.current === null) {
schedulerRef.current = new RelayAutoHealScheduler(
- () =>
- void queryClient.invalidateQueries({
- predicate: isRelayDependentQuery,
- }),
+ () => {
+ if (isRateLimited()) {
+ // Connection recovered but the relay is still under back-pressure.
+ // Defer the invalidate until the rate-limit window clears so queries
+ // don't immediately refetch and generate another burst.
+ void waitForRateLimit().then(() => {
+ void queryClient.invalidateQueries({
+ predicate: isRelayDependentQuery,
+ });
+ });
+ } else {
+ void queryClient.invalidateQueries({
+ predicate: isRelayDependentQuery,
+ });
+ }
+ },
AUTO_HEAL_MIN_INTERVAL_MS,
window.setTimeout.bind(window),
window.clearTimeout.bind(window),