jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(relay): reject a frame on its own acknowledgement channel (#6961)

Browse Source 
Pinky, an AI agent, updated this description on Wes's behalf after
taking over the startup investigation.

**Category:** fix

**User Impact:** An EVENT refused by WebSocket admission or handler
saturation receives a correlated `OK(event_id, false, reason)` instead
of an uncorrelated NOTICE, so the client can settle that refusal without
waiting for its publish timeout. Rate-limited refusals also arm client
backoff. This fixes a protocol failure mechanism; it does not establish
that every startup send will succeed or that the reported Desktop
startup incident is fully resolved.

**Problem:** Startup opens several live subscriptions and publishes at
once, and the relay's WebSocket admission gate is a fixed 5-second
window (`ws_admission_budget` = `human_ws_events_per_sec * 5`). If that
shared per-principal quota is exhausted, `enforce_ws_admission`
previously rejected an EVENT with a bare `["NOTICE", reason]`. Quota
pressure is a possible trigger, not proof of the original incident's
complete cause.

A NOTICE carries no event id. Both clients settle a pending publish
*only* from an `OK` keyed by event id (desktop `pendingEvents`, mobile
`_pendingEvents`), so nothing settled — and `handle_text_message`
returns early, so no `OK` ever followed either. The send **could not
fail**; it could only time out at `PUBLISH_TIMEOUT_MS` = 25s. That
explains how this rejection mechanism can produce a roughly 25-second
timeout; attributing the original report to it still requires the actual
startup/send workflow.

The handler-semaphore saturation path had the identical defect, and that
one needs no quota burst to fire.

**Solution:** NIP-01 gives each request type its own acknowledgement
channel, and a rejection is only actionable on the same one. Reject a
REQ with `CLOSED`, an EVENT with `OK(id, false, reason)`, and fall back
to `NOTICE` only where no per-request correlation exists. COUNT refusals
now also use `CLOSED(query_id, reason)` per NIP-45, covering both quota
admission and handler saturation (added in
`cd12c93804b87a24b61075dfd171dc471a0a527f`).

Reason strings are unchanged, so the `rate-limited:` prefix and `retry
in {N}s` hint that existing client gates parse keep working (desktop
`parseRateLimitHint`, mobile `RelayRateLimitGate`, buzz-acp
`set_rate_limit_gate`). Only the frame *type* changes, so
`docs/multi-tenant-relay.md` L7 stays satisfied.

Two notes on how this landed, both worth a reviewer's attention:

1. **A survived mutation became a design change.**
`send_admission_result` originally took a `RejectionTarget` parameter,
and reverting the *second* call site (the per-minute message quota)
survived the whole suite — with Redis unreachable the first quota check
short-circuits, so that line is unreachable in test. Rather than test
around it, the parameter is gone: the target is derived from the frame,
so no call site can name the wrong channel.

2. **The relay fix would have caused a client regression on its own.**
Gate arming lived only in the NOTICE branch. Once rejections arrive as
`OK:false`, `handleOk` failed the send without ever backing off — the
client would retry straight into the same quota. Desktop and Mobile now
arm on a `rate-limited:` OK rejection. ACP was subsequently fixed in
`3b06dd32493596ec650f20abf8805791c50fdc24`: it arms the gate and
re-parks only the refused observer frame, preserving other in-flight
frames. Desktop gets `activateRateLimitIfSignalled` as the single owner
of that prefix test, called from both `handleOk` and the NOTICE branch.

<details>
<summary>File changes</summary>

**crates/buzz-relay/src/rejection.rs** (new)
Owns the admission-rejection concern: `RejectionTarget`,
`rejection_target_for`, `request_rejection_message`,
`send_admission_result`, and `enforce_ws_admission`, moved out of
`connection.rs`. Six tests, two of which drive the real
`enforce_ws_admission` against a real `AppState`.

**crates/buzz-relay/src/connection.rs**
Fix the EVENT handler-semaphore rejection to correlate to the event id;
delegate admission to the new module. Add two tests that drive the real
`handle_text_message` with every handler permit held. Down from 1319 to
1116 lines.

**crates/buzz-relay/src/state.rs**
Widen the existing `test_state` helper to `pub(crate)` so the rejection
tests reuse it rather than adding a ninth copy of `AppState`
construction.

**desktop/src/shared/api/relayRateLimitGate.ts**
Add `activateRateLimitIfSignalled` — one owner for the `rate-limited:`
prefix test, since three inbound frame types now carry it.

**desktop/src/shared/api/relayClientSession.ts**
Arm the gate on a rate-limited OK rejection; route the NOTICE branch
through the same helper. Net zero lines, which keeps this
already-oversized file within the differential ratchet.

**desktop/src/shared/api/relayClientPublishRejection.test.mjs** (new)
Four tests against the real `RelayClient`: a rate-limited OK settles the
pending publish and arms the gate; an ordinary rejection does not arm
it; an accepted OK still resolves.

**mobile/lib/shared/relay/relay_session.dart**
Arm the gate in `_handleOk` for a rate-limited rejection.

**mobile/test/shared/relay/relay_session_test.dart**
Two tests driving the real `publish` + `debugHandleMessage` path.

</details>

<details>
<summary>Validation</summary>

**Mutation-tested — 5 mutations, all now killed.** Each production call
site was reverted to the defective behaviour to confirm a test fails.
This caught two false-negative tests:

| # | Mutation | Result |
|---|----------|--------|
| 1 | `rejection_target_for`: EVENT → `Connection` | 4 tests fail |
| 2 | EVENT handler-semaphore call site → bare NOTICE | **survived at
first** |
| 3 | per-minute quota call site → `Connection` | **survived**; fixed by
removing the parameter |
| 4 | desktop `handleOk` gate arming removed | 1 test fails |
| 5 | mobile `_handleOk` gate arming removed | 1 test fails |

Mutation 2 is the lesson: my first saturation test called
`request_rejection_message` directly, so reverting the real call site
inside the `match` arm left it green. It now drives
`handle_text_message` itself and dies on that mutation.

- `cargo test -p buzz-relay` — 928 passed, 1 failed:
`api::mesh_demo::tests::demo_join_forwarded_arm_round_trips_echo`,
**pre-existing**, reproduced with all changes stashed at `4dd4d73de`.
- `cd desktop && npm test` — 5721 passed, 0 failed (full suite).
- `cd mobile && flutter test` — 1876 passed, 0 failed (full suite).
- `just fmt-check`, `just clippy`, `just desktop-check`, `just
mobile-check`, `just file-size-check` — clean. Desktop's 5 biome
warnings are pre-existing (reproduced with changes stashed).
- All 9 pre-push lanes green, including `rust-tests` and
`desktop-tauri-checks`.

**Not verified:** not reproduced end-to-end against a live relay under a
forced quota burst. The causal chain is source-proven and
mutation-proven at the frame level; the ~25s attribution follows from
`PUBLISH_TIMEOUT_MS` but is not directly measured. A packaged-build
click-through would close that gap.

</details>

Related work: #6957 bounds Desktop HTTP event submission, but safe
retained-operation recovery after exhausted/ambiguous outcomes remains
unfinished. #6998 is the separately reviewable Desktop
readiness/duplicate-subscription slice. Neither is claimed to complete
native before/after startup-send validation.

Diagnosis note: `RESEARCH/DESKTOP_STARTUP_SEND_STALL_2026_08_27.md`
(Brain's workspace).

## Current review disposition (2026-08-28)

The [review on
`cd12c938`](https://github.com/block/buzz/pull/6961#pullrequestreview-5052902510)
identified ACP's missing rate-limited-OK handling. Commit
`3b06dd32493596ec650f20abf8805791c50fdc24` fixes gate arming, re-parking
the specifically refused observer frame, and the stale NOTICE comment.
Two regressions drive the real frame dispatcher. See [the implementation
and validation
response](https://github.com/block/buzz/pull/6961#issuecomment-5455032054).

The Mobile generation-check inline thread is resolved: its `async
publish` returns a failed Future when superseded; it does not throw
synchronously at invocation. No further production change was indicated
by that comment.

The validation counts above describe the original slice, not a new
rerun. At `3b06dd324`, the current GitHub check rollup has successful
completed test/build checks (non-applicable jobs skipped). The
security-review comment still requires review for the current base/head
range; do not read a green authorization job as a completed security
review. Approval and merge remain human decisions.

---------

Signed-off-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz>
Signed-off-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz>
Co-authored-by: Carl <9d00794d3df50972eb8b615511783cab12a77a8fd5dd5edd58073ec73b54bd8b@buzz.block.builderlab.xyz>
This commit is contained in:
Wes
2026-08-31 14:43:31 -06:00
committed by GitHub
parent 93237b4a74
commit 2f3dd850db
13 changed files with 1448 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
crates/buzz-acp/src/relay.rs
View File

@@ -1370,6 +1370,40 @@ impl BgState {
while let Some(event) = self.observer_in_flight.pop_back() {
self.gated_observer_pending.push_front(event);
}
self.trim_gated_observer_pending();
}
/// Re-park a frame the relay explicitly refused, ahead of frames parked
/// after the gate armed.
///
/// An `OK(id, false, …)` names the refused frame, so only that frame is
/// retried — frames still awaiting their own verdict stay in the
/// acknowledgment window. This is the correlated counterpart to
/// [`Self::requeue_observer_in_flight`], which must retry everything
/// because a NOTICE identifies nothing.
fn requeue_rejected_observer_frame(&mut self, event_id: &str) {
let Some(index) = self
.observer_in_flight
.iter()
.position(|event| event.id.to_hex() == event_id)
else {
return;
};
if let Some(event) = self.observer_in_flight.remove(index) {
if self.gated_observer_pending.len() >= GATED_OBSERVER_QUEUE_CAP {
self.gated_observer_pending.pop_front();
self.gated_observer_dropped += 1;
warn!(
dropped_total = self.gated_observer_dropped,
"gated observer queue full — dropped oldest parked frame for refused retry"
);
}
self.gated_observer_pending.push_front(event);
}
}
/// Enforce the parked-queue bound, counting evictions so loss stays visible.
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;
@@ -2362,7 +2396,10 @@ async fn handle_ws_message(
RelayMessage::Notice { message } => {
// Fix 4: NOTICE at warn level.
tracing::warn!("relay NOTICE: {message}");
// The relay sends NOTICE for rate-limited EVENT/COUNT frames.
// NOTICE now carries only connection-scoped refusals: an
// EVENT is refused via OK and a REQ/COUNT via CLOSED. A
// NOTICE names nothing, so every unacknowledged observer
// write must be retried.
if message.starts_with("rate-limited:") {
let secs = parse_rate_limit_retry_secs(&message).unwrap_or(0);
let deadline = state.set_rate_limit_gate(secs);
@@ -2530,6 +2567,25 @@ async fn handle_ws_message(
warn!("mid-session AUTH rejected (event {event_id}): {message} — triggering reconnect");
return false;
}
// A refused EVENT is acknowledged on its own channel, so the
// backoff must arm here — not only in the NOTICE arm. Without
// this the harness would publish straight back into the same
// quota it was just refused on.
if !accepted && message.starts_with("rate-limited:") {
let secs = parse_rate_limit_retry_secs(&message).unwrap_or(0);
let deadline = state.set_rate_limit_gate(secs);
// The OK names the refused frame, so re-park only that
// one rather than every unacknowledged frame.
state.requeue_rejected_observer_frame(&event_id);
warn!(
"rate-limit gate armed via OK for event {event_id} 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}");
}
@@ -6136,6 +6192,151 @@ mod tests {
);
}
/// A rate-limited `OK(id, false, …)` must arm the backoff gate and re-park
/// the refused frame, driven through the real frame dispatcher.
///
/// This is the buzz-acp side of the relay's rejection-correlation change:
/// a refused EVENT is now acknowledged on its own channel instead of via
/// NOTICE. Reverting either the gate arming or the requeue in the `Ok` arm
/// must fail this test.
#[tokio::test]
async fn rate_limited_ok_arms_gate_and_reparks_refused_observer_frame() {
let (mut client, _server) = test_ws_pair().await;
let (event_tx, _event_rx) = mpsc::channel::<Option<BuzzEvent>>(4);
let (observer_control_tx, _observer_control_rx) = mpsc::channel::<Event>(4);
let keys = Keys::generate();
let mut state = BgState::new();
let refused = make_observer_frame(&keys);
let still_pending = make_observer_frame(&keys);
state.track_observer_in_flight(Box::new(refused.clone()));
state.track_observer_in_flight(Box::new(still_pending.clone()));
assert!(
state.check_rate_gate().is_none(),
"gate must start disarmed"
);
let frame = json!([
"OK",
refused.id.to_hex(),
false,
"rate-limited: retry in 5s"
]);
let should_continue = handle_ws_message(
Message::Text(frame.to_string().into()),
&mut client,
&event_tx,
&observer_control_tx,
&mut state,
&keys,
"wss://relay.test",
"agent-pubkey",
None,
)
.await;
assert!(should_continue, "a rate-limited OK must keep the socket");
assert!(
state.check_rate_gate().is_some(),
"a rate-limited OK must arm the backoff gate, or the harness \
republishes straight into the same quota"
);
let parked: Vec<_> = state
.gated_observer_pending
.iter()
.map(|event| event.id)
.collect();
assert_eq!(
parked,
[refused.id],
"the refused frame must be re-parked for redelivery, not dropped"
);
let in_flight: Vec<_> = state
.observer_in_flight
.iter()
.map(|event| event.id)
.collect();
assert_eq!(
in_flight,
[still_pending.id],
"frames still awaiting their own verdict must stay in flight"
);
}
#[test]
fn rejected_observer_frame_displaces_oldest_parked_frame_at_capacity() {
let mut state = BgState::new();
let keys = Keys::generate();
let refused = make_observer_frame(&keys);
state.track_observer_in_flight(Box::new(refused.clone()));
let oldest = make_observer_frame(&keys);
state.park_gated_observer_frame(Box::new(oldest.clone()));
let mut survivors = Vec::with_capacity(GATED_OBSERVER_QUEUE_CAP - 1);
for _ in 1..GATED_OBSERVER_QUEUE_CAP {
let event = make_observer_frame(&keys);
survivors.push(event.id);
state.park_gated_observer_frame(Box::new(event));
}
state.requeue_rejected_observer_frame(&refused.id.to_hex());
let parked: Vec<_> = state
.gated_observer_pending
.iter()
.map(|event| event.id)
.collect();
assert_eq!(parked.len(), GATED_OBSERVER_QUEUE_CAP);
assert_eq!(parked.first(), Some(&refused.id));
assert_eq!(&parked[1..], survivors.as_slice());
assert!(!parked.contains(&oldest.id));
assert_eq!(state.gated_observer_dropped, 1);
assert!(state.observer_in_flight.is_empty());
}
/// A non-rate-limit refusal is terminal: retrying would be refused
/// identically, so the frame is retired rather than re-parked, and the
/// backoff gate stays disarmed.
#[tokio::test]
async fn non_rate_limited_ok_rejection_retires_frame_without_arming_gate() {
let (mut client, _server) = test_ws_pair().await;
let (event_tx, _event_rx) = mpsc::channel::<Option<BuzzEvent>>(4);
let (observer_control_tx, _observer_control_rx) = mpsc::channel::<Event>(4);
let keys = Keys::generate();
let mut state = BgState::new();
let refused = make_observer_frame(&keys);
state.track_observer_in_flight(Box::new(refused.clone()));
let frame = json!(["OK", refused.id.to_hex(), false, "invalid: bad signature"]);
let should_continue = handle_ws_message(
Message::Text(frame.to_string().into()),
&mut client,
&event_tx,
&observer_control_tx,
&mut state,
&keys,
"wss://relay.test",
"agent-pubkey",
None,
)
.await;
assert!(should_continue, "a rejected event must not drop the socket");
assert!(
state.check_rate_gate().is_none(),
"only a rate-limit refusal arms the backoff gate"
);
assert!(
state.gated_observer_pending.is_empty(),
"a permanently refused frame must not be requeued into a retry loop"
);
assert!(
state.observer_in_flight.is_empty(),
"a permanently refused frame must be retired from the window"
);
}
/// Build a signed observer telemetry frame (kind 24200) for gate tests.
fn make_observer_frame(keys: &Keys) -> Event {
let recipient = Keys::generate();

251
crates/buzz-relay/src/connection.rs
View File

@@ -14,12 +14,13 @@ use tracing::Instrument as _;
use tracing::{debug, info, trace, warn};
use uuid::Uuid;
use buzz_auth::{generate_challenge, AuthContext, LimitType};
use buzz_auth::{generate_challenge, AuthContext};
use buzz_core::tenant::TenantContext;
use nostr::Filter;
use crate::handlers;
use crate::protocol::{ClientMessage, RelayMessage};
use crate::rejection::{enforce_ws_admission, request_rejection_message, RejectionTarget};
use crate::state::{
run_registered_community_connection, AppState, CommunityConnectionControl,
CommunityDisconnectReason,
@@ -571,7 +572,10 @@ 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(
// Correlate to the event id: a bare NOTICE here strands the
// client's pending publish exactly as an over-quota one did.
conn.send(request_rejection_message(
RejectionTarget::Event(event.id),
"rate-limited: too many concurrent requests",
));
return;
@@ -600,7 +604,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;
@@ -621,7 +625,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::Subscription(&sub_id),
"rate-limited: too many concurrent requests",
));
return;
@@ -642,105 +647,140 @@ 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),
}
}
async fn enforce_ws_admission(
msg: &ClientMessage,
conn: &ConnectionState,
state: &AppState,
) -> bool {
let is_event = matches!(msg, ClientMessage::Event(_));
if !is_event && !matches!(msg, ClientMessage::Req { .. } | ClientMessage::Count { .. }) {
return true;
}
let (pubkey, is_agent) = {
let auth = conn.auth_state.read().await;
match &*auth {
AuthState::Authenticated(ctx) => (ctx.pubkey, ctx.agent_owner_pubkey.is_some()),
_ => return true,
}
};
let limits = &state.auth.config().rate_limits;
let (ws_window_secs, ws_limit) =
crate::admission::ws_admission_budget(limits.human_ws_events_per_sec);
let ws_result = crate::admission::check_principal(
state.admission_rate_limiter.as_ref(),
&conn.tenant,
&pubkey,
LimitType::WsEvents,
ws_window_secs,
ws_limit,
)
.await;
let sub_id = match msg {
ClientMessage::Req { sub_id, .. } => Some(sub_id.as_str()),
_ => None,
};
if !send_admission_result(conn, ws_result, sub_id) {
return false;
}
if is_event {
let message_limit = if is_agent {
limits.agent_standard_messages_per_min
} else {
limits.human_messages_per_min
};
let message_result = crate::admission::check_principal(
state.admission_rate_limiter.as_ref(),
&conn.tenant,
&pubkey,
LimitType::Messages,
60,
message_limit,
)
.await;
if !send_admission_result(conn, message_result, None) {
return false;
}
}
true
}
fn send_admission_result(
conn: &ConnectionState,
result: Result<(), crate::admission::AdmissionError>,
sub_id: Option<&str>,
) -> 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,
&format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"),
));
false
}
Err(crate::admission::AdmissionError::Unavailable) => {
metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "unavailable").increment(1);
conn.send(request_rejection_message(
sub_id,
"rate-limited: shared admission unavailable",
));
false
}
}
}
#[cfg(test)]
mod tests {
pub(crate) mod tests {
use super::*;
use std::sync::{Arc, Mutex};
use buzz_auth::AuthMethod;
use nostr::{EventBuilder, Keys, Kind};
/// A connection whose outbound frames a test can read back.
///
/// Lives here, next to `ConnectionState`, so the crate has one place that
/// knows how to build one. Shared with `crate::rejection`'s tests.
pub(crate) fn test_conn_with_auth(
auth: AuthState,
) -> (Arc<ConnectionState>, mpsc::Receiver<WsMessage>) {
let (send_tx, send_rx) = mpsc::channel(4);
let (ctrl_tx, _ctrl_rx) = mpsc::channel(4);
let conn = ConnectionState {
conn_id: Uuid::new_v4(),
tenant: TenantContext::resolved(
buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()),
"test.local".to_string(),
),
remote_addr: "127.0.0.1:1234".parse().expect("socket addr"),
auth_state: RwLock::new(auth),
subscriptions: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
send_tx,
ctrl_tx,
cancel: CancellationToken::new(),
backpressure_count: Arc::new(AtomicU8::new(0)),
grace_limit: 3,
};
(Arc::new(conn), send_rx)
}
/// An authenticated connection — the only state admission quotas apply to.
pub(crate) fn authenticated_state() -> AuthState {
AuthState::Authenticated(AuthContext {
pubkey: Keys::generate().public_key(),
scopes: Vec::new(),
channel_ids: None,
auth_method: AuthMethod::Nip42,
agent_owner_pubkey: None,
})
}
pub(crate) fn read_frame(rx: &mut mpsc::Receiver<WsMessage>) -> serde_json::Value {
match rx.try_recv().expect("a frame was sent") {
WsMessage::Text(text) => serde_json::from_str(&text).expect("valid JSON frame"),
other => panic!("unexpected websocket message: {other:?}"),
}
}
/// Drives the real `handle_text_message` with every handler permit held, so
/// the EVENT saturation branch is reached through production dispatch rather
/// than by calling its helpers directly.
///
/// This must go through `handle_text_message`: a test that renders the
/// rejection frame itself stays green when the call site inside the match
/// arm is reverted to a bare `NOTICE`.
#[tokio::test]
async fn saturated_handler_rejects_an_event_on_the_ok_channel() {
let state = crate::state::tests::test_state().await;
// An unauthenticated connection skips the admission quotas, so the
// semaphore is the only gate the frame can trip.
let (conn, mut rx) = test_conn_with_auth(AuthState::Failed);
let permits = state.handler_semaphore.available_permits();
let _held = Arc::clone(&state.handler_semaphore)
.acquire_many_owned(permits as u32)
.await
.expect("hold every handler permit");
let event = EventBuilder::new(Kind::TextNote, "hello")
.sign_with_keys(&Keys::generate())
.expect("sign event");
let event_id = event.id.to_hex();
let raw = serde_json::json!(["EVENT", event]).to_string();
handle_text_message(raw, Arc::clone(&conn), Arc::clone(&state)).await;
let frame = read_frame(&mut rx);
assert_eq!(
frame[0], "OK",
"an EVENT turned away for handler saturation must be rejected on the \
OK channel — a NOTICE carries no event id, so the client's pending \
publish cannot be settled and the send only times out"
);
assert_eq!(frame[1], event_id);
assert_eq!(frame[2], false);
assert_eq!(frame[3], "rate-limited: too many concurrent requests");
}
/// The REQ arm of the same branch still settles on CLOSED.
#[tokio::test]
async fn saturated_handler_rejects_a_req_on_the_closed_channel() {
let state = crate::state::tests::test_state().await;
let (conn, mut rx) = test_conn_with_auth(AuthState::Failed);
let permits = state.handler_semaphore.available_permits();
let _held = Arc::clone(&state.handler_semaphore)
.acquire_many_owned(permits as u32)
.await
.expect("hold every handler permit");
let raw = serde_json::json!(["REQ", "history-abc", {"kinds": [1]}]).to_string();
handle_text_message(raw, Arc::clone(&conn), Arc::clone(&state)).await;
let frame = read_frame(&mut rx);
assert_eq!(frame[0], "CLOSED");
assert_eq!(frame[1], "history-abc");
}
/// COUNT refusals follow NIP-45 and close the named query.
#[tokio::test]
async fn saturated_handler_rejects_a_count_on_the_closed_channel() {
let state = crate::state::tests::test_state().await;
let (conn, mut rx) = test_conn_with_auth(AuthState::Failed);
let permits = state.handler_semaphore.available_permits();
let _held = Arc::clone(&state.handler_semaphore)
.acquire_many_owned(permits as u32)
.await
.expect("hold every handler permit");
let raw = serde_json::json!(["COUNT", "count-abc", {"kinds": [1]}]).to_string();
handle_text_message(raw, Arc::clone(&conn), Arc::clone(&state)).await;
let frame = read_frame(&mut rx);
assert_eq!(frame[0], "CLOSED");
assert_eq!(frame[1], "count-abc");
assert_eq!(frame[2], "rate-limited: too many concurrent requests");
}
#[derive(Debug, Default)]
struct MockSinkState {
messages: Vec<WsMessage>,
@@ -834,19 +874,6 @@ mod tests {
.collect()
}
#[test]
fn req_rejections_are_subscription_scoped() {
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]));
}
#[tokio::test]
async fn send_loop_batches_queued_data_frames_into_one_flush() {
let (data_tx, data_rx) = mpsc::channel(MAX_WS_SEND_BATCH);

1
crates/buzz-relay/src/lib.rs
View File

@@ -4,6 +4,7 @@
mod admission;
mod build_info;
mod rejection;
/// REST API route handlers.
pub mod api;

336
crates/buzz-relay/src/rejection.rs Normal file
View File

@@ -0,0 +1,336 @@
//! How a rejected client frame is addressed back to the client.
//!
//! NIP-01 gives every request type its own acknowledgement channel, and a
//! rejection is only actionable if it travels on the same one: a REQ or COUNT
//! refusal settles on `CLOSED`, an EVENT on `OK`. Rejecting an EVENT with a bare
//! `NOTICE` leaves a client that tracks pending publishes by event id with
//! nothing to key on, so the send cannot fail — it can only time out.
use crate::admission::AdmissionError;
use crate::connection::{AuthState, ConnectionState};
use crate::protocol::{ClientMessage, RelayMessage};
use crate::state::AppState;
use buzz_auth::LimitType;
/// What a rejected client frame is correlated back to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RejectionTarget<'a> {
/// A REQ or COUNT names the query it opened.
Subscription(&'a str),
/// An EVENT names the event it submitted.
Event(nostr::EventId),
/// No per-request correlation exists — connection-scoped notice.
Connection,
}
/// Picks the acknowledgement channel a rejection of `msg` must travel on.
pub(crate) fn rejection_target_for(msg: &ClientMessage) -> RejectionTarget<'_> {
match msg {
ClientMessage::Req { sub_id, .. } | ClientMessage::Count { sub_id, .. } => {
RejectionTarget::Subscription(sub_id.as_str())
}
ClientMessage::Event(event) => RejectionTarget::Event(event.id),
_ => RejectionTarget::Connection,
}
}
/// Renders `reason` as the rejection frame `target`'s acknowledgement channel
/// expects.
pub(crate) fn request_rejection_message(target: RejectionTarget<'_>, reason: &str) -> String {
match target {
RejectionTarget::Subscription(sub_id) => RelayMessage::closed(sub_id, reason),
RejectionTarget::Event(event_id) => RelayMessage::ok(&event_id.to_hex(), false, reason),
RejectionTarget::Connection => RelayMessage::notice(reason),
}
}
/// Applies the WebSocket admission quotas to `msg`, returning whether it may be
/// handled. A rejection is addressed to the frame's own acknowledgement channel.
pub(crate) async fn enforce_ws_admission(
msg: &ClientMessage,
conn: &ConnectionState,
state: &AppState,
) -> bool {
let is_event = matches!(msg, ClientMessage::Event(_));
if !is_event && !matches!(msg, ClientMessage::Req { .. } | ClientMessage::Count { .. }) {
return true;
}
let (pubkey, is_agent) = {
let auth = conn.auth_state.read().await;
match &*auth {
AuthState::Authenticated(ctx) => (ctx.pubkey, ctx.agent_owner_pubkey.is_some()),
_ => return true,
}
};
let limits = &state.auth.config().rate_limits;
let (ws_window_secs, ws_limit) =
crate::admission::ws_admission_budget(limits.human_ws_events_per_sec);
let ws_result = crate::admission::check_principal(
state.admission_rate_limiter.as_ref(),
&conn.tenant,
&pubkey,
LimitType::WsEvents,
ws_window_secs,
ws_limit,
)
.await;
if !send_admission_result(conn, ws_result, msg) {
return false;
}
if is_event {
let message_limit = if is_agent {
limits.agent_standard_messages_per_min
} else {
limits.human_messages_per_min
};
let message_result = crate::admission::check_principal(
state.admission_rate_limiter.as_ref(),
&conn.tenant,
&pubkey,
LimitType::Messages,
60,
message_limit,
)
.await;
// The per-minute message quota only applies to EVENTs, and its
// rejection must be as correlatable as the burst quota's.
if !send_admission_result(conn, message_result, msg) {
return false;
}
}
true
}
/// Forwards an admission verdict to the client, returning whether the frame was
/// admitted.
///
/// The rejection target is derived from `msg` here rather than supplied by the
/// caller: every quota check in this module must address its rejection to the
/// rejected frame's own acknowledgement channel, so there is deliberately no way
/// for a call site to name a different one.
fn send_admission_result(
conn: &ConnectionState,
result: Result<(), AdmissionError>,
msg: &ClientMessage,
) -> bool {
let target = rejection_target_for(msg);
match result {
Ok(()) => true,
Err(AdmissionError::Exceeded { reset_in_secs }) => {
metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "quota").increment(1);
conn.send(request_rejection_message(
target,
&format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"),
));
false
}
Err(AdmissionError::Unavailable) => {
metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "unavailable").increment(1);
conn.send(request_rejection_message(
target,
"rate-limited: shared admission unavailable",
));
false
}
}
}
#[cfg(test)]
mod tests {
//! A rejected frame must be answerable on the acknowledgement channel the
//! client is actually waiting on.
//!
//! History: an over-quota EVENT used to be rejected with a bare
//! `["NOTICE", reason]`. A NOTICE carries no event id, and desktop/mobile
//! settle pending publishes only from an `OK` keyed by event id, so the
//! rejection was unaddressable: the send could not fail, it could only time
//! out (25s in Desktop, `PUBLISH_TIMEOUT_MS`) and surface as a message stuck
//! on "Sending…". Startup quota exhaustion made it routine in the first
//! seconds after launch.
//!
//! These tests drive the production rejection path — a real parsed
//! `ClientMessage` through `enforce_ws_admission` and
//! `send_admission_result` — and assert on the frame that reaches the
//! connection's outbound channel.
use std::sync::Arc;
use axum::extract::ws::Message as WsMessage;
use nostr::{EventBuilder, Keys, Kind};
use tokio::sync::mpsc;
use crate::connection::tests::{authenticated_state, read_frame, test_conn_with_auth};
use crate::connection::AuthState;
use super::*;
fn sent_frame(rx: &mut mpsc::Receiver<WsMessage>) -> serde_json::Value {
read_frame(rx)
}
fn test_conn() -> (Arc<ConnectionState>, mpsc::Receiver<WsMessage>) {
test_conn_with_auth(AuthState::Failed)
}
/// Parses a real EVENT frame exactly as the recv loop does, so the test is
/// coupled to production parsing and not to a hand-built target.
fn parsed_event_message() -> (ClientMessage, String) {
let event = EventBuilder::new(Kind::TextNote, "hello")
.sign_with_keys(&Keys::generate())
.expect("sign event");
let event_id = event.id.to_hex();
let frame = serde_json::json!(["EVENT", event]).to_string();
(ClientMessage::parse(&frame).expect("parse EVENT"), event_id)
}
/// The regression: an over-quota EVENT must be rejected with
/// `OK(event_id, false, reason)` so the client can settle the exact pending
/// publish it belongs to. A NOTICE here reintroduces the 25s send stall.
#[test]
fn over_quota_event_is_rejected_with_a_correlated_ok() {
let (conn, mut rx) = test_conn();
let (msg, event_id) = parsed_event_message();
let admitted = send_admission_result(
&conn,
Err(AdmissionError::Exceeded { reset_in_secs: 7 }),
&msg,
);
assert!(!admitted, "an over-quota frame is not admitted");
let frame = sent_frame(&mut rx);
assert_eq!(
frame[0], "OK",
"an EVENT rejection must travel on the OK channel — a NOTICE cannot \
be correlated to a pending publish, so the send hangs until the \
client's publish timeout instead of failing"
);
assert_eq!(
frame[1], event_id,
"the OK must name the rejected event id, which is what the client's \
pending-publish map is keyed by"
);
assert_eq!(frame[2], false, "and must be an explicit rejection");
assert_eq!(
frame[3], "rate-limited: quota exceeded; retry in 7s",
"the retry hint must survive so the client can arm its gate"
);
}
/// The same correlation is required when admission is unavailable rather
/// than exceeded — both branches strand a send if they emit a NOTICE.
#[test]
fn event_rejected_for_unavailable_admission_is_also_correlated() {
let (conn, mut rx) = test_conn();
let (msg, event_id) = parsed_event_message();
send_admission_result(&conn, Err(AdmissionError::Unavailable), &msg);
let frame = sent_frame(&mut rx);
assert_eq!(frame[0], "OK");
assert_eq!(frame[1], event_id);
assert_eq!(frame[2], false);
}
/// A REQ still settles on CLOSED, which carries the subscription id. This
/// pins the pre-existing behavior the fix must not disturb.
#[test]
fn over_quota_req_still_closes_the_subscription() {
let (conn, mut rx) = test_conn();
let raw = serde_json::json!(["REQ", "history-abc", {"kinds": [1]}]).to_string();
let msg = ClientMessage::parse(&raw).expect("parse REQ");
send_admission_result(
&conn,
Err(AdmissionError::Exceeded { reset_in_secs: 7 }),
&msg,
);
let frame = sent_frame(&mut rx);
assert_eq!(frame[0], "CLOSED");
assert_eq!(
frame[1], "history-abc",
"a REQ rejection must name the subscription it rejected"
);
assert_eq!(frame[2], "rate-limited: quota exceeded; retry in 7s");
}
/// NIP-45 uses `CLOSED(query_id, reason)` when a relay refuses a COUNT.
#[test]
fn over_quota_count_closes_the_query() {
let (conn, mut rx) = test_conn();
let raw = serde_json::json!(["COUNT", "count-abc", {"kinds": [1]}]).to_string();
let msg = ClientMessage::parse(&raw).expect("parse COUNT");
send_admission_result(
&conn,
Err(AdmissionError::Exceeded { reset_in_secs: 7 }),
&msg,
);
let frame = sent_frame(&mut rx);
assert_eq!(frame[0], "CLOSED");
assert_eq!(frame[1], "count-abc");
assert_eq!(frame[2], "rate-limited: quota exceeded; retry in 7s");
}
/// Drives the real entry point `handle_text_message` calls, so the wiring
/// between `enforce_ws_admission` and the target choice is under test and
/// not just the leaf renderer.
///
/// The state's Redis is deliberately unreachable, which makes admission
/// return `Unavailable` — a production rejection path that needs no live
/// quota burst to reach.
async fn enforce_against_unreachable_admission(raw: &str) -> serde_json::Value {
let state = crate::state::tests::test_state().await;
let (conn, mut rx) = test_conn_with_auth(authenticated_state());
let msg = ClientMessage::parse(raw).expect("parse client frame");
let admitted = enforce_ws_admission(&msg, &conn, &state).await;
assert!(!admitted, "an unadmitted frame must not be handled");
sent_frame(&mut rx)
}
#[tokio::test]
async fn enforce_ws_admission_rejects_an_event_on_the_ok_channel() {
let event = EventBuilder::new(Kind::TextNote, "hello")
.sign_with_keys(&Keys::generate())
.expect("sign event");
let event_id = event.id.to_hex();
let raw = serde_json::json!(["EVENT", event]).to_string();
let frame = enforce_against_unreachable_admission(&raw).await;
assert_eq!(
frame[0], "OK",
"the admission gate must reject an EVENT on the channel the client's \
pending publish is keyed by, or the send can only time out"
);
assert_eq!(frame[1], event_id);
assert_eq!(frame[2], false);
}
#[tokio::test]
async fn enforce_ws_admission_rejects_a_count_on_the_closed_channel() {
let raw = serde_json::json!(["COUNT", "count-abc", {"kinds": [1]}]).to_string();
let frame = enforce_against_unreachable_admission(&raw).await;
assert_eq!(frame[0], "CLOSED");
assert_eq!(frame[1], "count-abc");
}
#[tokio::test]
async fn enforce_ws_admission_rejects_a_req_on_the_closed_channel() {
let raw = serde_json::json!(["REQ", "history-abc", {"kinds": [1]}]).to_string();
let frame = enforce_against_unreachable_admission(&raw).await;
assert_eq!(frame[0], "CLOSED");
assert_eq!(frame[1], "history-abc");
}
}

7
crates/buzz-relay/src/state.rs
View File

@@ -1386,7 +1386,7 @@ impl std::fmt::Debug for AppState {
}
#[cfg(test)]
mod tests {
pub(crate) mod tests {
use super::*;
use crate::connection::{AuthState, ConnectionState};
use std::collections::HashMap;
@@ -1425,7 +1425,10 @@ mod tests {
(mgr, conn_id, rx, ctrl_rx, cancel, bp)
}
async fn test_state() -> Arc<AppState> {
/// A relay state whose Redis is deliberately unreachable, so admission
/// checks resolve to `AdmissionError::Unavailable` without any live
/// infrastructure. Shared with `crate::rejection`'s tests.
pub(crate) async fn test_state() -> Arc<AppState> {
let mut config = crate::config::Config::from_env().expect("default config loads");
config.require_relay_membership = false;
config.redis_url = "redis://127.0.0.1:1".to_string();

10
desktop/src/shared/api/readOnlyRelayClient.ts
View File

@@ -9,6 +9,10 @@ import {
type RelaySubscriptionFilter,
} from "@/shared/api/relayClientShared";
import { closeWebSocket } from "@/shared/api/relayWebSocketClose";
import {
activateRateLimitIfSignalled,
waitForRateLimit,
} from "@/shared/api/relayRateLimitGate";
import {
AUTH_TIMEOUT_MS,
HISTORY_TIMEOUT_MS,
@@ -107,7 +111,10 @@ export class ReadOnlyRelayClient {
async publishEvent(event: RelayEvent): Promise<void> {
await this.connect();
if (this.wsId === null) {
const generation = this.generation;
await waitForRateLimit();
if (generation !== this.generation || this.wsId === null) {
throw new Error("Read-only relay socket is not connected.");
}
@@ -281,6 +288,7 @@ export class ReadOnlyRelayClient {
if (success) {
publish.resolve();
} else {
activateRateLimitIfSignalled(message);
publish.reject(
new Error(message || "Observer relay rejected the event."),
);

156
desktop/src/shared/api/readOnlyRelayClientPublishRejection.test.mjs Normal file
View File

@@ -0,0 +1,156 @@
// ReadOnlyRelayClient publishes to inactive communities, but it shares the
// process-wide relay rate-limit gate with the primary session. Addressed EVENT
// refusals therefore need to settle this client's pending publish, arm the
// shared gate, and defer later sends until the advertised window expires.
import assert from "node:assert/strict";
import test from "node:test";
let fakeNow = 0;
const pendingTimers = new Map();
let nextTimerId = 1;
const sends = [];
globalThis.window = {
setTimeout: (fn, ms) => {
const id = nextTimerId++;
pendingTimers.set(id, { fn, fireAt: fakeNow + ms });
return id;
},
clearTimeout: (id) => pendingTimers.delete(id),
__TAURI_INTERNALS__: {
invoke: async (command, args) => {
if (command === "plugin:websocket|send") sends.push(args);
},
},
};
Date.now = () => fakeNow;
const { ReadOnlyRelayClient } = await import("./readOnlyRelayClient.ts");
const { activateRateLimit, isRateLimited, resetRateLimitGate } = await import(
"./relayRateLimitGate.ts"
);
function tickTo(ms) {
fakeNow = ms;
for (const [id, { fn, fireAt }] of Array.from(pendingTimers.entries())) {
if (fireAt <= fakeNow) {
pendingTimers.delete(id);
fn();
}
}
}
function reset() {
resetRateLimitGate();
fakeNow = 0;
pendingTimers.clear();
nextTimerId = 1;
sends.length = 0;
}
function connectedClient() {
const client = new ReadOnlyRelayClient("wss://inactive.example");
client.wsId = 7;
client.connect = async () => {};
return client;
}
function armPendingPublish(client, eventId) {
const settled = new Promise((resolve, reject) => {
client.publishes.set(eventId, {
resolve,
reject,
timeout: window.setTimeout(() => {}, 25_000),
});
});
return settled.then(
() => ({ status: "resolved" }),
(error) => ({ status: "rejected", error }),
);
}
function deliver(client, frame) {
return client.handleWsMessage(
{ type: "Text", data: JSON.stringify(frame) },
client.generation,
);
}
test("a rate-limited OK rejects the named publish and arms the shared gate", async () => {
reset();
const client = connectedClient();
const eventId = "a".repeat(64);
const settled = armPendingPublish(client, eventId);
await deliver(client, [
"OK",
eventId,
false,
"rate-limited: quota exceeded; retry in 4s",
]);
const outcome = await settled;
assert.equal(outcome.status, "rejected");
assert.match(outcome.error.message, /rate-limited/);
assert.equal(client.publishes.has(eventId), false);
assert.equal(isRateLimited(), true);
});
test("an ordinary OK rejection does not arm the shared gate", async () => {
reset();
const client = connectedClient();
const eventId = "b".repeat(64);
const settled = armPendingPublish(client, eventId);
await deliver(client, ["OK", eventId, false, "invalid: bad signature"]);
assert.equal((await settled).status, "rejected");
assert.equal(isRateLimited(), false);
});
test("publish waits outside its timeout and pending state, then sends and settles", async () => {
reset();
activateRateLimit(4);
const client = connectedClient();
const event = { id: "c".repeat(64), kind: 5 };
const published = client.publishEvent(event);
await Promise.resolve();
await Promise.resolve();
assert.equal(sends.length, 0, "EVENT must remain unsent while gated");
assert.equal(
client.publishes.has(event.id),
false,
"publish timeout and pending ownership start only after the gate expires",
);
assert.equal(pendingTimers.size, 1, "only the gate timer should be armed");
tickTo(4_000);
await Promise.resolve();
await Promise.resolve();
assert.equal(sends.length, 1);
assert.deepEqual(JSON.parse(sends[0].message.data), ["EVENT", event]);
assert.equal(client.publishes.has(event.id), true);
await deliver(client, ["OK", event.id, true, ""]);
await published;
assert.equal(client.publishes.has(event.id), false);
});
test("a disconnected client does not send after the shared gate expires", async () => {
reset();
activateRateLimit(4);
const client = connectedClient();
const event = { id: "d".repeat(64), kind: 5 };
const published = client.publishEvent(event);
await Promise.resolve();
client.disconnect();
tickTo(4_000);
await assert.rejects(published, /not connected/);
assert.equal(sends.length, 0);
assert.equal(client.publishes.has(event.id), false);
});

295
desktop/src/shared/api/relayClientPublishRejection.test.mjs Normal file
View File

@@ -0,0 +1,295 @@
// A relay rejection addressed to one event must settle that event's pending
// publish *and* arm the rate-limit gate.
//
// History: the relay rejected an over-quota EVENT with a bare
// `["NOTICE", "rate-limited: ..."]`. A NOTICE carries no event id, and
// `pendingEvents` is keyed by event id, so nothing settled — the publish sat
// until PUBLISH_TIMEOUT_MS (25s) and surfaced as a message stuck on
// "Sending…". Startup quota exhaustion made that routine in the first seconds
// after launch. The relay now rejects on the OK channel instead, so the gate
// arming that used to live in the NOTICE branch has to happen here too.
import assert from "node:assert/strict";
import test from "node:test";
const fakeNow = 0;
const pendingTimers = new Map();
let nextTimerId = 1;
const sendAttempts = [];
const deliveredFrames = [];
let sendTransport = async (args) => {
deliveredFrames.push(args);
};
globalThis.window = {
setTimeout: (fn, ms) => {
const id = nextTimerId++;
pendingTimers.set(id, { fn, fireAt: fakeNow + ms });
return id;
},
clearTimeout: (id) => pendingTimers.delete(id),
__TAURI_INTERNALS__: {
invoke: async (command, args) => {
if (command === "plugin:websocket|send") {
sendAttempts.push(args);
return sendTransport(args);
}
},
},
};
Date.now = () => fakeNow;
const { RelayClient } = await import("./relayClientSession.ts");
const { activateRateLimit, isRateLimited, resetRateLimitGate } = await import(
"./relayRateLimitGate.ts"
);
function reset() {
resetRateLimitGate();
pendingTimers.clear();
nextTimerId = 1;
sendAttempts.length = 0;
deliveredFrames.length = 0;
sendTransport = async (args) => {
deliveredFrames.push(args);
};
}
function connectedClient() {
const client = new RelayClient();
client.wsId = 7;
return client;
}
function eventFrames() {
return deliveredFrames.filter(
({ message }) => JSON.parse(message.data)[0] === "EVENT",
);
}
async function flushUntil(predicate, attempts = 20) {
for (let attempt = 0; attempt < attempts; attempt++) {
if (predicate()) return;
await Promise.resolve();
}
assert.fail("condition did not become true before the microtask limit");
}
function deferred() {
let resolve;
let reject;
const promise = new Promise((promiseResolve, promiseReject) => {
resolve = promiseResolve;
reject = promiseReject;
});
return { promise, resolve, reject };
}
/**
* Registers a pending publish the way `publishEvent` does, without needing a
* socket: the OK dispatch under test only reads `pendingEvents`.
*/
function armPendingPublish(client, eventId) {
const event = { id: eventId };
const settled = new Promise((resolve, reject) => {
client.pendingEvents.set(eventId, {
event,
resolve,
reject,
timeout: window.setTimeout(() => {}, 25_000),
});
});
// Keep the rejection from surfacing as an unhandled rejection.
return settled.then(
(value) => ({ status: "resolved", value }),
(error) => ({ status: "rejected", error }),
);
}
/** Feeds a raw relay frame through the real inbound dispatch path. */
function deliver(client, frame) {
return client.handleWsMessage(
{ type: "Text", data: JSON.stringify(frame) },
client.connectionGeneration,
);
}
test("a rate-limited OK rejection settles the pending publish", async () => {
resetRateLimitGate();
pendingTimers.clear();
const client = new RelayClient();
const eventId = "a".repeat(64);
const settled = armPendingPublish(client, eventId);
await deliver(client, [
"OK",
eventId,
false,
"rate-limited: quota exceeded; retry in 4s",
]);
const outcome = await settled;
assert.equal(
outcome.status,
"rejected",
"an over-quota publish must fail fast, not hang until the 25s publish timeout",
);
assert.match(outcome.error.message, /rate-limited/);
assert.equal(
client.pendingEvents.has(eventId),
false,
"the pending entry must be cleared",
);
});
test("a rate-limited OK rejection arms the rate-limit gate", async () => {
resetRateLimitGate();
pendingTimers.clear();
const client = new RelayClient();
const eventId = "b".repeat(64);
const settled = armPendingPublish(client, eventId);
assert.equal(isRateLimited(), false, "gate starts closed");
await deliver(client, [
"OK",
eventId,
false,
"rate-limited: quota exceeded; retry in 4s",
]);
await settled;
assert.equal(
isRateLimited(),
true,
"back-pressure now arrives on the OK channel — without arming here the " +
"client fails the send and immediately retries into the same quota",
);
});
test("an ordinary OK rejection does not arm the gate", async () => {
resetRateLimitGate();
pendingTimers.clear();
const client = new RelayClient();
const eventId = "c".repeat(64);
const settled = armPendingPublish(client, eventId);
await deliver(client, ["OK", eventId, false, "invalid: bad signature"]);
const outcome = await settled;
assert.equal(outcome.status, "rejected");
assert.equal(
isRateLimited(),
false,
"only `rate-limited:` rejections signal back-pressure",
);
});
test("an accepted OK still resolves the pending publish", async () => {
reset();
const client = new RelayClient();
const eventId = "d".repeat(64);
const settled = armPendingPublish(client, eventId);
await deliver(client, ["OK", eventId, true, ""]);
const outcome = await settled;
assert.equal(outcome.status, "resolved");
assert.equal(outcome.value.id, eventId);
});
test("a publish started during an ordinary outage reconnects once and settles", async () => {
reset();
const client = new RelayClient();
const event = { id: "0".repeat(64), kind: 1 };
let reconnects = 0;
client.ensureConnected = async () => {
reconnects++;
client.connectionGeneration++;
client.wsId = 8;
return client.connectionGeneration;
};
const published = client.publishEvent(event, "timed out", "send failed");
await flushUntil(() => eventFrames().length === 1);
assert.equal(reconnects, 1);
assert.equal(sendAttempts.length, 1);
assert.equal(client.pendingEvents.has(event.id), true);
await deliver(client, ["OK", event.id, true, ""]);
assert.equal(await published, event);
assert.equal(client.pendingEvents.size, 0);
});
test("a community switch while gated cannot publish through its replacement socket", async () => {
reset();
activateRateLimit(4);
const client = connectedClient();
const event = { id: "e".repeat(64), kind: 1 };
const published = client.publishEvent(event, "timed out", "send failed");
await Promise.resolve();
assert.equal(client.pendingEvents.size, 0);
client.disconnect();
resetRateLimitGate();
client.wsId = 8;
await assert.rejects(published, /community switch/);
assert.equal(client.pendingEvents.size, 0);
assert.equal(eventFrames().length, 0);
});
test("a community switch after send failure cannot retry through its replacement socket", async () => {
reset();
const client = connectedClient();
const event = { id: "f".repeat(64), kind: 1 };
const reconnect = deferred();
client.ensureConnected = async () => {
await reconnect.promise;
return client.connectionGeneration;
};
sendTransport = async () => {
throw new Error("old socket failed");
};
const published = client.publishEvent(event, "timed out", "send failed");
const outcome = published.then(
() => ({ status: "resolved" }),
(error) => ({ status: "rejected", error }),
);
await flushUntil(() => client.connectionGeneration === 1);
assert.equal(sendAttempts.length, 1);
assert.equal(eventFrames().length, 0);
assert.equal(
client.connectionGeneration,
1,
"the failed send reset its socket",
);
assert.equal(
client.pendingEvents.has(event.id),
true,
"the original publish remains owned while reconnect is pending",
);
client.disconnect();
client.wsId = 8;
const settledBeforeReconnect = await outcome;
assert.equal(
settledBeforeReconnect.status,
"rejected",
"community switch must settle the publish without waiting for reconnect",
);
assert.match(settledBeforeReconnect.error.message, /community switch/);
reconnect.resolve();
await published.catch(() => {});
await Promise.resolve();
assert.equal(client.pendingEvents.size, 0);
assert.equal(
sendAttempts.length,
1,
"the replacement socket must not be used",
);
assert.equal(eventFrames().length, 0);
});

99
desktop/src/shared/api/relayClientSession.ts
View File

@@ -38,11 +38,8 @@ import {
} from "@/shared/api/relayClosedRecovery";
import { getChannelReconnectRepairEvents } from "@/shared/api/channelReconnectRepair";
import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay";
import {
activateRateLimit,
parseRateLimitHint,
waitForRateLimit,
} from "@/shared/api/relayRateLimitGate";
import { publishSessionEvent } from "@/shared/api/relayEventPublisher";
import { activateRateLimitIfSignalled } from "@/shared/api/relayRateLimitGate";
import {
fetchChunkedHistory,
requestFirstEventGated,
@@ -64,7 +61,6 @@ import {
BACKOFF_RESET_STABLE_MS,
EVENT_BATCH_MS,
HISTORY_TIMEOUT_MS,
PUBLISH_TIMEOUT_MS,
RECONNECT_BASE_DELAY_MS,
RECONNECT_MAX_DELAY_MS,
STALL_CHECK_INTERVAL_MS,
@@ -82,7 +78,7 @@ import { buildThreadReferenceTags } from "@/features/messages/lib/threading";
export class RelayClient {
private wsId: number | null = null;
private relayUrl: string | null = null;
private connectPromise: Promise<void> | null = null;
private connectPromise: Promise<number> | null = null;
private reconnectTimeout: number | null = null;
private reconnectWaiters = new RelayReconnectWaiters();
private reconnectDelayMs = RECONNECT_BASE_DELAY_MS;
@@ -97,6 +93,7 @@ export class RelayClient {
private notifyReconnectListeners = false;
private onMessageChannel: Channel<unknown> | null = null;
private connectionGeneration = 0;
private sessionEpoch = 0;
private stabilityTimer: number | null = null;
private visibleChannelId: string | null = null;
private authOkTracker = new AuthOkTracker();
@@ -126,6 +123,7 @@ export class RelayClient {
this.stabilityTimer = null;
}
this.stallWatchdog.stop();
this.sessionEpoch++;
this.connectionGeneration++;
this.keepAliveRequested = false;
this.relayUrl = null;
@@ -496,7 +494,7 @@ export class RelayClient {
}
if (this.wsId !== null) {
return;
return this.connectionGeneration;
}
if (
@@ -507,14 +505,14 @@ export class RelayClient {
// The reconnect coordinator owns outage pacing. Query, publish, and
// subscription callers must wait for its scheduled attempt instead of
// clearing the timer and creating an immediate reconnect storm.
return this.reconnectWaiters.wait();
return this.reconnectWaiters.wait().then(() => this.connectionGeneration);
}
const connectPromise = this.connect();
this.connectPromise = connectPromise;
try {
await connectPromise;
return await connectPromise;
} finally {
if (this.connectPromise === connectPromise) {
this.connectPromise = null;
@@ -584,6 +582,7 @@ export class RelayClient {
await this.replayLiveSubscriptions();
this.stallWatchdog.start();
this.emitReconnectIfNeeded();
return generation;
} catch (error) {
const connectionError = this.normalizeRelayError(
error,
@@ -663,6 +662,17 @@ export class RelayClient {
});
}
private async sendRawForGeneration(payload: unknown[], generation: number) {
if (generation !== this.connectionGeneration || this.wsId === null) {
throw new Error("Relay publish was superseded by a session change.");
}
const wsId = this.wsId;
await invoke("plugin:websocket|send", {
id: wsId,
message: { type: "Text", data: JSON.stringify(payload) },
});
}
private normalizeRelayError(error: unknown, fallbackMessage: string) {
return error instanceof Error ? error : new Error(fallbackMessage);
}
@@ -712,47 +722,23 @@ export class RelayClient {
timeoutMessage: string,
sendErrorMessage: string,
) {
// Await the gate before sending EVENT; op timeout starts after the wait.
await waitForRateLimit();
return new Promise<RelayEvent>((resolve, reject) => {
const timeout = window.setTimeout(() => {
this.pendingEvents.delete(event.id);
reject(new Error(timeoutMessage));
}, PUBLISH_TIMEOUT_MS);
this.pendingEvents.set(event.id, {
event,
resolve,
reject,
timeout,
});
void this.sendRaw(["EVENT", event]).catch(async (error) => {
const pendingEvent = this.pendingEvents.get(event.id);
this.pendingEvents.delete(event.id);
const normalizedError = this.recoverFromSocketFailure(
error,
sendErrorMessage,
);
try {
await this.ensureConnected();
if (!pendingEvent) {
throw normalizedError;
}
this.pendingEvents.set(event.id, pendingEvent);
await this.sendRaw(["EVENT", event]);
} catch (retryError) {
window.clearTimeout(timeout);
this.pendingEvents.delete(event.id);
reject(
this.recoverFromSocketFailure(retryError, normalizedError.message),
);
}
});
});
return publishSessionEvent(
{
generation: () => this.connectionGeneration,
ownership: () => this.sessionEpoch,
pendingEvents: this.pendingEvents,
send: (payload, generation) =>
this.sendRawForGeneration(payload, generation),
reconnect: () => this.ensureConnected(),
normalizeError: (error, fallback) =>
this.normalizeRelayError(error, fallback),
recoverSocketFailure: (error, fallback) =>
this.recoverFromSocketFailure(error, fallback),
},
event,
timeoutMessage,
sendErrorMessage,
);
}
private async handleWsMessage(message: unknown, generation: number) {
@@ -829,11 +815,8 @@ export class RelayClient {
}
if (type === "NOTICE" && typeof rest[0] === "string") {
const notice: string = rest[0];
// Relay back-pressure — arm the gate until the window expires.
if (notice.startsWith("rate-limited:")) {
activateRateLimit(parseRateLimitHint(notice));
}
// Connection-scoped back-pressure — arm the gate until it expires.
activateRateLimitIfSignalled(rest[0]);
}
}
@@ -922,6 +905,10 @@ export class RelayClient {
if (success) {
pendingEvent.resolve(pendingEvent.event);
} else {
// Back-pressure now arrives here rather than as a NOTICE: the relay
// rejects an over-quota EVENT on the OK channel so this pending publish
// can be settled at all. Unarmed, the send retries into the same quota.
activateRateLimitIfSignalled(message);
pendingEvent.reject(new Error(message || "Relay rejected the event."));
}
}

84
desktop/src/shared/api/relayEventPublisher.ts Normal file
View File

@@ -0,0 +1,84 @@
import type { RelayEvent } from "@/shared/api/types";
import type { PendingEvent } from "@/shared/api/relayClientShared";
import { waitForRateLimit } from "@/shared/api/relayRateLimitGate";
import { PUBLISH_TIMEOUT_MS } from "@/shared/api/relayClientTimings";
type PublishSession = {
generation: () => number;
ownership: () => number;
pendingEvents: Map<string, PendingEvent>;
send: (payload: unknown[], generation: number) => Promise<void>;
reconnect: () => Promise<number>;
normalizeError: (error: unknown, fallback: string) => Error;
recoverSocketFailure: (error: unknown, fallback: string) => Error;
};
/** Publish once, with one reconnect retry, without crossing session ownership. */
export async function publishSessionEvent(
session: PublishSession,
event: RelayEvent,
timeoutMessage: string,
sendErrorMessage: string,
): Promise<RelayEvent> {
const publishOwnership = session.ownership();
await waitForRateLimit();
if (publishOwnership !== session.ownership()) {
throw new Error("Relay disconnected for community switch.");
}
const publishGeneration = session.generation();
return new Promise<RelayEvent>((resolve, reject) => {
const timeout = window.setTimeout(() => {
session.pendingEvents.delete(event.id);
reject(new Error(timeoutMessage));
}, PUBLISH_TIMEOUT_MS);
const pendingEvent = { event, resolve, reject, timeout };
session.pendingEvents.set(event.id, pendingEvent);
void session
.send(["EVENT", event], publishGeneration)
.catch(async (error) => {
// A disconnect may already have rejected this operation while the send
// was in flight. Its late failure must not reset the replacement session.
if (
publishOwnership !== session.ownership() ||
publishGeneration !== session.generation() ||
session.pendingEvents.get(event.id) !== pendingEvent
) {
return;
}
// Expected socket recovery must not reject the operation being retried.
session.pendingEvents.delete(event.id);
const sendError = session.recoverSocketFailure(error, sendErrorMessage);
session.pendingEvents.set(event.id, pendingEvent);
let retryGeneration: number | null = null;
try {
retryGeneration = await session.reconnect();
if (
publishOwnership !== session.ownership() ||
session.generation() !== retryGeneration ||
session.pendingEvents.get(event.id) !== pendingEvent
) {
throw new Error(
"Relay publish was superseded by a session change.",
);
}
await session.send(["EVENT", event], retryGeneration);
} catch (retryError) {
if (session.pendingEvents.get(event.id) !== pendingEvent) return;
window.clearTimeout(timeout);
session.pendingEvents.delete(event.id);
reject(
publishOwnership === session.ownership() &&
retryGeneration !== null &&
session.generation() === retryGeneration
? session.recoverSocketFailure(retryError, sendError.message)
: session.normalizeError(retryError, sendError.message),
);
}
});
});
}

18
desktop/src/shared/api/relayRateLimitGate.ts
View File

@@ -87,6 +87,24 @@ export function activateRateLimit(retryInSeconds: number | null): void {
}, durationMs);
}
/**
* Arms the gate if `message` is a relay back-pressure signal, and reports
* whether it was.
*
* The relay marks back-pressure with a `rate-limited:` prefix on whichever
* frame carries the rejection — `NOTICE` for connection-scoped limits, `OK`
* for one addressed to a single event, `CLOSED` for a subscription. Every
* inbound path needs the same test, so it lives here with the gate rather than
* being re-derived per call site.
*/
export function activateRateLimitIfSignalled(message: string): boolean {
if (!message.startsWith("rate-limited:")) {
return false;
}
activateRateLimit(parseRateLimitHint(message));
return true;
}
/** Returns `true` when the relay has signalled back-pressure and the gate is active. */
export function isRateLimited(): boolean {
return expiresAt !== null && Date.now() < expiresAt;

15
mobile/lib/shared/relay/relay_session.dart
View File

@@ -311,7 +311,13 @@ class RelaySessionNotifier extends Notifier<SessionState> {
Future<NostrEvent> publish(
NostrEvent event, {
Duration timeout = const Duration(seconds: 8),
}) {
}) async {
final generation = _connectionGeneration;
if (_rateLimitGate.isActive) await _rateLimitGate.wait();
if (!_isActiveConnection(generation) || !_socketConnected) {
throw StateError('Relay session is not connected');
}
final completer = Completer<NostrEvent>();
final timer = Timer(timeout, () {
@@ -824,6 +830,13 @@ class RelaySessionNotifier extends Notifier<SessionState> {
);
}
} else {
// Back-pressure now arrives here rather than as a NOTICE: the relay
// rejects an over-quota EVENT on the OK channel so this pending publish
// can be settled at all. Without arming the gate the send would fail
// without ever backing off.
if (message.startsWith('rate-limited:')) {
_rateLimitGate.activate(parseRateLimitRetrySeconds(message));
}
if (!pending.completer.isCompleted) {
pending.completer.completeError(
Exception(message.isNotEmpty ? message : 'Event rejected'),

148
mobile/test/shared/relay/relay_session_test.dart
View File

@@ -1382,6 +1382,150 @@ void main() {
expect(closedMessages, ['restricted: no longer valid']);
unsubscribe();
});
// The relay rejects an over-quota EVENT on the OK channel rather than with a
// bare NOTICE, because a NOTICE carries no event id and `_pendingEvents` is
// keyed by one — nothing settled, so the publish could only time out. The
// gate arming that used to depend on the NOTICE has to happen here too.
test(
'a rate-limited OK rejection fails the publish and arms the gate',
() async {
final gateTimers = <_ManualTimer>[];
final gate = RelayRateLimitGate(
now: () => DateTime(2026),
timerFactory: (duration, callback) {
final timer = _ManualTimer(duration, callback);
gateTimers.add(timer);
return timer;
},
);
final session = RelaySessionNotifier(rateLimitGate: gate);
session.debugAttachSocketForTest(_RecordingRelaySocket());
final publish = session.publish(_event());
session.debugHandleMessage([
'OK',
'event-1',
false,
'rate-limited: quota exceeded; retry in 4s',
]);
await expectLater(publish, throwsA(isA<Exception>()));
expect(
gate.isActive,
isTrue,
reason:
'back-pressure now arrives on the OK channel — without arming here '
'the client fails the send and retries into the same quota',
);
expect(gateTimers.single.duration, const Duration(seconds: 4));
},
);
test(
'publish waits out the rate-limit gate before timeout registration and send',
() async {
final gateTimers = <_ManualTimer>[];
final gate = RelayRateLimitGate(
now: () => DateTime(2026),
timerFactory: (duration, callback) {
final timer = _ManualTimer(duration, callback);
gateTimers.add(timer);
return timer;
},
);
final socket = _RecordingRelaySocket();
final session = RelaySessionNotifier(rateLimitGate: gate);
session.debugAttachSocketForTest(socket);
final firstPublish = session.publish(_event(id: 'event-a'));
session.debugHandleMessage([
'OK',
'event-a',
false,
'rate-limited: quota exceeded; retry in 4s',
]);
await expectLater(firstPublish, throwsA(isA<Exception>()));
var secondSettled = false;
final secondPublish = session.publish(
_event(id: 'event-b'),
timeout: Duration.zero,
);
unawaited(secondPublish.whenComplete(() => secondSettled = true));
await Future<void>.delayed(Duration.zero);
expect(
socket.messages.where((message) => message.first == 'EVENT'),
hasLength(1),
reason: 'the next EVENT must remain unsent while the gate is active',
);
expect(
secondSettled,
isFalse,
reason:
'the publish timeout must not start until after the gate expires',
);
gateTimers.single.fire();
await Future<void>.microtask(() {});
final events = socket.messages
.where((message) => message.first == 'EVENT')
.toList();
expect(events, hasLength(2));
expect((events.last[1] as Map<String, dynamic>)['id'], 'event-b');
session.debugHandleMessage(['OK', 'event-b', true, '']);
expect((await secondPublish).id, 'event-b');
},
);
test(
'a gated publish is cancelled if the connection changes while waiting',
() async {
final gateTimers = <_ManualTimer>[];
final gate = RelayRateLimitGate(
now: () => DateTime(2026),
timerFactory: (duration, callback) {
final timer = _ManualTimer(duration, callback);
gateTimers.add(timer);
return timer;
},
);
final socket = _RecordingRelaySocket();
final session = RelaySessionNotifier(rateLimitGate: gate);
session.debugAttachSocketForTest(socket);
gate.activate(4);
final publish = session.publish(_event(id: 'event-b'));
session.debugSupersedeConnection();
gateTimers.single.fire();
await expectLater(publish, throwsA(isA<StateError>()));
expect(socket.messages, isEmpty);
},
);
test('an ordinary OK rejection does not arm the gate', () async {
final gate = RelayRateLimitGate(now: () => DateTime(2026));
final session = RelaySessionNotifier(rateLimitGate: gate);
session.debugAttachSocketForTest(_RecordingRelaySocket());
final publish = session.publish(_event());
session.debugHandleMessage([
'OK',
'event-1',
false,
'invalid: bad signature',
]);
await expectLater(publish, throwsA(isA<Exception>()));
expect(
gate.isActive,
isFalse,
reason: 'only `rate-limited:` rejections signal back-pressure',
);
});
}
class _ControlledHttpClient extends http.BaseClient {
@@ -1524,9 +1668,9 @@ class _FakeRelayConfigNotifier extends RelayConfigNotifier {
RelayConfig build() => RelayConfig(baseUrl: _baseUrl, nsec: _nsec);
}
NostrEvent _event({int createdAt = 20}) {
NostrEvent _event({int createdAt = 20, String id = 'event-1'}) {
return NostrEvent(
id: 'event-1',
id: id,
pubkey: 'alice',
createdAt: createdAt,
kind: EventKind.streamMessageV2,