From e3a3247b13a458d020327979a275d16f04553db6 Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 20:44:46 -0400 Subject: [PATCH 1/2] Answer a rate-limited EVENT with OK false and retry it once the gate clears MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first message after a community switch (or a cold launch) spun for 25s, failed with "Timed out while sending the message", then worked on the second try. The relay's WS admission window (50 frames / 5s per pubkey) counts REQ and EVENT together, and the post-(re)connect REQ fan-out alone can exhaust it. An over-quota REQ already got a correlated `CLOSED rate-limited:`, but an over-quota EVENT got a bare `NOTICE rate-limited:` — nothing ever settled the publisher's pending OK, so it sat until PUBLISH_TIMEOUT_MS. Relay (`connection.rs`): replace the `Option` passed into `request_rejection_message` with a `RejectionTarget` so an EVENT refused by admission or the handler semaphore is answered with `["OK", , false, "rate-limited: …"]`. REQ keeps CLOSED; COUNT keeps NOTICE. The `rate-limited:` prefix and `retry in Ns` hint are unchanged so every client's existing parser still applies. Desktop (`relayClientSession.ts`): on `OK false rate-limited:` arm the gate and re-send the EVENT once it clears instead of rejecting the publish — the relay dedups by id, so the resend cannot double-post. Exactly one retry per publish; a second refusal, or a gate longer than the publish timeout, fails fast with the relay's reason. The NOTICE branch gives in-flight publishes the same single retry so the fix also holds against relays that predate this change. buzz-acp (`relay.rs`): an `OK false rate-limited:` used to fall through to `acknowledge_observer_frame`, silently dropping the refused observer frame from the in-flight window. Arm the gate and put that one frame back at the head of the paced drain instead. Mobile (`relay_session.dart`): an `OK false rate-limited:` now arms the gate before failing the publish, so concurrent work backs off. Tests: relay frame-shape unit test; desktop fail-first unit tests for the retry, its bound, non-rate-limit rejection, legacy NOTICE, and a reset during the wait; acp requeue-only-the-refused-frame; mobile gate-armed-on-OK-false (verified failing before the lib change). Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/relay.rs | 72 +++++++- crates/buzz-relay/src/connection.rs | 77 +++++--- ...layClientSession.publishRateLimit.test.mjs | 170 ++++++++++++++++++ desktop/src/shared/api/relayClientSession.ts | 45 +++++ desktop/src/shared/api/relayClientShared.ts | 2 + mobile/lib/shared/relay/relay_session.dart | 5 + .../test/shared/relay/relay_session_test.dart | 43 +++++ 7 files changed, 385 insertions(+), 29 deletions(-) create mode 100644 desktop/src/shared/api/relayClientSession.publishRateLimit.test.mjs diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 17a818867dd..8eb8fb8a560 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -1234,6 +1234,19 @@ impl BgState { while let Some(event) = self.observer_in_flight.pop_back() { self.gated_observer_pending.push_front(event); } + self.trim_gated_observer_pending(); + } + + /// Restore one observer write the relay refused with `OK false + /// rate-limited:` — correlated by id, so only that frame is retried. + fn requeue_observer_frame(&mut self, event_id: &str) { + if let Some(event) = self.take_observer_in_flight(event_id) { + self.gated_observer_pending.push_front(event); + self.trim_gated_observer_pending(); + } + } + + fn trim_gated_observer_pending(&mut self) { while self.gated_observer_pending.len() > GATED_OBSERVER_QUEUE_CAP { self.gated_observer_pending.pop_front(); self.gated_observer_dropped += 1; @@ -1253,13 +1266,15 @@ impl BgState { } fn acknowledge_observer_frame(&mut self, event_id: &str) { - if let Some(index) = self + self.take_observer_in_flight(event_id); + } + + fn take_observer_in_flight(&mut self, event_id: &str) -> Option> { + let index = self .observer_in_flight .iter() - .position(|event| event.id.to_hex() == event_id) - { - self.observer_in_flight.remove(index); - } + .position(|event| event.id.to_hex() == event_id)?; + self.observer_in_flight.remove(index) } } @@ -2392,6 +2407,22 @@ async fn handle_ws_message( warn!("mid-session AUTH rejected (event {event_id}): {message} — triggering reconnect"); return false; } + if !accepted && message.starts_with("rate-limited:") { + // The relay refused this EVENT for back-pressure. Arm the + // gate and put the frame (if it was a durable observer + // write) back at the head of the paced drain. + let secs = parse_rate_limit_retry_secs(&message).unwrap_or(0); + let deadline = state.set_rate_limit_gate(secs); + state.requeue_observer_frame(&event_id); + warn!( + "event {event_id} rate-limited — gate armed until ~{:.1}s from now", + deadline + .checked_duration_since(tokio::time::Instant::now()) + .unwrap_or_default() + .as_secs_f64() + ); + return true; + } state.acknowledge_observer_frame(&event_id); debug!("OK for event {event_id}: accepted={accepted} message={message}"); } @@ -6027,6 +6058,37 @@ mod tests { assert!(state.observer_in_flight.is_empty()); } + #[test] + fn rate_limited_ok_requeues_only_the_refused_frame() { + let mut state = BgState::new(); + let keys = Keys::generate(); + let still_in_flight = make_observer_frame(&keys); + let refused = make_observer_frame(&keys); + let later = make_observer_frame(&keys); + + state.track_observer_in_flight(Box::new(still_in_flight.clone())); + state.track_observer_in_flight(Box::new(refused.clone())); + state.park_gated_observer_frame(Box::new(later.clone())); + state.requeue_observer_frame(&refused.id.to_hex()); + + let pending: Vec<_> = state + .gated_observer_pending + .iter() + .map(|event| event.id) + .collect(); + assert_eq!( + pending, + [refused.id, later.id], + "refused frame drains first" + ); + let in_flight: Vec<_> = state.observer_in_flight.iter().map(|e| e.id).collect(); + assert_eq!( + in_flight, + [still_in_flight.id], + "unrefused frame keeps waiting for OK" + ); + } + /// The parked-frame queue is bounded: overflow evicts the oldest frame and /// counts it; the drain resets the counter after logging the summary. #[tokio::test] diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 5fcfe70b91c..2a22c5b4336 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -16,7 +16,7 @@ use uuid::Uuid; use buzz_auth::{generate_challenge, AuthContext, LimitType}; use buzz_core::tenant::TenantContext; -use nostr::Filter; +use nostr::{Event, Filter}; use crate::handlers; use crate::protocol::{ClientMessage, RelayMessage}; @@ -571,7 +571,8 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar let permit = match state.handler_semaphore.clone().try_acquire_owned() { Ok(p) => p, Err(_) => { - conn.send(RelayMessage::notice( + conn.send(request_rejection_message( + RejectionTarget::Event(&event), "rate-limited: too many concurrent requests", )); return; @@ -600,7 +601,7 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar Ok(p) => p, Err(_) => { conn.send(request_rejection_message( - Some(&sub_id), + RejectionTarget::Subscription(&sub_id), "rate-limited: too many concurrent requests", )); return; @@ -642,10 +643,24 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar } } -fn request_rejection_message(sub_id: Option<&str>, reason: &str) -> String { - match sub_id { - Some(sub_id) => RelayMessage::closed(sub_id, reason), - None => RelayMessage::notice(reason), +/// Which client-side frame a rejection is correlated with. +/// +/// A publisher waits on `OK ` and a subscriber waits on +/// `CLOSED `; an uncorrelated `NOTICE` leaves the client hanging +/// until its own timeout fires. +#[derive(Debug, Clone, Copy)] +enum RejectionTarget<'a> { + Event(&'a Event), + Subscription(&'a str), + /// COUNT has no correlated rejection frame; fall back to NOTICE. + Uncorrelated, +} + +fn request_rejection_message(target: RejectionTarget<'_>, reason: &str) -> String { + match target { + RejectionTarget::Event(event) => RelayMessage::ok(&event.id.to_hex(), false, reason), + RejectionTarget::Subscription(sub_id) => RelayMessage::closed(sub_id, reason), + RejectionTarget::Uncorrelated => RelayMessage::notice(reason), } } @@ -679,11 +694,12 @@ async fn enforce_ws_admission( ws_limit, ) .await; - let sub_id = match msg { - ClientMessage::Req { sub_id, .. } => Some(sub_id.as_str()), - _ => None, + let target = match msg { + ClientMessage::Event(event) => RejectionTarget::Event(event), + ClientMessage::Req { sub_id, .. } => RejectionTarget::Subscription(sub_id), + _ => RejectionTarget::Uncorrelated, }; - if !send_admission_result(conn, ws_result, sub_id) { + if !send_admission_result(conn, ws_result, target) { return false; } @@ -702,7 +718,7 @@ async fn enforce_ws_admission( message_limit, ) .await; - if !send_admission_result(conn, message_result, None) { + if !send_admission_result(conn, message_result, target) { return false; } } @@ -713,14 +729,14 @@ async fn enforce_ws_admission( fn send_admission_result( conn: &ConnectionState, result: Result<(), crate::admission::AdmissionError>, - sub_id: Option<&str>, + target: RejectionTarget<'_>, ) -> bool { match result { Ok(()) => true, Err(crate::admission::AdmissionError::Exceeded { reset_in_secs }) => { metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "quota").increment(1); conn.send(request_rejection_message( - sub_id, + target, &format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"), )); false @@ -728,7 +744,7 @@ fn send_admission_result( Err(crate::admission::AdmissionError::Unavailable) => { metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "unavailable").increment(1); conn.send(request_rejection_message( - sub_id, + target, "rate-limited: shared admission unavailable", )); false @@ -835,16 +851,29 @@ mod tests { } #[test] - fn req_rejections_are_subscription_scoped() { + fn rejections_are_correlated_with_the_rejected_frame() { let reason = "rate-limited: too many concurrent requests"; - let closed: serde_json::Value = - serde_json::from_str(&request_rejection_message(Some("history-123"), reason)) - .expect("parse CLOSED"); - assert_eq!(closed, serde_json::json!(["CLOSED", "history-123", reason])); - - let notice: serde_json::Value = - serde_json::from_str(&request_rejection_message(None, reason)).expect("parse NOTICE"); - assert_eq!(notice, serde_json::json!(["NOTICE", reason])); + let parse = |target| -> serde_json::Value { + serde_json::from_str(&request_rejection_message(target, reason)).expect("parse frame") + }; + + // A rejected EVENT must settle the publisher's pending OK, not leave + // it waiting on the publish timeout. + let event = nostr::EventBuilder::text_note("hi") + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign event"); + assert_eq!( + parse(RejectionTarget::Event(&event)), + serde_json::json!(["OK", event.id.to_hex(), false, reason]) + ); + assert_eq!( + parse(RejectionTarget::Subscription("history-123")), + serde_json::json!(["CLOSED", "history-123", reason]) + ); + assert_eq!( + parse(RejectionTarget::Uncorrelated), + serde_json::json!(["NOTICE", reason]) + ); } #[tokio::test] diff --git a/desktop/src/shared/api/relayClientSession.publishRateLimit.test.mjs b/desktop/src/shared/api/relayClientSession.publishRateLimit.test.mjs new file mode 100644 index 00000000000..42173b37a19 --- /dev/null +++ b/desktop/src/shared/api/relayClientSession.publishRateLimit.test.mjs @@ -0,0 +1,170 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +// ── Fake-timer setup ────────────────────────────────────────────────────────── +// The rate-limit gate and publish timeout both use window.setTimeout. + +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); +} + +async function tickTo(ms) { + fakeNow = ms; + for (const [id, { fn, fireAt }] of Array.from(pendingTimers.entries())) { + if (fireAt <= fakeNow) { + pendingTimers.delete(id); + fn(); + } + } + // Let promise chains hanging off fired timers (the gate) run. + await new Promise((resolve) => setImmediate(resolve)); +} + +Date.now = () => fakeNow; +globalThis.window = { + setTimeout: fakeSetTimeout, + clearTimeout: fakeClearTimeout, +}; + +// Import after the window shim is installed. +const { resetRateLimitGate, isRateLimited } = await import( + "./relayRateLimitGate.ts" +); +const { RelayClient } = await import("./relayClientSession.ts"); + +const RATE_LIMITED = "rate-limited: quota exceeded; retry in 3s"; + +function makeClient() { + pendingTimers.clear(); + nextTimerId = 1; + fakeNow = 0; + resetRateLimitGate(); + + const client = new RelayClient(); + const sent = []; + // Pretend the socket is open; capture frames instead of invoking Tauri. + client.wsId = 1; + client.sendRaw = async (payload) => { + sent.push(payload); + }; + return { client, sent }; +} + +function deliver(client, frame) { + return client.handleWsMessage(JSON.stringify(frame), 0); +} + +function watch(promise) { + const state = { settled: false, value: undefined, error: undefined }; + promise.then( + (value) => Object.assign(state, { settled: true, value }), + (error) => Object.assign(state, { settled: true, error }), + ); + return state; +} + +const event = { id: "e1", kind: 9, content: "hi" }; + +test("OK=false rate-limited arms the gate and re-sends the EVENT once it clears", async () => { + const { client, sent } = makeClient(); + const publish = watch(client.publishEvent(event, "timeout", "send failed")); + await tickTo(0); + assert.equal(sent.length, 1); + + await deliver(client, ["OK", "e1", false, RATE_LIMITED]); + await tickTo(0); + assert.equal(publish.settled, false, "first refusal must not fail the send"); + assert.equal(isRateLimited(), true, "gate armed from the OK reason"); + assert.equal(sent.length, 1, "no resend while the gate is closed"); + + await tickTo(3_000); + assert.equal(sent.length, 2, "EVENT re-sent when the gate cleared"); + assert.deepEqual(sent[1], ["EVENT", event]); + + await deliver(client, ["OK", "e1", true, ""]); + await tickTo(3_000); + assert.equal( + publish.value, + event, + "retry's OK resolves the original publish", + ); +}); + +test("a second rate-limited refusal fails fast with the relay's reason", async () => { + const { client, sent } = makeClient(); + const publish = watch(client.publishEvent(event, "timeout", "send failed")); + await tickTo(0); + + await deliver(client, ["OK", "e1", false, RATE_LIMITED]); + await tickTo(3_000); + assert.equal(sent.length, 2); + + await deliver(client, ["OK", "e1", false, RATE_LIMITED]); + await tickTo(3_000); + assert.equal(publish.error?.message, RATE_LIMITED); + assert.equal(sent.length, 2, "retry is bounded to one"); +}); + +test("a non-rate-limit OK=false still rejects immediately", async () => { + const { client } = makeClient(); + const publish = watch(client.publishEvent(event, "timeout", "send failed")); + await tickTo(0); + + await deliver(client, [ + "OK", + "e1", + false, + "restricted: not a channel member", + ]); + await tickTo(0); + assert.equal(publish.error?.message, "restricted: not a channel member"); +}); + +test("legacy NOTICE rate-limited gives pending publishes the same single retry", async () => { + const { client, sent } = makeClient(); + const publish = watch(client.publishEvent(event, "timeout", "send failed")); + await tickTo(0); + + await deliver(client, ["NOTICE", RATE_LIMITED]); + await tickTo(0); + assert.equal(publish.settled, false); + assert.equal(sent.length, 1); + + // A second uncorrelated NOTICE cannot be attributed to this publish; it must + // neither fail it nor grant a second retry. + await deliver(client, ["NOTICE", RATE_LIMITED]); + await tickTo(3_000); + assert.equal(publish.settled, false); + assert.equal(sent.length, 2, "exactly one resend"); + + await deliver(client, ["OK", "e1", true, ""]); + await tickTo(3_000); + assert.equal(publish.value, event); +}); + +test("a retry scheduled before a connection reset does not fire on the dead socket", async () => { + const { client, sent } = makeClient(); + const publish = watch(client.publishEvent(event, "timeout", "send failed")); + await tickTo(0); + + await deliver(client, ["OK", "e1", false, RATE_LIMITED]); + await tickTo(0); + client.resetConnection(new Error("Relay connection closed."), { + reconnect: false, + }); + await tickTo(0); + assert.equal(publish.error?.message, "Relay connection closed."); + + await tickTo(3_000); + assert.equal(sent.length, 1, "no resend after the publish was settled"); +}); diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 988013bfbc7..6d13f0eca78 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -41,6 +41,7 @@ import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay"; import { activateRateLimit, parseRateLimitHint, + rateLimitRemainingMs, waitForRateLimit, } from "@/shared/api/relayRateLimitGate"; import { @@ -833,10 +834,49 @@ export class RelayClient { // Relay back-pressure — arm the gate until the window expires. if (notice.startsWith("rate-limited:")) { activateRateLimit(parseRateLimitHint(notice)); + // Older relays answer an over-quota EVENT with this uncorrelated + // NOTICE instead of `OK false`; the publish would otherwise sit until + // PUBLISH_TIMEOUT_MS. Give every in-flight publish its one retry. + for (const pendingEvent of this.pendingEvents.values()) { + this.retryPublishAfterRateLimit(pendingEvent); + } } } } + /** + * The relay refused an EVENT for back-pressure. Re-send it once the gate + * clears instead of failing the user's send — the burst that tripped the + * limit is typically our own post-(re)connect REQ fan-out, and the relay + * dedups by id so a resend can never double-post. One retry only; a second + * refusal is reported. Returns `false` when no retry is available. + */ + private retryPublishAfterRateLimit(pendingEvent: PendingEvent): boolean { + if ( + pendingEvent.retriedAfterRateLimit || + rateLimitRemainingMs() >= PUBLISH_TIMEOUT_MS + ) { + return false; + } + pendingEvent.retriedAfterRateLimit = true; + const { event } = pendingEvent; + void waitForRateLimit().then(async () => { + // Settled meanwhile (timeout, reset, community switch) — nothing to do. + if (this.pendingEvents.get(event.id) !== pendingEvent) return; + try { + await this.sendRaw(["EVENT", event]); + } catch (error) { + if (this.pendingEvents.get(event.id) !== pendingEvent) return; + window.clearTimeout(pendingEvent.timeout); + this.pendingEvents.delete(event.id); + pendingEvent.reject( + this.normalizeRelayError(error, "Failed to re-send event to relay."), + ); + } + }); + return true; + } + private async handleAuthChallenge(challenge: string, generation: number) { if (!this.relayUrl) { this.relayUrl = await getRelayWsUrl(); @@ -916,6 +956,11 @@ export class RelayClient { return; } + if (!success && message.startsWith("rate-limited:")) { + activateRateLimit(parseRateLimitHint(message)); + if (this.retryPublishAfterRateLimit(pendingEvent)) return; + } + window.clearTimeout(pendingEvent.timeout); this.pendingEvents.delete(eventId); diff --git a/desktop/src/shared/api/relayClientShared.ts b/desktop/src/shared/api/relayClientShared.ts index 5c602cc6de3..1e668a5f05e 100644 --- a/desktop/src/shared/api/relayClientShared.ts +++ b/desktop/src/shared/api/relayClientShared.ts @@ -89,6 +89,8 @@ export type PendingEvent = { resolve: (event: RelayEvent) => void; reject: (error: Error) => void; timeout: number; + /** Set once the EVENT has been re-sent after a `rate-limited:` refusal. */ + retriedAfterRateLimit?: boolean; }; export type RelaySubscription = diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 77e10e66ede..100985254f1 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -815,6 +815,11 @@ class RelaySessionNotifier extends Notifier { ); } } else { + // The relay refused the EVENT for back-pressure: arm the gate so + // concurrent subscribes/publishes back off instead of piling on. + if (classifyRelayClosed(message) == RelayClosedClass.rateLimited) { + _rateLimitGate.activate(parseRateLimitRetrySeconds(message)); + } if (!pending.completer.isCompleted) { pending.completer.completeError( Exception(message.isNotEmpty ? message : 'Event rejected'), diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index 7332075c3f6..abd91c4575d 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -943,6 +943,49 @@ void main() { expect(disposeTimer.isActive, isFalse); }); + test('rate-limited OK=false fails the publish and arms the gate', () async { + final gateTimers = <_ManualTimer>[]; + final gate = RelayRateLimitGate( + timerFactory: (duration, callback) { + final timer = _ManualTimer(duration, callback); + gateTimers.add(timer); + return timer; + }, + ); + final session = RelaySessionNotifier(rateLimitGate: gate); + session.debugAttachSocketForTest(_RecordingRelaySocket()); + final event = NostrEvent( + id: 'e1', + pubkey: '', + createdAt: 0, + kind: 9, + tags: const [], + content: 'hi', + sig: '', + ); + + final publish = session.publish(event); + session.debugHandleMessage([ + 'OK', + 'e1', + false, + 'rate-limited: quota exceeded; retry in 4s', + ]); + + await expectLater( + publish, + throwsA( + isA().having( + (error) => error.toString(), + 'message', + contains('rate-limited: quota exceeded'), + ), + ), + ); + expect(gate.isActive, isTrue); + expect(gateTimers.single.duration, const Duration(seconds: 4)); + }); + test('rate-limited live CLOSED honours the gate floor', () async { final retryTimers = <_ManualTimer>[]; final gateTimers = <_ManualTimer>[]; From 43c5dc8d2caddd917d410f2dece41795cb8c43c1 Mon Sep 17 00:00:00 2001 From: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Date: Sat, 22 Aug 2026 21:14:11 -0400 Subject: [PATCH 2/2] relay clients: keep rate-limit retry out of oversized session files Review follow-up for the first-send-after-community-switch fix. Desktop: move the OK-frame settle/retry logic from relayClientSession.ts (already over the file-size ratchet; grew 1084 -> 1129) into a pure `handlePublishOk` in relayPublishRecovery.ts, taking the pending map and a send callback. The session file now shrinks to 1078. Drop the uncorrelated NOTICE retry-all. A `NOTICE rate-limited:` may have been triggered by a REQ or COUNT, and an ephemeral event (e.g. presence, kind 20001) the relay already fanned out has no id dedup, so resending every in-flight publish could double-deliver. Only the correlated `OK false rate-limited:` is retried; the NOTICE still arms the gate. The test now asserts NOTICE never resends, and the reset test stubs the Tauri websocket disconnect so it runs without a stderr TypeError. Mobile: relay_session.dart tripped the same ratchet (1000 -> 1005). The classify-then-arm sequence appeared three times, so it becomes `RelayRateLimitGate.activateIfRateLimited(message)`; the session file drops to 995. Gate unit test added. Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz> --- ...layClientSession.publishRateLimit.test.mjs | 22 +++-- desktop/src/shared/api/relayClientSession.ts | 66 ++------------- .../src/shared/api/relayPublishRecovery.ts | 81 +++++++++++++++++++ .../shared/relay/relay_rate_limit_gate.dart | 9 +++ mobile/lib/shared/relay/relay_session.dart | 17 +--- .../relay/relay_rate_limit_gate_test.dart | 13 +++ 6 files changed, 129 insertions(+), 79 deletions(-) create mode 100644 desktop/src/shared/api/relayPublishRecovery.ts diff --git a/desktop/src/shared/api/relayClientSession.publishRateLimit.test.mjs b/desktop/src/shared/api/relayClientSession.publishRateLimit.test.mjs index 42173b37a19..dc9a676256c 100644 --- a/desktop/src/shared/api/relayClientSession.publishRateLimit.test.mjs +++ b/desktop/src/shared/api/relayClientSession.publishRateLimit.test.mjs @@ -34,6 +34,14 @@ Date.now = () => fakeNow; globalThis.window = { setTimeout: fakeSetTimeout, clearTimeout: fakeClearTimeout, + // resetConnection() closes the socket through the Tauri bridge; answer it + // so the test stays quiet and a real warning cannot hide in the noise. + __TAURI_INTERNALS__: { + invoke: async (command) => { + if (command === "plugin:websocket|disconnect") return; + throw new Error(`Unexpected Tauri command: ${command}`); + }, + }, }; // Import after the window shim is installed. @@ -130,22 +138,20 @@ test("a non-rate-limit OK=false still rejects immediately", async () => { assert.equal(publish.error?.message, "restricted: not a channel member"); }); -test("legacy NOTICE rate-limited gives pending publishes the same single retry", async () => { +test("an uncorrelated NOTICE rate-limited arms the gate but never re-sends", async () => { const { client, sent } = makeClient(); const publish = watch(client.publishEvent(event, "timeout", "send failed")); await tickTo(0); + // The NOTICE may have been triggered by a REQ or COUNT, and an ephemeral + // event the relay already fanned out would be delivered twice on resend. await deliver(client, ["NOTICE", RATE_LIMITED]); await tickTo(0); - assert.equal(publish.settled, false); - assert.equal(sent.length, 1); + assert.equal(isRateLimited(), true, "gate armed from the NOTICE"); + assert.equal(publish.settled, false, "NOTICE cannot be attributed; no fail"); - // A second uncorrelated NOTICE cannot be attributed to this publish; it must - // neither fail it nor grant a second retry. - await deliver(client, ["NOTICE", RATE_LIMITED]); await tickTo(3_000); - assert.equal(publish.settled, false); - assert.equal(sent.length, 2, "exactly one resend"); + assert.equal(sent.length, 1, "no resend from an uncorrelated NOTICE"); await deliver(client, ["OK", "e1", true, ""]); await tickTo(3_000); diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 6d13f0eca78..b3227fc8a83 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -38,10 +38,10 @@ import { } from "@/shared/api/relayClosedRecovery"; import { getChannelReconnectRepairEvents } from "@/shared/api/channelReconnectRepair"; import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay"; +import { handlePublishOk } from "@/shared/api/relayPublishRecovery"; import { activateRateLimit, parseRateLimitHint, - rateLimitRemainingMs, waitForRateLimit, } from "@/shared/api/relayRateLimitGate"; import { @@ -834,49 +834,10 @@ export class RelayClient { // Relay back-pressure — arm the gate until the window expires. if (notice.startsWith("rate-limited:")) { activateRateLimit(parseRateLimitHint(notice)); - // Older relays answer an over-quota EVENT with this uncorrelated - // NOTICE instead of `OK false`; the publish would otherwise sit until - // PUBLISH_TIMEOUT_MS. Give every in-flight publish its one retry. - for (const pendingEvent of this.pendingEvents.values()) { - this.retryPublishAfterRateLimit(pendingEvent); - } } } } - /** - * The relay refused an EVENT for back-pressure. Re-send it once the gate - * clears instead of failing the user's send — the burst that tripped the - * limit is typically our own post-(re)connect REQ fan-out, and the relay - * dedups by id so a resend can never double-post. One retry only; a second - * refusal is reported. Returns `false` when no retry is available. - */ - private retryPublishAfterRateLimit(pendingEvent: PendingEvent): boolean { - if ( - pendingEvent.retriedAfterRateLimit || - rateLimitRemainingMs() >= PUBLISH_TIMEOUT_MS - ) { - return false; - } - pendingEvent.retriedAfterRateLimit = true; - const { event } = pendingEvent; - void waitForRateLimit().then(async () => { - // Settled meanwhile (timeout, reset, community switch) — nothing to do. - if (this.pendingEvents.get(event.id) !== pendingEvent) return; - try { - await this.sendRaw(["EVENT", event]); - } catch (error) { - if (this.pendingEvents.get(event.id) !== pendingEvent) return; - window.clearTimeout(pendingEvent.timeout); - this.pendingEvents.delete(event.id); - pendingEvent.reject( - this.normalizeRelayError(error, "Failed to re-send event to relay."), - ); - } - }); - return true; - } - private async handleAuthChallenge(challenge: string, generation: number) { if (!this.relayUrl) { this.relayUrl = await getRelayWsUrl(); @@ -951,24 +912,13 @@ export class RelayClient { return; } - const pendingEvent = this.pendingEvents.get(eventId); - if (!pendingEvent) { - return; - } - - if (!success && message.startsWith("rate-limited:")) { - activateRateLimit(parseRateLimitHint(message)); - if (this.retryPublishAfterRateLimit(pendingEvent)) return; - } - - window.clearTimeout(pendingEvent.timeout); - this.pendingEvents.delete(eventId); - - if (success) { - pendingEvent.resolve(pendingEvent.event); - } else { - pendingEvent.reject(new Error(message || "Relay rejected the event.")); - } + handlePublishOk({ + pendingEvents: this.pendingEvents, + eventId, + success, + message, + sendEvent: (event) => this.sendRaw(["EVENT", event]), + }); } private hasLiveSubscriptions() { diff --git a/desktop/src/shared/api/relayPublishRecovery.ts b/desktop/src/shared/api/relayPublishRecovery.ts new file mode 100644 index 00000000000..026d3e2897c --- /dev/null +++ b/desktop/src/shared/api/relayPublishRecovery.ts @@ -0,0 +1,81 @@ +import type { PendingEvent } from "@/shared/api/relayClientShared"; +import { PUBLISH_TIMEOUT_MS } from "@/shared/api/relayClientTimings"; +import { + activateRateLimit, + parseRateLimitHint, + rateLimitRemainingMs, + waitForRateLimit, +} from "@/shared/api/relayRateLimitGate"; + +/** + * Settle a pending publish from the relay's `OK` frame. + * + * An `OK false "rate-limited: …"` is not a verdict on the event, only on + * timing: the burst that tripped the limit is typically our own + * post-(re)connect REQ fan-out. Arm the gate and re-send the EVENT once it + * clears instead of failing the user's send. Because the refusal is + * correlated, the relay has not applied the event, so the resend cannot + * double-deliver — unlike an uncorrelated `NOTICE`, which is never retried. + * One retry per publish; a second refusal, or a gate longer than the publish + * timeout, rejects with the relay's reason. + */ +export function handlePublishOk({ + pendingEvents, + eventId, + success, + message, + sendEvent, +}: { + pendingEvents: Map; + eventId: string; + success: boolean; + message: string; + sendEvent: (event: PendingEvent["event"]) => Promise; +}) { + const pendingEvent = pendingEvents.get(eventId); + if (!pendingEvent) return; + + if (!success && message.startsWith("rate-limited:")) { + activateRateLimit(parseRateLimitHint(message)); + if ( + !pendingEvent.retriedAfterRateLimit && + rateLimitRemainingMs() < PUBLISH_TIMEOUT_MS + ) { + pendingEvent.retriedAfterRateLimit = true; + void waitForRateLimit().then(async () => { + // Settled meanwhile (timeout, reset, community switch) — nothing to do. + if (pendingEvents.get(eventId) !== pendingEvent) return; + try { + await sendEvent(pendingEvent.event); + } catch (error) { + if (pendingEvents.get(eventId) !== pendingEvent) return; + settle(pendingEvents, eventId, pendingEvent, () => + pendingEvent.reject( + error instanceof Error + ? error + : new Error("Failed to re-send event to relay."), + ), + ); + } + }); + return; + } + } + + settle(pendingEvents, eventId, pendingEvent, () => + success + ? pendingEvent.resolve(pendingEvent.event) + : pendingEvent.reject(new Error(message || "Relay rejected the event.")), + ); +} + +function settle( + pendingEvents: Map, + eventId: string, + pendingEvent: PendingEvent, + finish: () => void, +) { + window.clearTimeout(pendingEvent.timeout); + pendingEvents.delete(eventId); + finish(); +} diff --git a/mobile/lib/shared/relay/relay_rate_limit_gate.dart b/mobile/lib/shared/relay/relay_rate_limit_gate.dart index 61640139368..da5e031d1e5 100644 --- a/mobile/lib/shared/relay/relay_rate_limit_gate.dart +++ b/mobile/lib/shared/relay/relay_rate_limit_gate.dart @@ -1,6 +1,8 @@ import 'dart:async'; import 'dart:math'; +import 'relay_closed_policy.dart'; + /// Creates a timer used by [RelayRateLimitGate]. typedef RelayTimerFactory = Timer Function(Duration duration, void Function() callback); @@ -47,6 +49,13 @@ class RelayRateLimitGate { _timer = _timerFactory(duration, _expire); } + /// Arms the gate from any relay refusal (`CLOSED`, `OK false`, HTTP error) + /// whose message classifies as back-pressure; a no-op otherwise. + void activateIfRateLimited(String message) { + if (classifyRelayClosed(message) != RelayClosedClass.rateLimited) return; + activate(parseRateLimitRetrySeconds(message)); + } + /// Resolves when the active rate-limit window expires. Future wait() { if (!isActive) return Future.value(); diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 100985254f1..19d7e79021e 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -220,11 +220,7 @@ class RelaySessionNotifier extends Notifier { } if (decoded is! Map) return; final message = decoded['error']; - if (message is! String || - classifyRelayClosed(message) != RelayClosedClass.rateLimited) { - return; - } - _rateLimitGate.activate(parseRateLimitRetrySeconds(message)); + if (message is String) _rateLimitGate.activateIfRateLimited(message); } /// Fetch historical events matching [filter]. Sends REQ, collects events @@ -692,9 +688,7 @@ class RelaySessionNotifier extends Notifier { final historySub = _historySubscriptions.remove(subId); if (historySub != null) { - if (closedClass == RelayClosedClass.rateLimited) { - _rateLimitGate.activate(parseRateLimitRetrySeconds(message)); - } + _rateLimitGate.activateIfRateLimited(message); historySub.timeout.cancel(); if (!historySub.completer.isCompleted) { historySub.completer.completeError(Exception(message)); @@ -815,11 +809,8 @@ class RelaySessionNotifier extends Notifier { ); } } else { - // The relay refused the EVENT for back-pressure: arm the gate so - // concurrent subscribes/publishes back off instead of piling on. - if (classifyRelayClosed(message) == RelayClosedClass.rateLimited) { - _rateLimitGate.activate(parseRateLimitRetrySeconds(message)); - } + // Back-pressure refusal: arm the gate so concurrent requests back off. + _rateLimitGate.activateIfRateLimited(message); if (!pending.completer.isCompleted) { pending.completer.completeError( Exception(message.isNotEmpty ? message : 'Event rejected'), diff --git a/mobile/test/shared/relay/relay_rate_limit_gate_test.dart b/mobile/test/shared/relay/relay_rate_limit_gate_test.dart index 811d439572d..0a8d3afdf0a 100644 --- a/mobile/test/shared/relay/relay_rate_limit_gate_test.dart +++ b/mobile/test/shared/relay/relay_rate_limit_gate_test.dart @@ -92,6 +92,19 @@ void main() { expect(timers.single.isActive, isFalse); expect(gate.isActive, isFalse); }); + + test('activateIfRateLimited arms only on a back-pressure message', () { + final gate = RelayRateLimitGate( + now: () => DateTime.utc(2026), + timerFactory: (duration, callback) => _ManualTimer(duration, callback), + ); + + gate.activateIfRateLimited('restricted: not a channel member'); + expect(gate.isActive, isFalse); + + gate.activateIfRateLimited('rate-limited: quota exceeded; retry in 3s'); + expect(gate.remainingMs(), 3000); + }); } class _ManualTimer implements Timer {