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] 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 | 75 +++++--- ...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, 384 insertions(+), 28 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 17a818867d..8eb8fb8a56 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 5fcfe70b91..2a22c5b433 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 parse = |target| -> serde_json::Value { + serde_json::from_str(&request_rejection_message(target, reason)).expect("parse frame") + }; - 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])); + // 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 0000000000..42173b37a1 --- /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 988013bfbc..6d13f0eca7 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 5c602cc6de..1e668a5f05 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 77e10e66ed..100985254f 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 7332075c3f..abd91c4575 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>[];