Answer a rate-limited EVENT with OK false and retry it once the gate clears

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 <subId> 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<sub_id>` passed into
`request_rejection_message` with a `RejectionTarget` so an EVENT refused
by admission or the handler semaphore is answered with
`["OK", <id>, 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

@@ -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<Box<Event>> {
+        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]

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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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 <event_id>` and a subscriber waits on
+/// `CLOSED <sub_id>`; 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]

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");
+});

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);
 

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 =

mobile/lib/shared/relay/relay_session.dart

@@ -815,6 +815,11 @@ class RelaySessionNotifier extends Notifier<SessionState> {
         );
       }
     } 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'),

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<Exception>().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>[];