From 5511b56fcf0047f9d0d4887dc75125b6933a6bbd Mon Sep 17 00:00:00 2001 From: Wes Date: Sat, 19 Sep 2026 01:39:37 -0700 Subject: [PATCH] fix(relay): exclude ephemeral activity from message quota (#7736) ## Summary Stop charging ephemeral WebSocket activity against the per-minute `Messages` quota. Presence, typing, and observer traffic should not consume the allowance intended for persistable event attempts. - Use the existing `buzz_core::kind::is_ephemeral` predicate (kinds `20000..=29999`) at the second admission gate. All authenticated EVENT/REQ/COUNT traffic still consumes the shared `WsEvents` budget first. Non-ephemeral EVENT attempts remain charged, including attempts later rejected by authorization or validation. - Add bounded `bucket` labels (`messages`, `ws_operations`, `api_calls`) to `buzz_admission_rejections_total`. HTTP changes are metric labeling only. Wire rejection text remains byte-for-byte unchanged, preserving existing client parsers. - Add real-Redis admission regressions and select them in the existing relay integration CI job. No quota increases, identity/keying changes, client changes, new limiter, or deployment. Downstream authorization and fail-closed Redis behavior are unchanged. ### Related issue Duplicate search found overlapping work: open #4912 and closed #4902. They propose the same ephemeral-accounting correction alongside broader wire/client/config changes. The relay fix is a deliberately narrower four-file implementation on the current admission module, with metrics-only diagnostics and real-Redis regression coverage. A separate test-only CI repair is documented below. It does not close or claim to supersede those PRs. Related symptom: #7411. This PR does **not** resolve subscription bursts exhausting the shared WS budget; that budget is intentionally preserved. It also does not establish or fix the cause of HTTP quota exhaustion. ### Testing Validation ran against base `779af8886caae1317b4de962082429867ab61503` plus the exact patch now committed as `70ea0528594e25e6dcb40cd74620bd5ef0177592` (binary diff SHA-256 `5064e1cbb87ecb9eff10336bae14b912cac442e1edc247aa4f093de30e4f9717`). Normal commit/push hooks subsequently passed on the committed head, without bypass. Passed: - Independent review and rerun: 11 rejection tests (including real Redis) plus 4 admission tests. Covers ephemeral range boundaries, human/agent budgets, persistable events, ephemeral admission after message exhaustion, mixed EVENT/REQ/COUNT flood exhaustion, Redis outage, exact wire payloads, and bucket metrics. Three mutations were caught: reintroducing ephemeral billing, exempting stored events, and bypassing WS rejection. - `just test-unit`: all 17 runner groups; `just test-integration`: all 3 runner groups. Auth integration reported no tests; ignored database integration tests are not claimed as executed. - Relay all-target Clippy, formatting, workflow actionlint, pre-commit/signoff hooks, and all applicable pre-push lanes (`push-head-scope`, `branch-skew`, `file-size-check`, `rust-tests`, `desktop-tauri-checks`). - Release relay/CLI/admin build and isolated release-binary workflow: channel creation, signed root/reply sends, exact-ID readback, and thread retrieval. Debug and release WS smoke admitted three presence events with a Messages limit of two, admitted two persistable events, rejected the third, still admitted presence afterward, and rejected author mismatch. Debug Prometheus confirmed `bucket="messages"`. **Outstanding validation gap:** the additional `cargo test -p buzz-relay --no-fail-fast` run was not green: library 1,055 passed / 2 failed / 100 ignored; binary 13 passed / 1 ignored; boot integration 9 passed. Failures were `api::mesh_demo::tests::demo_join_forwarded_arm_round_trips_echo` (504 instead of 200) and `telemetry::tests::trace_context_lookup_does_not_enable_callsites` (tracing assertion). Neither failure was diagnosed or repaired in this scope, and this PR does not claim a fully green package suite. `just ci` was not run locally; hosted CI remains to validate the PR. All runtime tests used isolated local services. Owned services have been stopped; nothing was deployed and no production data was touched. ### CI cancellation-test repair At the original relay head, hosted Rust unit tests failed on the pre-existing cancellation harness race tracked in #4945. With owner authorization, this PR now adapts the one-file repair from #5003 (original commit `cc8dc6124957acfd67fe2336396b58e0ff43ec60`), preserving author credit. #5003's old hosted workflows failed without starting jobs; their status is not used as validation evidence. Commit `55fe5f16618a51d03ad48d6f2f97b2f8b463607a` holds the fake provider response until a successful cancellation acknowledgement, then replays all pre-acknowledgement usage/prompt frames in order. It retains the cancellation and usage-before-response assertions and explicitly requires prompt completion. No production code changes in this follow-up. Validation on parent `70ea052859` plus this test patch, before the pre-commit-only assertion wrapping: full `cargo test -p buzz-agent --no-fail-fast` passed (715 tests, one ignored); 100 targeted repetitions passed with four concurrent workers; all-target buzz-agent Clippy passed. The unmodified parent also passed 50 local repetitions, so this is not claimed as a local reproduction of the intermittent CI failure. The synchronization defect was confirmed from the source, original CI log, and #4945. Hosted CI must validate the updated head. Independent source review cleared exact commit `55fe5f16618a51d03ad48d6f2f97b2f8b463607a`. Normal pre-commit/signoff and all applicable pre-push hooks passed without bypass. The original relay-package mesh-demo and telemetry failures remain disclosed above and were not modified by this test-only repair. --------- Signed-off-by: Wes Co-authored-by: Carl <32a2e2c9d428ee08902cab75d956da2c1d235a22d4766b0dd4138bf6e2e5db1d@buzz.block.builderlab.xyz> Co-authored-by: kiranmagic7 <262980978+kiranmagic7@users.noreply.github.com> --- .github/workflows/_ci-relay.yml | 8 + crates/buzz-agent/tests/fake_llm.rs | 36 ++- crates/buzz-relay/src/api/bridge.rs | 4 +- crates/buzz-relay/src/rejection.rs | 29 +- .../buzz-relay/src/rejection/quota_tests.rs | 250 ++++++++++++++++++ 5 files changed, 304 insertions(+), 23 deletions(-) create mode 100644 crates/buzz-relay/src/rejection/quota_tests.rs diff --git a/.github/workflows/_ci-relay.yml b/.github/workflows/_ci-relay.yml index 388d48dd94..5776f37fe3 100644 --- a/.github/workflows/_ci-relay.yml +++ b/.github/workflows/_ci-relay.yml @@ -497,6 +497,14 @@ jobs: --archive-file target/ci/backend-integration-tests.tar.zst \ -E 'package(buzz-relay) and test(/api::git::settings::tests::external_infra::/)' \ --run-ignored ignored-only + - name: Relay quota accounting and rejection regressions + env: + BUZZ_TEST_REDIS_URL: redis://localhost:6379 + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-relay) and (test(/^rejection::/) or test(/^admission::/))' \ + --run-ignored all - name: Workflow message provenance unit tests # The relay's workflow_sink suite is not selected by the infra-free # unit job. Its ignored database cases run in the isolated PostgreSQL diff --git a/crates/buzz-agent/tests/fake_llm.rs b/crates/buzz-agent/tests/fake_llm.rs index 9822243d5f..2451deae08 100644 --- a/crates/buzz-agent/tests/fake_llm.rs +++ b/crates/buzz-agent/tests/fake_llm.rs @@ -1320,7 +1320,7 @@ async fn mid_turn_usage_includes_earlier_turns() { /// Setup: round 1 is a tool call WITH usage (tokens are captured). After the /// tool_call_update notification (proving round 1 is fully processed), we gate /// the round-2 LLM response behind a `oneshot` barrier that only releases after -/// cancel is sent. This guarantees the turn exits with `stopReason: "cancelled"` +/// cancel is acknowledged. This guarantees the turn exits with `stopReason: "cancelled"` /// deterministically, even on a slow CI worker. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn cancelled_turn_with_usage_emits_notification_before_response() { @@ -1335,7 +1335,7 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() { // in-flight TCP request can resolve. The queue is empty for round 2, so the // agent receives the fallback "no canned response" body which it treats as // an LLM error; the cancel check at the round boundary fires first because - // the gate is only released after cancel is enqueued. + // the gate is only released after cancel is acknowledged. let responses = vec![openai_tool_call_with_usage( "call_cancel_test", "fake__noop", @@ -1371,7 +1371,7 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() { } } // For request 2+ (round 2), wait for the gate to open before - // responding. This ensures cancel is sent before round 2 resolves, + // responding. This ensures cancel is processed before round 2 resolves, // making stopReason: cancelled deterministic. if req_num >= 2 { let rx = gate.lock().await.take(); @@ -1415,20 +1415,27 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() { }) .await; - // Now send cancel and release the round-2 gate. Cancel is enqueued before - // round 2 can respond, so the turn exits with stopReason: cancelled. + // Writing to stdin does not prove the agent processed cancel. Keep the + // provider blocked until the acknowledgement, retaining all earlier frames + // so usage/prompt ordering is checked even if the turn finishes before ACK. let c_id = h.send("session/cancel", json!({"sessionId": sid})).await; + let (frames_before_cancel_ack, cancel_ack) = + recv_until_with_drain(&mut h, |v| v["id"] == json!(c_id)).await; + assert_eq!(cancel_ack.get("result"), Some(&Value::Null), "{cancel_ack}"); + assert!(cancel_ack.get("error").is_none(), "{cancel_ack}"); let _ = gate_tx.send(()); // unblock round 2 let mut saw_usage_before_prompt_response = false; let mut saw_usage = false; - let mut saw_cancel_ok = false; let mut saw_prompt_response = false; - for _ in 0..40 { - let v = h.recv().await; - if v["id"] == json!(c_id) { - saw_cancel_ok = true; - } else if is_usage_update(&v) { + let mut pending_frames = VecDeque::from(frames_before_cancel_ack); + let frame_budget = 40 + pending_frames.len(); + for _ in 0..frame_budget { + let v = match pending_frames.pop_front() { + Some(v) => v, + None => h.recv().await, + }; + if is_usage_update(&v) { saw_usage = true; if !saw_prompt_response { saw_usage_before_prompt_response = true; @@ -1441,11 +1448,14 @@ async fn cancelled_turn_with_usage_emits_notification_before_response() { "turn must end with stopReason: cancelled" ); } - if saw_usage && saw_prompt_response && saw_cancel_ok { + if saw_usage && saw_prompt_response { break; } } - assert!(saw_cancel_ok, "session/cancel was not acknowledged"); + assert!( + saw_prompt_response, + "session/prompt did not finish after cancel" + ); assert!( saw_usage, "expected usage_update notification for cancelled turn with observed tokens" diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 37c549610d..6ec0483cb9 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -39,14 +39,14 @@ pub(crate) async fn enforce_http_admission( { Ok(()) => Ok(()), Err(crate::admission::AdmissionError::Exceeded { reset_in_secs }) => { - metrics::counter!("buzz_admission_rejections_total", "transport" => "http", "reason" => "quota").increment(1); + metrics::counter!("buzz_admission_rejections_total", "transport" => "http", "reason" => "quota", "bucket" => "api_calls").increment(1); Err(api_error( StatusCode::TOO_MANY_REQUESTS, &format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"), )) } Err(crate::admission::AdmissionError::Unavailable) => { - metrics::counter!("buzz_admission_rejections_total", "transport" => "http", "reason" => "unavailable").increment(1); + metrics::counter!("buzz_admission_rejections_total", "transport" => "http", "reason" => "unavailable", "bucket" => "api_calls").increment(1); Err(api_error( StatusCode::SERVICE_UNAVAILABLE, "rate-limited: shared admission unavailable", diff --git a/crates/buzz-relay/src/rejection.rs b/crates/buzz-relay/src/rejection.rs index 733c427f95..1fb760ec4d 100644 --- a/crates/buzz-relay/src/rejection.rs +++ b/crates/buzz-relay/src/rejection.rs @@ -75,11 +75,12 @@ pub(crate) async fn enforce_ws_admission( ws_limit, ) .await; - if !send_admission_result(conn, ws_result, msg) { + if !send_admission_result(conn, ws_result, msg, "ws_operations") { return false; } - if is_event { + if matches!(msg, ClientMessage::Event(event) if !buzz_core::kind::is_ephemeral(event.kind.as_u16() as u32)) + { let message_limit = if is_agent { limits.agent_standard_messages_per_min } else { @@ -94,9 +95,9 @@ pub(crate) async fn enforce_ws_admission( 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) { + // Only persistable EVENT attempts consume the minute quota. Ephemeral activity + // still passes the shared WS flood limit above and normal event auth. + if !send_admission_result(conn, message_result, msg, "messages") { return false; } } @@ -115,12 +116,13 @@ fn send_admission_result( conn: &ConnectionState, result: Result<(), AdmissionError>, msg: &ClientMessage, + bucket: &'static str, ) -> 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); + metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "quota", "bucket" => bucket).increment(1); conn.send(request_rejection_message( target, &format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"), @@ -128,7 +130,7 @@ fn send_admission_result( false } Err(AdmissionError::Unavailable) => { - metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "unavailable").increment(1); + metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "unavailable", "bucket" => bucket).increment(1); conn.send(request_rejection_message( target, "rate-limited: shared admission unavailable", @@ -198,6 +200,7 @@ mod tests { &conn, Err(AdmissionError::Exceeded { reset_in_secs: 7 }), &msg, + "ws_operations", ); assert!(!admitted, "an over-quota frame is not admitted"); @@ -227,7 +230,12 @@ mod tests { let (conn, mut rx) = test_conn(); let (msg, event_id) = parsed_event_message(); - send_admission_result(&conn, Err(AdmissionError::Unavailable), &msg); + send_admission_result( + &conn, + Err(AdmissionError::Unavailable), + &msg, + "ws_operations", + ); let frame = sent_frame(&mut rx); assert_eq!(frame[0], "OK"); @@ -247,6 +255,7 @@ mod tests { &conn, Err(AdmissionError::Exceeded { reset_in_secs: 7 }), &msg, + "ws_operations", ); let frame = sent_frame(&mut rx); @@ -269,6 +278,7 @@ mod tests { &conn, Err(AdmissionError::Exceeded { reset_in_secs: 7 }), &msg, + "ws_operations", ); let frame = sent_frame(&mut rx); @@ -333,3 +343,6 @@ mod tests { assert_eq!(frame[1], "history-abc"); } } + +#[cfg(test)] +mod quota_tests; diff --git a/crates/buzz-relay/src/rejection/quota_tests.rs b/crates/buzz-relay/src/rejection/quota_tests.rs new file mode 100644 index 0000000000..9f546469e2 --- /dev/null +++ b/crates/buzz-relay/src/rejection/quota_tests.rs @@ -0,0 +1,250 @@ +//! Real Redis counters through the production admission entrypoints. Tests use +//! disposable identities and expiring keys, never FLUSHDB or a shared identity. + +use super::*; +use axum::extract::ws::Message as WsMessage; +use buzz_auth::{AuthService, RateLimitConfig}; +use buzz_pubsub::rate_limiter::RedisRateLimiter; +use metrics_util::debugging::{DebugValue, DebuggingRecorder}; +use nostr::{EventBuilder, Keys, Kind}; +use std::sync::Arc; +use tokio::sync::mpsc; + +use crate::connection::tests::{authenticated_state, read_frame, test_conn_with_auth}; + +async fn state_with_limits(messages: u64, agent_messages: u64, ws: u64) -> AppState { + let url = std::env::var("BUZZ_TEST_REDIS_URL").expect("explicit isolated BUZZ_TEST_REDIS_URL"); + let mut state = (*crate::state::tests::test_state().await).clone(); + let mut auth = state.auth.config().clone(); + auth.rate_limits = RateLimitConfig { + human_messages_per_min: messages, + agent_standard_messages_per_min: agent_messages, + human_ws_events_per_sec: ws, + human_api_calls_per_min: 2, + ..RateLimitConfig::default() + }; + state.auth = Arc::new(AuthService::new(auth)); + state.redis_pool = deadpool_redis::Config::from_url(url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .unwrap(); + state.admission_rate_limiter = Arc::new(RedisRateLimiter::new(state.redis_pool.clone())); + state +} + +fn connection(agent: bool) -> (Arc, mpsc::Receiver) { + let AuthState::Authenticated(mut auth) = authenticated_state() else { + unreachable!() + }; + if agent { + auth.agent_owner_pubkey = Some(Keys::generate().public_key()); + } + test_conn_with_auth(AuthState::Authenticated(auth)) +} + +fn event(kind: u16) -> ClientMessage { + let event = EventBuilder::new(Kind::from(kind), "") + .sign_with_keys(&Keys::generate()) + .unwrap(); + ClientMessage::parse(&serde_json::json!(["EVENT", event]).to_string()).unwrap() +} + +fn pubkey(conn: &ConnectionState) -> nostr::PublicKey { + let AuthState::Authenticated(auth) = conn.auth_state_snapshot() else { + unreachable!() + }; + auth.pubkey +} + +async fn counter(state: &AppState, conn: &ConnectionState, bucket: LimitType) -> Option { + let key = buzz_auth::rate_limit::rate_limit_key(&conn.tenant, &pubkey(conn), &bucket); + let mut redis = state.redis_pool.get().await.unwrap(); + redis::cmd("GET") + .arg(key) + .query_async(&mut *redis) + .await + .unwrap() +} + +fn assert_metric( + recorder: &DebuggingRecorder, + transport: &str, + reason: &str, + bucket: &str, + count: u64, +) { + let snapshot = recorder.snapshotter().snapshot().into_vec(); + let labels = [ + ("transport", transport), + ("reason", reason), + ("bucket", bucket), + ]; + let actual = snapshot.iter().find_map(|(key, _, _, value)| { + (key.key().name() == "buzz_admission_rejections_total" + && labels + .iter() + .all(|(k, v)| key.key().labels().any(|l| l.key() == *k && l.value() == *v))) + .then(|| match value { + DebugValue::Counter(value) => *value, + _ => panic!("expected counter"), + }) + }); + assert_eq!(actual, Some(count)); +} + +fn assert_quota_frame(frame: &serde_json::Value, msg: &ClientMessage, max_retry: u64) { + let reason = match msg { + ClientMessage::Event(event) => { + assert_eq!(frame[0], "OK"); + assert_eq!(frame[1], event.id.to_hex()); + assert_eq!(frame[2], false); + frame[3].as_str().unwrap() + } + ClientMessage::Req { sub_id, .. } | ClientMessage::Count { sub_id, .. } => { + assert_eq!(frame[0], "CLOSED"); + assert_eq!(frame[1], *sub_id); + frame[2].as_str().unwrap() + } + _ => panic!("unexpected frame"), + }; + // Exact grammar: clients including buzz-app anchor both ends of this hint. + let seconds: u64 = reason + .strip_prefix("rate-limited: quota exceeded; retry in ") + .unwrap() + .strip_suffix('s') + .unwrap() + .parse() + .unwrap(); + assert!(seconds <= max_retry); +} + +#[tokio::test] +#[ignore = "requires explicit isolated BUZZ_TEST_REDIS_URL"] +async fn ephemeral_events_leave_human_and_agent_message_budgets_untouched() { + let state = state_with_limits(2, 3, 100).await; + let recorder = DebuggingRecorder::new(); + let _guard = metrics::set_default_local_recorder(&recorder); + // Both range boundaries, typing, presence, and observer frames. Adjacent + // non-ephemeral kinds below must still consume the message allowance. + let ephemeral_kinds = [20000, 20001, 20002, 24200, 29999]; + for (agent, limit) in [(false, 2), (true, 3)] { + for stored_kind in [0, 1, 9, 19999, 30000, 40002] { + let (conn, mut rx) = connection(agent); + for kind in ephemeral_kinds { + assert!(enforce_ws_admission(&event(kind), &conn, &state).await); + } + assert_eq!(counter(&state, &conn, LimitType::Messages).await, None); + assert_eq!(counter(&state, &conn, LimitType::WsEvents).await, Some(5)); + for expected in 1..=limit { + assert!(enforce_ws_admission(&event(stored_kind), &conn, &state).await); + assert_eq!( + counter(&state, &conn, LimitType::Messages).await, + Some(expected) + ); + } + let rejected = event(stored_kind); + assert!(!enforce_ws_admission(&rejected, &conn, &state).await); + assert_quota_frame(&read_frame(&mut rx), &rejected, 60); + for kind in ephemeral_kinds { + assert!(enforce_ws_admission(&event(kind), &conn, &state).await); + } + assert_eq!( + counter(&state, &conn, LimitType::Messages).await, + Some(limit + 1) + ); + assert!(rx.try_recv().is_err()); + } + } + assert_metric(&recorder, "websocket", "quota", "messages", 12); +} + +#[tokio::test] +#[ignore = "requires explicit isolated BUZZ_TEST_REDIS_URL"] +async fn ws_flood_budget_is_shared_by_ephemeral_stored_req_and_count() { + let state = state_with_limits(10, 20, 1).await; // Five operations per window. + let recorder = DebuggingRecorder::new(); + let _guard = metrics::set_default_local_recorder(&recorder); + let (conn, mut rx) = connection(false); + let req = ClientMessage::parse(r#"["REQ","history",{"kinds":[9]}]"#).unwrap(); + let count = ClientMessage::parse(r#"["COUNT","count",{"kinds":[9]}]"#).unwrap(); + for msg in [ + event(20001), + event(20002), + event(9), + req.clone(), + count.clone(), + ] { + assert!(enforce_ws_admission(&msg, &conn, &state).await); + } + assert_eq!(counter(&state, &conn, LimitType::Messages).await, Some(1)); + for msg in [event(24200), event(9), req, count] { + assert!(!enforce_ws_admission(&msg, &conn, &state).await); + assert_quota_frame(&read_frame(&mut rx), &msg, 5); + } + assert_eq!(counter(&state, &conn, LimitType::Messages).await, Some(1)); + assert_metric(&recorder, "websocket", "quota", "ws_operations", 4); +} + +#[tokio::test] +#[ignore = "requires explicit isolated BUZZ_TEST_REDIS_URL"] +async fn http_quota_preserves_wire_error_and_reports_api_bucket() { + let state = state_with_limits(0, 0, 0).await; + let recorder = DebuggingRecorder::new(); + let _guard = metrics::set_default_local_recorder(&recorder); + let (conn, _) = connection(false); + for _ in 0..2 { + crate::api::bridge::enforce_http_admission(&state, &conn.tenant, &pubkey(&conn)) + .await + .unwrap(); + } + let (status, body) = + crate::api::bridge::enforce_http_admission(&state, &conn.tenant, &pubkey(&conn)) + .await + .unwrap_err(); + assert_eq!(status, axum::http::StatusCode::TOO_MANY_REQUESTS); + let seconds: u64 = body.0["error"] + .as_str() + .unwrap() + .strip_prefix("rate-limited: quota exceeded; retry in ") + .unwrap() + .strip_suffix('s') + .unwrap() + .parse() + .unwrap(); + assert!(seconds <= 60); + assert_eq!(counter(&state, &conn, LimitType::Messages).await, None); + assert_eq!(counter(&state, &conn, LimitType::WsEvents).await, None); + assert_metric(&recorder, "http", "quota", "api_calls", 1); +} + +#[tokio::test] +async fn ephemeral_and_http_admission_still_fail_closed_when_redis_is_unavailable() { + let state = crate::state::tests::test_state().await; + let recorder = DebuggingRecorder::new(); + let _guard = metrics::set_default_local_recorder(&recorder); + let (conn, mut rx) = connection(false); + let msg = event(20002); + assert!(!enforce_ws_admission(&msg, &conn, &state).await); + let ClientMessage::Event(event) = msg else { + unreachable!() + }; + assert_eq!( + read_frame(&mut rx), + serde_json::json!([ + "OK", + event.id.to_hex(), + false, + "rate-limited: shared admission unavailable" + ]) + ); + assert_metric(&recorder, "websocket", "unavailable", "ws_operations", 1); + let (status, body) = + crate::api::bridge::enforce_http_admission(&state, &conn.tenant, &pubkey(&conn)) + .await + .unwrap_err(); + assert_eq!(status, axum::http::StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + body.0["error"], + "rate-limited: shared admission unavailable" + ); + assert_metric(&recorder, "http", "unavailable", "api_calls", 1); +}