From 891208914ed51c55b11ef6102f10ed08b5d9bb30 Mon Sep 17 00:00:00 2001 From: ravarora2 <130506156+ravarora2@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:06:33 -0400 Subject: [PATCH] Add authenticated WebSocket recovery telemetry (#7546) ## Summary Argo rollout recovery needs to distinguish an open WebSocket from a connection that completed NIP-42 authentication and is usable. The relay currently exposes total WebSocket connections, but it cannot answer whether authenticated clients recovered after a pod drain or why authentication failed. This change adds a bounded recovery contract: - `buzz_auth_attempts_total{method="nip42"}` - `buzz_auth_outcomes_total{method="nip42",outcome}` - `buzz_auth_duration_seconds{method="nip42",outcome}` - `buzz_ws_authenticated_connections_active` - `buzz_auth_post_terminal_frames_total{state}` `buzz_auth_attempts_total` now has one consistent recovery-oriented unit: a challenge lifecycle successfully queued to the connection writer. This intentionally replaces the historical unit of AUTH frames that reached the pending handler. AUTH frames received after a lifecycle is already authenticated or failed are protocol noise; they use the separate bounded post-terminal counter and cannot inflate rollout-gating attempts or outcomes. Each issued challenge reaches exactly one terminal outcome. Authentication transitions and active-gauge changes share one short synchronous state lock, cancellation can claim a pending database-backed AUTH before a late handler result, and a drop guard reconciles aborted or panicked connection futures. Accounting is terminalized before writer joins, while terminal socket delivery has a one-second best-effort bound so a stalled sink cannot retain the authenticated gauge or connection permit indefinitely. Dependency failures are distinct from policy verdicts: allowlist and relay-membership lookup errors now emit `allowlist_check_error` and `relay_membership_check_error`, never `allowlist_denied` or `not_relay_member`. The fail-closed allow/deny policy is unchanged. All rollout-facing labels come from fixed enums; identities, challenges, URLs, and raw error strings are never exported. All fixed counter and histogram label sets, including zero histogram buckets, counts, and sums, are present on the first real exporter scrape. This is the first implementation slice of the rollout-reliability plan. A later PR will consume these metrics from opt-in Argo Rollout analysis; merging this PR alone does not change deployment behavior. ### Related issue None found. ### Testing Validated at final head `f9992cac835284c25ee0f99e17ddba0b932d7c69`, which contains current `origin/main`: - `cargo fmt --all -- --check` - `cargo clippy -p buzz-relay --all-targets -- -D warnings` - compiled every `buzz-relay` test target - release builds for `buzz-relay`, `buzz`, `buzz-admin`, and `buzz-test-cli` - 16/16 connection lifecycle/writer/production-dispatch tests - 7/7 AUTH handler tests - bounded metric-contract exporter test - 13/13 relay main-target tests (1 PostgreSQL-only test ignored) - 9/9 real boot-lifecycle tests, including a production install and first Prometheus scrape Mutation checks proved that the production regressions fail when any of these are removed or reintroduced: - challenge attempt start in `handle_active_connection` - authentication series initialization in the real metric installer - separation of post-terminal AUTH floods from authoritative attempts The exact-head release binaries were also exercised against the isolated review database: 1. Started the release relay and verified health/readiness. 2. Created a channel with the regular CLI. 3. Held one authenticated WebSocket subscription open and authenticated a second WebSocket publisher. 4. Published a message and read it back through the regular CLI. 5. Observed `attempt_delta=2`, `success_delta=2`, `active_while_subscribed=1`, and `active_after_disconnect=0`. 6. Sent SIGTERM and waited for graceful relay shutdown. ### Local caveat The full relay library target consistently reached 1,049 passes with 89 intentionally ignored tests, but the untouched `api::mesh_demo::tests::demo_join_forwarded_arm_round_trips_echo` intermittently returned HTTP 504 instead of 200. One isolated rerun passed; later isolated and full reruns reproduced the 504. No mesh files are changed by this PR, and all changed-path, main-target, boot-lifecycle, lint, build, mutation, and live-local checks above passed at the final head. Generated with Codex --------- Signed-off-by: Ravneet Arora --- crates/buzz-relay/src/connection.rs | 941 +++++++++++++++++++++- crates/buzz-relay/src/handlers/auth.rs | 420 ++++++++-- crates/buzz-relay/src/handlers/count.rs | 3 +- crates/buzz-relay/src/handlers/event.rs | 10 +- crates/buzz-relay/src/handlers/req.rs | 3 +- crates/buzz-relay/src/main.rs | 8 + crates/buzz-relay/src/metrics.rs | 280 +++++++ crates/buzz-relay/src/rejection.rs | 3 +- crates/buzz-relay/src/state.rs | 35 +- crates/buzz-relay/tests/boot_lifecycle.rs | 45 ++ 10 files changed, 1636 insertions(+), 112 deletions(-) diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index e284e7fa6a..12a450861e 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -3,12 +3,12 @@ use std::collections::HashMap; use std::net::SocketAddr; use std::sync::atomic::{AtomicU8, Ordering}; -use std::sync::Arc; -use std::time::Duration; +use std::sync::{Arc, Mutex as StdMutex, PoisonError}; +use std::time::{Duration, Instant}; use axum::extract::ws::{Message as WsMessage, WebSocket}; use futures_util::{Sink, SinkExt, StreamExt}; -use tokio::sync::{mpsc, watch, Mutex, RwLock}; +use tokio::sync::{mpsc, watch, Mutex}; use tokio_util::sync::CancellationToken; use tracing::Instrument as _; use tracing::{debug, info, trace, warn}; @@ -19,6 +19,7 @@ use buzz_core::tenant::TenantContext; use nostr::Filter; use crate::handlers; +use crate::metrics::AuthOutcome; use crate::protocol::{ClientMessage, RelayMessage}; use crate::rejection::{enforce_ws_admission, request_rejection_message, RejectionTarget}; use crate::state::{ @@ -29,6 +30,10 @@ use crate::state::{ /// Maximum time a new socket may hold a connection slot without completing NIP-42 auth. const AUTH_TIMEOUT: Duration = Duration::from_secs(5); +/// Maximum time the writer may spend flushing terminal frames after cancellation. +/// This stays well inside the process-wide 30-second hard drain. +const WS_TERMINAL_FLUSH_TIMEOUT: Duration = Duration::from_secs(1); + /// Shared mutable subscription map for a single WebSocket connection. pub(crate) type ConnectionSubscriptions = Arc>>>; @@ -47,6 +52,8 @@ pub enum AuthState { Pending { /// The random challenge string sent to the client. challenge: String, + /// When the challenge was delivered and this attempt began. + started_at: Instant, }, /// Client has successfully authenticated. Authenticated(AuthContext), @@ -55,7 +62,7 @@ pub enum AuthState { } /// Per-connection state split by access pattern: -/// - `auth_state`: RwLock (read-heavy after initial auth) +/// - `auth_state`: synchronous mutex (short, non-awaiting transitions; drop-safe cleanup) /// - `subscriptions`: Mutex (write-heavy during REQ/CLOSE) /// - `send_tx`, `ctrl_tx`, `cancel`: outside any lock (Clone+Send, no coordination needed) pub struct ConnectionState { @@ -68,7 +75,7 @@ pub struct ConnectionState { /// Remote socket address of the client. pub remote_addr: SocketAddr, /// Current NIP-42 authentication state. - pub auth_state: RwLock, + pub auth_state: StdMutex, /// Active subscriptions keyed by subscription ID. pub subscriptions: ConnectionSubscriptions, /// Sender for outbound data messages (EVENT, NOTICE, OK, etc.). @@ -88,6 +95,104 @@ pub struct ConnectionState { } impl ConnectionState { + fn lock_auth_state(&self) -> std::sync::MutexGuard<'_, AuthState> { + self.auth_state + .lock() + .unwrap_or_else(PoisonError::into_inner) + } + + /// Snapshot the current authentication state without holding its lock over an await. + pub(crate) fn auth_state_snapshot(&self) -> AuthState { + self.lock_auth_state().clone() + } + + fn transition_pending_auth(&self, next: AuthState, outcome: AuthOutcome) -> bool { + let mut auth = self.lock_auth_state(); + let AuthState::Pending { started_at, .. } = &*auth else { + return false; + }; + let duration = started_at.elapsed(); + let became_authenticated = matches!(next, AuthState::Authenticated(_)); + *auth = next; + crate::metrics::record_auth_outcome(outcome, duration); + if became_authenticated { + // Keep the gauge update under the same state lock as the + // Pending -> Authenticated transition. Cleanup must never observe + // Authenticated before its increment and decrement first. + metrics::gauge!("buzz_ws_authenticated_connections_active").increment(1.0); + } + true + } + + /// Atomically finish the initial challenge as authenticated. + pub(crate) fn authenticate(&self, auth_context: AuthContext) -> bool { + self.transition_pending_auth(AuthState::Authenticated(auth_context), AuthOutcome::Success) + } + + /// Atomically finish the initial challenge with a bounded denial. + pub(crate) fn reject_auth(&self, outcome: AuthOutcome) -> bool { + debug_assert!(!matches!(outcome, AuthOutcome::Success)); + self.transition_pending_auth(AuthState::Failed, outcome) + } + + /// Finish a pending challenge on timeout and preserve the historical rule + /// that an already-failed connection is closed when its timeout expires. + fn expire_auth(&self) -> bool { + let mut auth = self.lock_auth_state(); + match &*auth { + AuthState::Pending { started_at, .. } => { + let duration = started_at.elapsed(); + *auth = AuthState::Failed; + crate::metrics::record_auth_outcome(AuthOutcome::Timeout, duration); + true + } + AuthState::Failed => true, + AuthState::Authenticated(_) => false, + } + } + + /// Finalize authentication accounting when a connection closes. + /// + /// Replacing the state with `Failed` makes cleanup idempotent: an + /// authenticated gauge can be decremented at most once, and a pending + /// attempt can receive at most one disconnect/shutdown terminal. + fn finish_auth_on_close(&self, outcome: AuthOutcome) -> Option { + debug_assert!(matches!( + outcome, + AuthOutcome::Disconnect | AuthOutcome::Shutdown + )); + let mut auth = self.lock_auth_state(); + match std::mem::replace(&mut *auth, AuthState::Failed) { + AuthState::Pending { started_at, .. } => { + crate::metrics::record_auth_outcome(outcome, started_at.elapsed()); + None + } + AuthState::Authenticated(auth_context) => { + metrics::gauge!("buzz_ws_authenticated_connections_active").decrement(1.0); + Some(auth_context) + } + AuthState::Failed => None, + } + } + + /// Let the cancellation watcher claim only a still-pending attempt. + /// Authenticated cleanup remains owned by `AuthLifecycleGuard`, which must + /// retain the authentication context for presence cleanup after task joins. + fn finish_pending_auth_on_cancel(&self, outcome: AuthOutcome) -> bool { + debug_assert!(matches!( + outcome, + AuthOutcome::Disconnect | AuthOutcome::Shutdown + )); + let mut auth = self.lock_auth_state(); + let AuthState::Pending { started_at, .. } = &*auth else { + return false; + }; + let duration = started_at.elapsed(); + *auth = AuthState::Failed; + crate::metrics::record_auth_outcome(outcome, duration); + true + } + /// Sends a data message to this connection's outbound channel. /// /// On a full buffer, increments the backpressure counter. The first @@ -119,6 +224,37 @@ impl ConnectionState { } } +/// Owns authentication accounting for exactly the lifetime of the production +/// connection future. Explicit teardown selects the precise close outcome; +/// aborts and panics fall back to `disconnect` and cancel child tasks. +struct AuthLifecycleGuard { + conn: Arc, + finished: bool, +} + +impl AuthLifecycleGuard { + fn new(conn: Arc) -> Self { + Self { + conn, + finished: false, + } + } + + fn finish(&mut self, outcome: AuthOutcome) -> Option { + self.finished = true; + self.conn.finish_auth_on_close(outcome) + } +} + +impl Drop for AuthLifecycleGuard { + fn drop(&mut self) { + if !self.finished { + self.conn.cancel.cancel(); + self.conn.finish_auth_on_close(AuthOutcome::Disconnect); + } + } +} + /// Entry point for a new WebSocket connection. /// /// Acquires a connection semaphore permit, sends the NIP-42 AUTH challenge, @@ -184,8 +320,9 @@ async fn handle_active_connection( conn_id, tenant, remote_addr: addr, - auth_state: RwLock::new(AuthState::Pending { + auth_state: StdMutex::new(AuthState::Pending { challenge: challenge.clone(), + started_at: Instant::now(), }), subscriptions: Arc::clone(&subscriptions), send_tx: tx.clone(), @@ -215,6 +352,8 @@ async fn handle_active_connection( // Gauge incremented AFTER challenge send succeeds — early disconnects // don't leak. Decremented in the cleanup path below. metrics::gauge!("buzz_ws_connections_active").increment(1.0); + crate::metrics::record_auth_attempt_started(); + let mut auth_lifecycle = AuthLifecycleGuard::new(Arc::clone(&conn)); // Register after challenge succeeds — avoids leaked entries on early disconnect. state.conn_manager.register( @@ -254,11 +393,7 @@ async fn handle_active_connection( let auth_timeout_task = tokio::spawn(async move { tokio::select! { _ = tokio::time::sleep(AUTH_TIMEOUT) => { - let authenticated = matches!( - *auth_timeout_conn.auth_state.read().await, - AuthState::Authenticated(_) - ); - if !authenticated { + if auth_timeout_conn.expire_auth() { warn!( conn_id = %auth_timeout_conn.conn_id, timeout_secs = AUTH_TIMEOUT.as_secs(), @@ -272,6 +407,22 @@ async fn handle_active_connection( } }); + // Cancellation races database-backed AUTH work. This watcher claims the + // pending lifecycle under the same lock as success/denial transitions, so + // whichever terminal happens first wins and a late handler cannot overwrite it. + let auth_cancel_conn = Arc::clone(&conn); + let auth_cancel_state = Arc::clone(&state); + let auth_cancel_token = cancel.clone(); + let auth_cancel_task = tokio::spawn(async move { + auth_cancel_token.cancelled().await; + let outcome = if auth_cancel_state.shutting_down.load(Ordering::Acquire) { + AuthOutcome::Shutdown + } else { + AuthOutcome::Disconnect + }; + auth_cancel_conn.finish_pending_auth_on_cancel(outcome); + }); + recv_loop( ws_recv, Arc::clone(&conn), @@ -282,9 +433,19 @@ async fn handle_active_connection( .await; cancel.cancel(); + let close_outcome = if state.shutting_down.load(Ordering::Acquire) { + AuthOutcome::Shutdown + } else { + AuthOutcome::Disconnect + }; + // Terminalize before joining writer/heartbeat tasks. A blocked socket sink + // must not keep the authenticated gauge high during shutdown. + let authenticated = auth_lifecycle.finish(close_outcome); + let _ = send_task.await; let _ = heartbeat_task.await; let _ = auth_timeout_task.await; + let _ = auth_cancel_task.await; for removed in state.sub_registry.remove_connection(conn.conn_id) { if removed.scope.is_global() { @@ -301,7 +462,7 @@ async fn handle_active_connection( } } state.conn_manager.deregister(conn.conn_id); - if let AuthState::Authenticated(ref auth_ctx) = *conn.auth_state.read().await { + if let Some(auth_ctx) = authenticated { let remaining = state.conn_manager.connection_ids_for_pubkey_in_community( conn.tenant.community(), auth_ctx.pubkey.to_bytes().as_slice(), @@ -344,6 +505,93 @@ async fn send_loop( .await; } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum WriterStep { + Completed, + Cancelled, + Failed, +} + +async fn send_or_cancel( + sink: &mut S, + message: WsMessage, + cancel: &CancellationToken, +) -> WriterStep +where + S: Sink + Unpin, +{ + tokio::select! { + biased; + _ = cancel.cancelled() => WriterStep::Cancelled, + result = sink.send(message) => { + if result.is_ok() { WriterStep::Completed } else { WriterStep::Failed } + } + } +} + +async fn feed_or_cancel( + sink: &mut S, + message: WsMessage, + cancel: &CancellationToken, +) -> WriterStep +where + S: Sink + Unpin, +{ + tokio::select! { + biased; + _ = cancel.cancelled() => WriterStep::Cancelled, + result = sink.feed(message) => { + if result.is_ok() { WriterStep::Completed } else { WriterStep::Failed } + } + } +} + +async fn flush_or_cancel(sink: &mut S, cancel: &CancellationToken) -> WriterStep +where + S: Sink + Unpin, +{ + tokio::select! { + biased; + _ = cancel.cancelled() => WriterStep::Cancelled, + result = sink.flush() => { + if result.is_ok() { WriterStep::Completed } else { WriterStep::Failed } + } + } +} + +/// Best-effort terminal delivery with one shared deadline. A socket that never +/// becomes writable cannot retain its connection task or semaphore permit. +async fn flush_terminal_frames( + sink: &mut S, + ctrl_rx: &mut mpsc::Receiver, + disconnect_reason: &watch::Receiver>, + first_ctrl: Option, +) where + S: Sink + Unpin, +{ + let deadline = tokio::time::Instant::now() + WS_TERMINAL_FLUSH_TIMEOUT; + if let Some(ctrl_msg) = first_ctrl { + if !matches!( + tokio::time::timeout_at(deadline, sink.send(ctrl_msg)).await, + Ok(Ok(())) + ) { + return; + } + } + while let Ok(ctrl_msg) = ctrl_rx.try_recv() { + if !matches!( + tokio::time::timeout_at(deadline, sink.send(ctrl_msg)).await, + Ok(Ok(())) + ) { + return; + } + } + let close = disconnect_reason + .borrow() + .map_or(WsMessage::Close(None), |reason| reason.close_message()); + let _ = tokio::time::timeout_at(deadline, sink.send(close)).await; +} + async fn send_loop_inner( mut ws_send: S, mut data_rx: mpsc::Receiver, @@ -357,8 +605,19 @@ async fn send_loop_inner( loop { // Priority: drain all pending control frames before data. while let Ok(ctrl_msg) = ctrl_rx.try_recv() { - if ws_send.send(ctrl_msg).await.is_err() { - return; + match send_or_cancel(&mut ws_send, ctrl_msg.clone(), &cancel).await { + WriterStep::Completed => {} + WriterStep::Cancelled => { + flush_terminal_frames( + &mut ws_send, + &mut ctrl_rx, + &disconnect_reason, + Some(ctrl_msg), + ) + .await; + return; + } + WriterStep::Failed => return, } } @@ -368,13 +627,16 @@ async fn send_loop_inner( // cancellation can fall back to an unacknowledged close. biased; Some(restart) = restart_rx.recv() => { - let sent = ws_send - .send(WsMessage::Close(Some(axum::extract::ws::CloseFrame { + let sent = matches!( + tokio::time::timeout( + WS_TERMINAL_FLUSH_TIMEOUT, + ws_send.send(WsMessage::Close(Some(axum::extract::ws::CloseFrame { code: axum::extract::ws::close_code::RESTART, reason: axum::extract::ws::Utf8Bytes::from_static("relay restarting"), - }))) - .await - .is_ok(); + }))), + ).await, + Ok(Ok(())) + ); let _ = restart.flushed.send(sent); break; } @@ -385,33 +647,58 @@ async fn send_loop_inner( // would send Close first and the client would never learn why // (the top-of-loop drain does not run again after we break). // This makes "queue frame on ctrl, then cancel" a safe idiom. - while let Ok(ctrl_msg) = ctrl_rx.try_recv() { - if ws_send.send(ctrl_msg).await.is_err() { - break; - } - } - let close = disconnect_reason - .borrow() - .map_or(WsMessage::Close(None), |reason| reason.close_message()); - let _ = ws_send.send(close).await; + flush_terminal_frames(&mut ws_send, &mut ctrl_rx, &disconnect_reason, None).await; break; } Some(ctrl_msg) = ctrl_rx.recv() => { - if ws_send.send(ctrl_msg).await.is_err() { - break; + match send_or_cancel(&mut ws_send, ctrl_msg.clone(), &cancel).await { + WriterStep::Completed => {} + WriterStep::Cancelled => { + flush_terminal_frames( + &mut ws_send, + &mut ctrl_rx, + &disconnect_reason, + Some(ctrl_msg), + ) + .await; + break; + } + WriterStep::Failed => break, } } Some(msg) = data_rx.recv() => { let mut batched = 1usize; - if ws_send.feed(msg).await.is_err() { - break; + match feed_or_cancel(&mut ws_send, msg, &cancel).await { + WriterStep::Completed => {} + WriterStep::Cancelled => { + flush_terminal_frames( + &mut ws_send, + &mut ctrl_rx, + &disconnect_reason, + None, + ) + .await; + break; + } + WriterStep::Failed => break, } while batched < MAX_WS_SEND_BATCH { match data_rx.try_recv() { Ok(next) => { - if ws_send.feed(next).await.is_err() { - return; + match feed_or_cancel(&mut ws_send, next, &cancel).await { + WriterStep::Completed => {} + WriterStep::Cancelled => { + flush_terminal_frames( + &mut ws_send, + &mut ctrl_rx, + &disconnect_reason, + None, + ) + .await; + return; + } + WriterStep::Failed => return, } batched += 1; } @@ -420,8 +707,19 @@ async fn send_loop_inner( } } - if ws_send.flush().await.is_err() { - break; + match flush_or_cancel(&mut ws_send, &cancel).await { + WriterStep::Completed => {} + WriterStep::Cancelled => { + flush_terminal_frames( + &mut ws_send, + &mut ctrl_rx, + &disconnect_reason, + None, + ) + .await; + break; + } + WriterStep::Failed => break, } metrics::histogram!("buzz_ws_send_batch_size").record(batched as f64); } @@ -560,11 +858,15 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar match msg { ClientMessage::Auth(event) => { - // Auth is synchronous in the WS loop — no span context is lost. + // AUTH remains inline so only one frame can race the connection's + // pending lifecycle, but cancellation can preempt dependency waits. let span = tracing::info_span!("ws.auth", conn_id = %conn.conn_id); - handlers::auth::handle_auth(event, Arc::clone(&conn), Arc::clone(&state)) - .instrument(span) - .await; + tokio::select! { + biased; + _ = conn.cancel.cancelled() => {} + _ = handlers::auth::handle_auth(event, Arc::clone(&conn), Arc::clone(&state)) + .instrument(span) => {} + } } ClientMessage::Event(event) => { let conn = Arc::clone(&conn); @@ -654,10 +956,15 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar #[cfg(test)] pub(crate) mod tests { use super::*; + use axum::{extract::ws::WebSocketUpgrade, routing::get, Router}; + use metrics_util::debugging::DebugValue; use std::sync::{Arc, Mutex}; use buzz_auth::AuthMethod; - use nostr::{EventBuilder, Keys, Kind}; + use nostr::{EventBuilder, Keys, Kind, RelayUrl}; + use tokio::net::TcpListener; + use tokio::sync::Notify; + use tokio_tungstenite::{connect_async, tungstenite::Message}; /// A connection whose outbound frames a test can read back. /// @@ -675,7 +982,7 @@ pub(crate) mod tests { "test.local".to_string(), ), remote_addr: "127.0.0.1:1234".parse().expect("socket addr"), - auth_state: RwLock::new(auth), + auth_state: StdMutex::new(auth), subscriptions: Arc::new(tokio::sync::Mutex::new(HashMap::new())), send_tx, ctrl_tx, @@ -688,13 +995,109 @@ pub(crate) mod tests { /// An authenticated connection — the only state admission quotas apply to. pub(crate) fn authenticated_state() -> AuthState { - AuthState::Authenticated(AuthContext { + AuthState::Authenticated(auth_context()) + } + + fn auth_context() -> AuthContext { + AuthContext { pubkey: Keys::generate().public_key(), scopes: Vec::new(), channel_ids: None, auth_method: AuthMethod::Nip42, agent_owner_pubkey: None, - }) + } + } + + fn pending_state() -> AuthState { + AuthState::Pending { + challenge: "test-challenge".to_owned(), + started_at: Instant::now(), + } + } + + type MetricSnapshot = Vec<( + metrics_util::CompositeKey, + Option, + Option, + DebugValue, + )>; + + fn counter_value(snapshot: &MetricSnapshot, name: &str, outcome: Option<&str>) -> u64 { + snapshot + .iter() + .find_map(|(key, _, _, value)| { + if key.key().name() != name { + return None; + } + let labels = key.key().labels().collect::>(); + if outcome.is_some_and(|expected| { + !labels + .iter() + .any(|label| label.key() == "outcome" && label.value() == expected) + }) { + return None; + } + let DebugValue::Counter(value) = value else { + panic!("{name} must be a counter"); + }; + Some(*value) + }) + .unwrap_or_default() + } + + fn labeled_counter_value( + snapshot: &MetricSnapshot, + name: &str, + label_key: &str, + label_value: &str, + ) -> u64 { + snapshot + .iter() + .find_map(|(key, _, _, value)| { + (key.key().name() == name + && key + .key() + .labels() + .any(|label| label.key() == label_key && label.value() == label_value)) + .then(|| match value { + DebugValue::Counter(value) => *value, + _ => panic!("{name} must be a counter"), + }) + }) + .unwrap_or_default() + } + + fn labeled_gauge_value(snapshot: &MetricSnapshot, name: &str, labels: &[(&str, &str)]) -> f64 { + snapshot + .iter() + .find_map(|(key, _, _, value)| { + let matches = key.key().name() == name + && labels.iter().all(|(expected_key, expected_value)| { + key.key().labels().any(|label| { + label.key() == *expected_key && label.value() == *expected_value + }) + }); + matches.then(|| match value { + DebugValue::Gauge(value) => value.into_inner(), + _ => panic!("{name} must be a gauge"), + }) + }) + .unwrap_or_default() + } + + fn authenticated_gauge(snapshot: &MetricSnapshot) -> f64 { + snapshot + .iter() + .find_map(|(key, _, _, value)| { + if key.key().name() != "buzz_ws_authenticated_connections_active" { + return None; + } + let DebugValue::Gauge(value) = value else { + panic!("authenticated connections must be a gauge"); + }; + Some(value.into_inner()) + }) + .unwrap_or_default() } pub(crate) fn read_frame(rx: &mut mpsc::Receiver) -> serde_json::Value { @@ -704,6 +1107,387 @@ pub(crate) mod tests { } } + /// Exercise the real state-transition methods for every terminal. The + /// attempt counter must reconcile with exactly one terminal per completed + /// attempt, and repeated/racing terminal calls must not drive the active + /// authenticated gauge below zero. + #[tokio::test(flavor = "current_thread")] + async fn auth_lifecycle_reconciles_every_terminal_and_never_leaks_gauge() { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + + crate::metrics::record_auth_attempt_started(); + let (success, _rx) = test_conn_with_auth(pending_state()); + assert!(success.authenticate(auth_context())); + assert!( + !success.authenticate(auth_context()), + "a terminal attempt cannot authenticate twice" + ); + assert!( + !success.finish_pending_auth_on_cancel(AuthOutcome::Disconnect), + "the cancellation watcher must leave authenticated context for lifecycle cleanup" + ); + assert!(success + .finish_auth_on_close(AuthOutcome::Disconnect) + .is_some()); + assert!( + success + .finish_auth_on_close(AuthOutcome::Shutdown) + .is_none(), + "cleanup must be idempotent" + ); + + for outcome in [ + AuthOutcome::Invalid, + AuthOutcome::Banned, + AuthOutcome::BanCheckError, + AuthOutcome::AllowlistCheckError, + AuthOutcome::AllowlistDenied, + AuthOutcome::RelayMembershipCheckError, + AuthOutcome::NotRelayMember, + ] { + crate::metrics::record_auth_attempt_started(); + let (denied, _rx) = test_conn_with_auth(pending_state()); + assert!(denied.reject_auth(outcome)); + assert!(!denied.reject_auth(outcome), "a denial cannot record twice"); + assert!(denied + .finish_auth_on_close(AuthOutcome::Disconnect) + .is_none()); + } + + crate::metrics::record_auth_attempt_started(); + let (timed_out, _rx) = test_conn_with_auth(pending_state()); + assert!(timed_out.expire_auth()); + assert!(timed_out + .finish_auth_on_close(AuthOutcome::Disconnect) + .is_none()); + + for outcome in [AuthOutcome::Disconnect, AuthOutcome::Shutdown] { + crate::metrics::record_auth_attempt_started(); + let (closed, _rx) = test_conn_with_auth(pending_state()); + assert!(closed.finish_auth_on_close(outcome).is_none()); + assert!(closed.finish_auth_on_close(outcome).is_none()); + } + + let snapshot = snapshotter.snapshot().into_vec(); + let attempts = counter_value(&snapshot, "buzz_auth_attempts_total", None); + let outcomes = AuthOutcome::ALL + .iter() + .map(|outcome| { + let value = counter_value( + &snapshot, + "buzz_auth_outcomes_total", + Some(outcome.as_str()), + ); + assert_eq!( + value, + 1, + "{} terminal must be recorded exactly once", + outcome.as_str() + ); + value + }) + .sum::(); + + assert_eq!(attempts, AuthOutcome::ALL.len() as u64); + assert_eq!(attempts, outcomes); + assert_eq!(authenticated_gauge(&snapshot), 0.0); + } + + /// Post-terminal AUTH floods traverse the production frame dispatcher but + /// cannot mint rollout-gating challenge lifecycles or outcomes. + #[tokio::test(flavor = "current_thread")] + async fn auth_flood_after_terminal_only_increments_protocol_noise_metric() { + const FRAMES_PER_STATE: u64 = 64; + + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + let state = crate::state::tests::test_state().await; + let mut authoritative_attempts = 0; + let mut authoritative_outcomes = 0; + + for (auth_state, expected_state) in [ + ( + authenticated_state(), + crate::metrics::AuthPostTerminalState::Authenticated, + ), + ( + AuthState::Failed, + crate::metrics::AuthPostTerminalState::Failed, + ), + ] { + let (conn, mut rx) = test_conn_with_auth(auth_state); + for sequence in 0..FRAMES_PER_STATE { + let event = EventBuilder::new(Kind::Authentication, format!("noise-{sequence}")) + .sign_with_keys(&Keys::generate()) + .expect("sign AUTH noise event"); + let raw = serde_json::json!(["AUTH", event]).to_string(); + handle_text_message(raw, Arc::clone(&conn), Arc::clone(&state)).await; + let frame = read_frame(&mut rx); + assert_eq!(frame[2], false); + } + assert!(!conn.cancel.is_cancelled()); + let snapshot = snapshotter.snapshot().into_vec(); + authoritative_attempts += counter_value(&snapshot, "buzz_auth_attempts_total", None); + authoritative_outcomes += counter_value(&snapshot, "buzz_auth_outcomes_total", None); + assert_eq!( + labeled_counter_value( + &snapshot, + "buzz_auth_post_terminal_frames_total", + "state", + expected_state.as_str(), + ), + FRAMES_PER_STATE + ); + } + + let snapshot = snapshotter.snapshot().into_vec(); + authoritative_attempts += counter_value(&snapshot, "buzz_auth_attempts_total", None); + authoritative_outcomes += counter_value(&snapshot, "buzz_auth_outcomes_total", None); + assert_eq!( + authoritative_attempts, 0, + "post-terminal protocol noise cannot create authoritative attempts" + ); + assert_eq!( + authoritative_outcomes, 0, + "post-terminal protocol noise cannot create authoritative outcomes" + ); + } + + /// Aborting the production lifecycle owner must synchronously terminalize + /// pending accounting and release an authenticated gauge exactly once. + #[tokio::test(flavor = "current_thread")] + async fn aborting_lifecycle_owner_is_drop_safe() { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + + crate::metrics::record_auth_attempt_started(); + let (authenticated, _rx) = test_conn_with_auth(pending_state()); + assert!(authenticated.authenticate(auth_context())); + let authenticated_started = Arc::new(Notify::new()); + let task = { + let conn = Arc::clone(&authenticated); + let started = Arc::clone(&authenticated_started); + tokio::spawn(async move { + let _guard = AuthLifecycleGuard::new(conn); + started.notify_one(); + std::future::pending::<()>().await; + }) + }; + authenticated_started.notified().await; + task.abort(); + assert!(task.await.expect_err("task was aborted").is_cancelled()); + + crate::metrics::record_auth_attempt_started(); + let (pending, _rx) = test_conn_with_auth(pending_state()); + let pending_started = Arc::new(Notify::new()); + let task = { + let conn = Arc::clone(&pending); + let started = Arc::clone(&pending_started); + tokio::spawn(async move { + let _guard = AuthLifecycleGuard::new(conn); + started.notify_one(); + std::future::pending::<()>().await; + }) + }; + pending_started.notified().await; + task.abort(); + assert!(task.await.expect_err("task was aborted").is_cancelled()); + + let snapshot = snapshotter.snapshot().into_vec(); + assert_eq!( + counter_value(&snapshot, "buzz_auth_attempts_total", None), + 2 + ); + assert_eq!( + counter_value( + &snapshot, + "buzz_auth_outcomes_total", + Some(AuthOutcome::Success.as_str()), + ), + 1 + ); + assert_eq!( + counter_value( + &snapshot, + "buzz_auth_outcomes_total", + Some(AuthOutcome::Disconnect.as_str()), + ), + 1 + ); + assert_eq!(authenticated_gauge(&snapshot), 0.0); + assert!(matches!(pending.auth_state_snapshot(), AuthState::Failed)); + } + + /// Drive the real upgraded WebSocket lifecycle until AUTH is waiting for + /// the sole database connection. Production shutdown must claim the + /// pending attempt before that dependency is released, and the late DB + /// result must not turn the terminal into success. + #[tokio::test(flavor = "current_thread")] + async fn shutdown_terminalizes_auth_stalled_on_database_before_late_success() { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + + // Occupy the pool's only connection attempt with a fake PostgreSQL + // endpoint that accepts TCP and never completes the startup handshake. + // The production AUTH query then waits for the pool permit without + // requiring a developer database or relying on timing alone. + let fake_database = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind fake PostgreSQL endpoint"); + let database_url = format!( + "postgres://buzz@{}/buzz", + fake_database.local_addr().expect("fake database address") + ); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_secs(30)) + .connect_lazy(&database_url) + .expect("create lifecycle test pool"); + let blocker_pool = pool.clone(); + let blocker = tokio::spawn(async move { blocker_pool.acquire().await }); + let (blocked_database_stream, _) = fake_database + .accept() + .await + .expect("pool reached fake PostgreSQL endpoint"); + let state = crate::state::tests::test_state_with_database_pool(pool.clone()).await; + let tenant = TenantContext::resolved( + buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()), + "test.local".to_owned(), + ); + let expected_relay_url: RelayUrl = + crate::api::bridge::nip42_expected_relay_url(&state.config.relay_url, &tenant) + .parse() + .expect("expected NIP-42 relay URL"); + + let connection_finished = Arc::new(Notify::new()); + let route_state = Arc::clone(&state); + let route_tenant = tenant.clone(); + let route_finished = Arc::clone(&connection_finished); + let app = Router::new().route( + "/", + get(move |ws: WebSocketUpgrade| { + let state = Arc::clone(&route_state); + let tenant = route_tenant.clone(); + let finished = Arc::clone(&route_finished); + async move { + ws.on_upgrade(move |socket| async move { + let cancel = CancellationToken::new(); + let control = CommunityConnectionControl::new(cancel); + handle_active_connection( + socket, + state, + "127.0.0.1:1234".parse().expect("client address"), + tenant, + Uuid::new_v4(), + control, + ) + .await; + finished.notify_one(); + }) + } + }), + ); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind lifecycle WebSocket listener"); + let address = listener.local_addr().expect("listener address"); + let server = tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("serve lifecycle WebSocket"); + }); + + let (mut client, _) = connect_async(format!("ws://{address}/")) + .await + .expect("connect lifecycle WebSocket"); + let challenge_frame = client + .next() + .await + .expect("challenge frame") + .expect("read challenge"); + let Message::Text(challenge_text) = challenge_frame else { + panic!("expected text challenge") + }; + let challenge_json: serde_json::Value = + serde_json::from_str(&challenge_text).expect("parse challenge"); + assert_eq!(challenge_json[0], "AUTH"); + let challenge = challenge_json[1].as_str().expect("challenge string"); + let auth_event = EventBuilder::auth(challenge, expected_relay_url) + .sign_with_keys(&Keys::generate()) + .expect("sign NIP-42 AUTH"); + client + .send(Message::Text( + serde_json::json!(["AUTH", auth_event]).to_string().into(), + )) + .await + .expect("send NIP-42 AUTH"); + + let attempts_before_shutdown = tokio::time::timeout(Duration::from_secs(2), async { + let mut attempts = 0; + loop { + let snapshot = snapshotter.snapshot().into_vec(); + attempts += counter_value(&snapshot, "buzz_auth_attempts_total", None); + if labeled_gauge_value( + &snapshot, + "buzz_db_pool_waiters", + &[("pool_role", "writer"), ("operation", "authorization")], + ) >= 1.0 + { + break attempts; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("AUTH reached the blocked production DB acquisition"); + assert_eq!( + attempts_before_shutdown, 1, + "the production-issued challenge must own the attempt start" + ); + + state.begin_shutdown(); + assert_eq!(state.conn_manager.drain_all(), 1); + tokio::time::timeout(Duration::from_secs(2), connection_finished.notified()) + .await + .expect("production connection lifecycle finishes after shutdown"); + + // Release the dependency only after production shutdown has claimed + // the pending lifecycle, then prove no late handler result can win. + drop(blocked_database_stream); + blocker.abort(); + let _ = blocker.await; + pool.close().await; + + let snapshot = snapshotter.snapshot().into_vec(); + assert_eq!( + counter_value( + &snapshot, + "buzz_auth_outcomes_total", + Some(AuthOutcome::Shutdown.as_str()), + ), + 1 + ); + assert_eq!( + counter_value( + &snapshot, + "buzz_auth_outcomes_total", + Some(AuthOutcome::Success.as_str()), + ), + 0, + "releasing the dependency after shutdown must not record late success" + ); + assert_eq!(authenticated_gauge(&snapshot), 0.0); + + drop(client); + server.abort(); + let _ = server.await; + } + /// 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. @@ -857,6 +1641,41 @@ pub(crate) mod tests { } } + #[derive(Debug)] + struct NeverReadySink { + ready_polled: Arc, + } + + impl Sink for NeverReadySink { + type Error = std::io::Error; + + fn poll_ready( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.ready_polled.notify_one(); + std::task::Poll::Pending + } + + fn start_send(self: std::pin::Pin<&mut Self>, _item: WsMessage) -> Result<(), Self::Error> { + panic!("a never-ready sink must not accept a frame") + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Pending + } + + fn poll_close( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Pending + } + } + fn ordinary_disconnect_reason() -> watch::Receiver> { let (_tx, rx) = watch::channel(None); rx @@ -878,6 +1697,40 @@ pub(crate) mod tests { .collect() } + /// Cancellation must break a writer blocked in `poll_ready`, and terminal + /// close delivery gets one bounded best-effort window before teardown wins. + #[tokio::test(start_paused = true)] + async fn cancelled_never_ready_sink_cannot_retain_writer_task() { + let (data_tx, data_rx) = mpsc::channel(1); + let (_ctrl_tx, ctrl_rx) = mpsc::channel(1); + let (_restart_tx, restart_rx) = mpsc::channel(1); + let cancel = CancellationToken::new(); + let ready_polled = Arc::new(Notify::new()); + data_tx + .send(WsMessage::Text("blocked".into())) + .await + .expect("queue blocked frame"); + + let writer = tokio::spawn(send_loop_inner( + NeverReadySink { + ready_polled: Arc::clone(&ready_polled), + }, + data_rx, + ctrl_rx, + restart_rx, + cancel.clone(), + ordinary_disconnect_reason(), + )); + + ready_polled.notified().await; + cancel.cancel(); + tokio::task::yield_now().await; + tokio::time::advance(WS_TERMINAL_FLUSH_TIMEOUT + Duration::from_millis(1)).await; + writer + .await + .expect("writer exits after bounded terminal flush"); + } + #[tokio::test] async fn send_loop_batches_queued_data_frames_into_one_flush() { let (data_tx, data_rx) = mpsc::channel(MAX_WS_SEND_BATCH); diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 02e2cc03a6..930fa7af79 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -15,9 +15,63 @@ use axum::extract::ws::Message as WsMessage; use tracing::{debug, info, warn}; use crate::connection::{AuthState, ConnectionState}; +use crate::metrics::{AuthOutcome, AuthPostTerminalState}; use crate::protocol::RelayMessage; use crate::state::AppState; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum BanOutcome { + Clear, + Banned, + DbError, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum PolicyCheck { + Allowed(T), + Denied, + DependencyError, +} + +fn classify_allowlist(result: Result) -> PolicyCheck<()> { + match result { + Ok(true) => PolicyCheck::Allowed(()), + Ok(false) => PolicyCheck::Denied, + Err(_) => PolicyCheck::DependencyError, + } +} + +fn classify_relay_membership( + result: Result, +) -> PolicyCheck> { + use crate::api::relay_members::MembershipDecision; + + match result { + Ok(MembershipDecision::OpenRelay | MembershipDecision::Member) => { + PolicyCheck::Allowed(None) + } + Ok(MembershipDecision::ViaOwner(owner)) => PolicyCheck::Allowed(Some(owner)), + Ok(MembershipDecision::Denied) => PolicyCheck::Denied, + Err(_) => PolicyCheck::DependencyError, + } +} + +fn ban_denial(outcome: BanOutcome) -> Option<(&'static str, &'static str, AuthOutcome)> { + match outcome { + BanOutcome::Clear => None, + BanOutcome::Banned => Some(( + "banned", + "blocked: you are banned from this community", + AuthOutcome::Banned, + )), + BanOutcome::DbError => Some(( + "ban_check_error", + "error: internal error checking restriction state", + AuthOutcome::BanCheckError, + )), + } +} + /// Extract a NIP-OA `auth` tag from a verified AUTH event and serialize it as /// the JSON-array string that [`buzz_sdk::nip_oa::verify_auth_tag`] expects. /// @@ -43,11 +97,13 @@ pub fn extract_auth_tag_json(event: &nostr::Event) -> Option { pub async fn handle_auth(event: nostr::Event, conn: Arc, state: Arc) { let event_id_hex = event.id.to_hex(); let (challenge, conn_id) = { - let auth = conn.auth_state.read().await; - match &*auth { - AuthState::Pending { challenge } => (challenge.clone(), conn.conn_id), + match conn.auth_state_snapshot() { + AuthState::Pending { challenge, .. } => (challenge, conn.conn_id), AuthState::Authenticated(_) => { debug!(conn_id = %conn.conn_id, "AUTH received but already authenticated"); + crate::metrics::record_post_terminal_auth_frame( + AuthPostTerminalState::Authenticated, + ); conn.send(RelayMessage::ok( &event_id_hex, false, @@ -57,6 +113,7 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: } AuthState::Failed => { debug!(conn_id = %conn.conn_id, "AUTH received after failed auth"); + crate::metrics::record_post_terminal_auth_frame(AuthPostTerminalState::Failed); conn.send(RelayMessage::ok( &event_id_hex, false, @@ -82,8 +139,6 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: crate::api::bridge::nip42_expected_relay_url(&state.config.relay_url, &conn.tenant); let auth_svc = Arc::clone(&state.auth); - metrics::counter!("buzz_auth_attempts_total", "method" => "nip42").increment(1); - // Pure NIP-42 verification — crypto only, no DB lookups. match auth_svc .verify_auth_event(event, &challenge, &relay_url) @@ -111,12 +166,6 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: // pinning `Failed` for the connection's life on a false premise. // `Banned` claims the ban; `DbError` denies with `error: internal` // (mirrors the ingest write-path gate). - enum BanOutcome { - Clear, - Banned, - DbError, - } - let mut outcome = match state .db .moderation_restriction_state(conn.tenant.community(), pubkey.as_bytes()) @@ -156,22 +205,13 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: } } - let denial: Option<(&str, &str)> = match outcome { - BanOutcome::Clear => None, - BanOutcome::Banned => { - Some(("banned", "blocked: you are banned from this community")) - } - BanOutcome::DbError => Some(( - "ban_check_error", - "error: internal error checking restriction state", - )), - }; - - if let Some((metric_reason, deny_reason)) = denial { + if let Some((metric_reason, deny_reason, auth_outcome)) = ban_denial(outcome) { warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), reason = deny_reason, "principal denied at ban seam"); metrics::counter!("buzz_auth_failures_total", "reason" => metric_reason) .increment(1); - *conn.auth_state.write().await = AuthState::Failed; + if !conn.reject_auth(auth_outcome) { + return; + } // Decision 4: banned ⇒ OK false + immediate WebSocket close. // Route the reason frame on the control channel (not `send`, // which uses the data channel and would race the cancel), so @@ -189,48 +229,68 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: if state.config.pubkey_allowlist_enabled && auth_ctx.auth_method == buzz_auth::AuthMethod::Nip42 { - let allowed = match state + let allowlist = state .db .is_pubkey_allowed(conn.tenant.community(), pubkey.as_bytes()) - .await - { - Ok(v) => v, - Err(e) => { - warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), error = %e, + .await; + if let Err(e) = &allowlist { + warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), error = %e, "allowlist DB lookup failed, denying (fail-closed)"); - false + } + match classify_allowlist(allowlist) { + PolicyCheck::Allowed(()) => {} + PolicyCheck::Denied => { + warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), "pubkey not in allowlist"); + metrics::counter!("buzz_auth_failures_total", "reason" => "allowlist_denied") + .increment(1); + if !conn.reject_auth(AuthOutcome::AllowlistDenied) { + return; + } + conn.send(RelayMessage::ok( + &event_id_hex, + false, + "auth-required: verification failed", + )); + return; + } + PolicyCheck::DependencyError => { + metrics::counter!("buzz_auth_failures_total", "reason" => "allowlist_check_error") + .increment(1); + if !conn.reject_auth(AuthOutcome::AllowlistCheckError) { + return; + } + conn.send(RelayMessage::ok( + &event_id_hex, + false, + "error: internal error checking allowlist", + )); + return; } - }; - if !allowed { - warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), "pubkey not in allowlist"); - metrics::counter!("buzz_auth_failures_total", "reason" => "allowlist_denied") - .increment(1); - *conn.auth_state.write().await = AuthState::Failed; - conn.send(RelayMessage::ok( - &event_id_hex, - false, - "auth-required: verification failed", - )); - return; } } // Relay membership gate — uses the shared helper with NIP-OA fallback. - let nip_oa_owner = match crate::api::relay_members::enforce_relay_membership( + let membership = crate::api::relay_members::check_relay_membership( &state, conn.tenant.community(), pubkey.as_bytes(), auth_tag_json.as_deref(), Some(signed_auth_created_at), ) - .await - { - Ok(owner) => owner, - Err(e) => { - warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), error = ?e, "not a relay member"); + .await; + if let Err(e) = &membership { + warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), error = %e, + "relay membership DB lookup failed, denying (fail-closed)"); + } + let nip_oa_owner = match classify_relay_membership(membership) { + PolicyCheck::Allowed(owner) => owner, + PolicyCheck::Denied => { + warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), "not a relay member"); metrics::counter!("buzz_auth_failures_total", "reason" => "not_relay_member") .increment(1); - *conn.auth_state.write().await = AuthState::Failed; + if !conn.reject_auth(AuthOutcome::NotRelayMember) { + return; + } conn.send(RelayMessage::ok( &event_id_hex, false, @@ -238,6 +298,19 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: )); return; } + PolicyCheck::DependencyError => { + metrics::counter!("buzz_auth_failures_total", "reason" => "relay_membership_check_error") + .increment(1); + if !conn.reject_auth(AuthOutcome::RelayMembershipCheckError) { + return; + } + conn.send(RelayMessage::ok( + &event_id_hex, + false, + "error: internal error checking relay membership", + )); + return; + } }; // Open relay NIP-OA backfill: extract owner for agent→owner DB mapping @@ -279,7 +352,9 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: } info!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), "NIP-42 auth successful"); - *conn.auth_state.write().await = AuthState::Authenticated(auth_ctx); + if !conn.authenticate(auth_ctx) { + return; + } state .conn_manager .set_authenticated_pubkey(conn_id, pubkey.to_bytes().to_vec()); @@ -288,7 +363,9 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: Err(e) => { warn!(conn_id = %conn_id, error = %e, "NIP-42 auth failed"); metrics::counter!("buzz_auth_failures_total", "reason" => "nip42_invalid").increment(1); - *conn.auth_state.write().await = AuthState::Failed; + if !conn.reject_auth(AuthOutcome::Invalid) { + return; + } conn.send(RelayMessage::ok( &event_id_hex, false, @@ -300,8 +377,53 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: #[cfg(test)] mod tests { - use super::extract_auth_tag_json; - use nostr::{EventBuilder, Keys, Kind, Tag}; + use super::{ + ban_denial, classify_allowlist, classify_relay_membership, extract_auth_tag_json, + handle_auth, BanOutcome, PolicyCheck, + }; + use crate::api::relay_members::MembershipDecision; + use crate::connection::{tests::test_conn_with_auth, AuthState}; + use crate::metrics::{AuthOutcome, AuthPostTerminalState}; + use metrics_util::debugging::DebugValue; + use nostr::{EventBuilder, Keys, Kind, RelayUrl, Tag}; + use std::time::Instant; + + type MetricSnapshot = Vec<( + metrics_util::CompositeKey, + Option, + Option, + DebugValue, + )>; + + fn metric_counter(snapshot: &MetricSnapshot, name: &str, outcome: Option<&str>) -> u64 { + snapshot + .iter() + .find_map(|(key, _, _, value)| { + if key.key().name() != name { + return None; + } + let labels = key.key().labels().collect::>(); + if outcome.is_some_and(|expected| { + !labels + .iter() + .any(|label| label.key() == "outcome" && label.value() == expected) + }) { + return None; + } + let DebugValue::Counter(value) = value else { + panic!("{name} must be a counter"); + }; + Some(*value) + }) + .unwrap_or_default() + } + + fn pending(challenge: &str) -> AuthState { + AuthState::Pending { + challenge: challenge.to_owned(), + started_at: Instant::now(), + } + } /// Build a signed NIP-98 (kind 27235) event carrying the given tags. The /// `auth` tag lives inside the signed event exactly as the git and @@ -351,4 +473,194 @@ mod tests { ]); assert_eq!(extract_auth_tag_json(&event), None); } + + #[test] + fn ban_decisions_map_to_bounded_public_outcomes() { + assert_eq!(ban_denial(BanOutcome::Clear), None); + assert_eq!( + ban_denial(BanOutcome::Banned), + Some(( + "banned", + "blocked: you are banned from this community", + AuthOutcome::Banned, + )) + ); + assert_eq!( + ban_denial(BanOutcome::DbError), + Some(( + "ban_check_error", + "error: internal error checking restriction state", + AuthOutcome::BanCheckError, + )) + ); + } + + #[test] + fn dependency_failures_are_distinct_from_policy_denials() { + assert_eq!( + classify_allowlist(Ok::<_, &str>(true)), + PolicyCheck::Allowed(()) + ); + assert_eq!( + classify_allowlist(Ok::<_, &str>(false)), + PolicyCheck::Denied + ); + assert_eq!( + classify_allowlist(Err::("database unavailable")), + PolicyCheck::DependencyError + ); + + let owner = Keys::generate().public_key(); + assert_eq!( + classify_relay_membership(Ok(MembershipDecision::OpenRelay)), + PolicyCheck::Allowed(None) + ); + assert_eq!( + classify_relay_membership(Ok(MembershipDecision::Member)), + PolicyCheck::Allowed(None) + ); + assert_eq!( + classify_relay_membership(Ok(MembershipDecision::ViaOwner(owner))), + PolicyCheck::Allowed(Some(owner)) + ); + assert_eq!( + classify_relay_membership(Ok(MembershipDecision::Denied)), + PolicyCheck::Denied + ); + assert_eq!( + classify_relay_membership(Err("database unavailable".to_owned())), + PolicyCheck::DependencyError + ); + } + + /// The handler owns retry classification, so drive its real terminal-state + /// branches rather than calling the metric helper directly. A malformed + /// signature also traverses the real NIP-42 verifier before terminalizing. + #[tokio::test(flavor = "current_thread")] + async fn handler_separates_post_terminal_frames_from_invalid_attempts() { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + let state = crate::state::tests::test_state().await; + + let (authenticated, mut authenticated_rx) = + test_conn_with_auth(crate::connection::tests::authenticated_state()); + let duplicate = signed_event_with_tags(Vec::new()); + handle_auth(duplicate, authenticated, state.clone()).await; + let duplicate_frame = crate::connection::tests::read_frame(&mut authenticated_rx); + assert_eq!(duplicate_frame[2], false); + assert_eq!(duplicate_frame[3], "auth-required: already authenticated"); + + let (failed, mut failed_rx) = test_conn_with_auth(AuthState::Failed); + let after_failure = signed_event_with_tags(Vec::new()); + handle_auth(after_failure, failed, state.clone()).await; + let failed_frame = crate::connection::tests::read_frame(&mut failed_rx); + assert_eq!(failed_frame[2], false); + assert_eq!( + failed_frame[3], + "auth-required: authentication already failed" + ); + + let challenge = "invalid-signature-challenge"; + let (invalid_conn, mut invalid_rx) = test_conn_with_auth(pending(challenge)); + crate::metrics::record_auth_attempt_started(); + let relay_url: RelayUrl = crate::api::bridge::nip42_expected_relay_url( + &state.config.relay_url, + &invalid_conn.tenant, + ) + .parse() + .expect("test relay URL"); + let mut invalid = EventBuilder::auth(challenge, relay_url) + .sign_with_keys(&Keys::generate()) + .expect("sign auth event"); + invalid.content.push('x'); + handle_auth(invalid, invalid_conn.clone(), state).await; + let invalid_frame = crate::connection::tests::read_frame(&mut invalid_rx); + assert_eq!(invalid_frame[2], false); + assert_eq!(invalid_frame[3], "auth-required: verification failed"); + assert!(matches!( + invalid_conn.auth_state_snapshot(), + AuthState::Failed + )); + + let snapshot = snapshotter.snapshot().into_vec(); + let attempts = metric_counter(&snapshot, "buzz_auth_attempts_total", None); + assert_eq!(attempts, 1); + assert_eq!( + metric_counter( + &snapshot, + "buzz_auth_outcomes_total", + Some(AuthOutcome::Invalid.as_str()), + ), + 1 + ); + for state in AuthPostTerminalState::ALL { + let count = snapshot + .iter() + .find_map(|(key, _, _, value)| { + (key.key().name() == "buzz_auth_post_terminal_frames_total" + && key + .key() + .labels() + .any(|label| label.key() == "state" && label.value() == state.as_str())) + .then(|| match value { + DebugValue::Counter(value) => *value, + _ => panic!("post-terminal frame metric must be a counter"), + }) + }) + .unwrap_or_default(); + assert_eq!(count, 1, "{} post-terminal frame", state.as_str()); + } + } + + /// A verified signature followed by an unavailable restriction database + /// must deny fail-closed and expose a dependency error, not mislabel the + /// principal as banned or let the attempt disappear. + #[tokio::test(flavor = "current_thread")] + async fn handler_accounts_ban_check_database_error() { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + let state = crate::state::tests::test_state_with_database_url( + "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", + ) + .await; + + let challenge = "ban-check-error-challenge"; + let (conn, _rx) = test_conn_with_auth(pending(challenge)); + crate::metrics::record_auth_attempt_started(); + let relay_url: RelayUrl = + crate::api::bridge::nip42_expected_relay_url(&state.config.relay_url, &conn.tenant) + .parse() + .expect("test relay URL"); + let event = EventBuilder::auth(challenge, relay_url) + .sign_with_keys(&Keys::generate()) + .expect("sign auth event"); + + handle_auth(event, conn.clone(), state).await; + + assert!(matches!(conn.auth_state_snapshot(), AuthState::Failed)); + assert!(conn.cancel.is_cancelled()); + let snapshot = snapshotter.snapshot().into_vec(); + assert_eq!( + metric_counter( + &snapshot, + "buzz_auth_outcomes_total", + Some(AuthOutcome::BanCheckError.as_str()), + ), + 1 + ); + assert_eq!( + metric_counter( + &snapshot, + "buzz_auth_outcomes_total", + Some(AuthOutcome::Banned.as_str()), + ), + 0 + ); + assert_eq!( + metric_counter(&snapshot, "buzz_auth_attempts_total", None), + 1 + ); + } } diff --git a/crates/buzz-relay/src/handlers/count.rs b/crates/buzz-relay/src/handlers/count.rs index 938674301e..5ea30b238b 100644 --- a/crates/buzz-relay/src/handlers/count.rs +++ b/crates/buzz-relay/src/handlers/count.rs @@ -23,8 +23,7 @@ pub async fn handle_count( ) { // Require auth let (pubkey_bytes, token_channel_ids) = { - let auth = conn.auth_state.read().await; - match &*auth { + match conn.auth_state_snapshot() { AuthState::Authenticated(ctx) => { (ctx.pubkey.to_bytes().to_vec(), ctx.channel_ids.clone()) } diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 082522b123..c117392c16 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -632,8 +632,7 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc ( conn.conn_id, ctx.pubkey.to_bytes().to_vec(), @@ -1059,8 +1058,7 @@ async fn handle_agent_observer_event( // Fast path: if this connection authenticated via NIP-OA and the verified // owner matches the observer frame's target owner, skip the DB lookup entirely. let session_owner_match = { - let auth = conn.auth_state.read().await; - if let crate::connection::AuthState::Authenticated(ctx) = &*auth { + if let crate::connection::AuthState::Authenticated(ctx) = conn.auth_state_snapshot() { ctx.agent_owner_pubkey.as_ref() == Some(&route.owner) } else { false @@ -1230,7 +1228,7 @@ mod tests { OBSERVER_FRAME_TELEMETRY, }; use nostr::{EventBuilder, Keys, Kind, Tag}; - use tokio::sync::{mpsc, Mutex, RwLock}; + use tokio::sync::{mpsc, Mutex}; use tokio_util::sync::CancellationToken; use uuid::Uuid; @@ -1447,7 +1445,7 @@ mod tests { conn_id: Uuid::new_v4(), tenant: buzz_core::TenantContext::resolved(community_b, "b.example"), remote_addr: "127.0.0.1:1234".parse().expect("socket addr"), - auth_state: RwLock::new(crate::connection::AuthState::Authenticated( + auth_state: std::sync::Mutex::new(crate::connection::AuthState::Authenticated( buzz_auth::AuthContext { pubkey: agent.public_key(), scopes: vec![], diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 9a141e86ff..97d252cb7e 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -57,8 +57,7 @@ pub async fn handle_req( state: Arc, ) { let (conn_id, pubkey_bytes, token_channel_ids) = { - let auth = conn.auth_state.read().await; - match &*auth { + match conn.auth_state_snapshot() { AuthState::Authenticated(ctx) => { if !ctx.scopes.is_empty() && !ctx.scopes.contains(&Scope::MessagesRead) { conn.send(RelayMessage::notice("restricted: insufficient scope")); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index f2e539604c..d6b3b55124 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1609,6 +1609,7 @@ impl InMemoryMetricKey { /// racing the lifecycle-relative increments and decrements. fn refresh_legacy_active_gauge_recency() { metrics::gauge!("buzz_ws_connections_active").increment(0.0); + metrics::gauge!("buzz_ws_authenticated_connections_active").increment(0.0); metrics::gauge!("buzz_subscriptions_active").increment(0.0); } @@ -2291,13 +2292,16 @@ mod tests { metrics::with_local_recorder(&recorder, || { let connections = metrics::gauge!("buzz_ws_connections_active"); + let authenticated = metrics::gauge!("buzz_ws_authenticated_connections_active"); let subscriptions = metrics::gauge!("buzz_subscriptions_active"); connections.increment(1.0); + authenticated.increment(1.0); subscriptions.increment(1.0); refresh_legacy_active_gauge_recency(); connections.decrement(1.0); + authenticated.decrement(1.0); subscriptions.increment(1.0); }); @@ -2314,6 +2318,10 @@ mod tests { .collect::>(); assert_eq!(values.get("buzz_ws_connections_active"), Some(&0.0)); + assert_eq!( + values.get("buzz_ws_authenticated_connections_active"), + Some(&0.0) + ); assert_eq!(values.get("buzz_subscriptions_active"), Some(&2.0)); } diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 761b821d5e..a81a432764 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -24,6 +24,80 @@ use axum::{ use metrics_exporter_prometheus::{BuildError, Matcher, PrometheusBuilder}; use metrics_util::MetricKindMask; +/// The fixed method label for the WebSocket NIP-42 challenge lifecycle. +pub(crate) const AUTH_METHOD_NIP42: &str = "nip42"; + +/// Bounded terminal results for one WebSocket authentication attempt. +/// +/// Keep this enum free of user-controlled or error-string values. Rollout +/// analysis relies on these labels having a small, stable cardinality. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum AuthOutcome { + Success, + Invalid, + Banned, + BanCheckError, + AllowlistCheckError, + AllowlistDenied, + RelayMembershipCheckError, + NotRelayMember, + Timeout, + Disconnect, + Shutdown, +} + +impl AuthOutcome { + pub(crate) const ALL: [Self; 11] = [ + Self::Success, + Self::Invalid, + Self::Banned, + Self::BanCheckError, + Self::AllowlistCheckError, + Self::AllowlistDenied, + Self::RelayMembershipCheckError, + Self::NotRelayMember, + Self::Timeout, + Self::Disconnect, + Self::Shutdown, + ]; + + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Success => "success", + Self::Invalid => "invalid", + Self::Banned => "banned", + Self::BanCheckError => "ban_check_error", + Self::AllowlistCheckError => "allowlist_check_error", + Self::AllowlistDenied => "allowlist_denied", + Self::RelayMembershipCheckError => "relay_membership_check_error", + Self::NotRelayMember => "not_relay_member", + Self::Timeout => "timeout", + Self::Disconnect => "disconnect", + Self::Shutdown => "shutdown", + } + } +} + +/// Bounded state of an AUTH frame received after the issued challenge had +/// already reached a terminal. These frames are protocol misuse, not recovery +/// lifecycles, and therefore stay outside rollout-gating attempts/outcomes. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum AuthPostTerminalState { + Authenticated, + Failed, +} + +impl AuthPostTerminalState { + pub(crate) const ALL: [Self; 2] = [Self::Authenticated, Self::Failed]; + + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Authenticated => "authenticated", + Self::Failed => "failed", + } + } +} + /// HTTP latency buckets (milliseconds) — only for `http_request_latency_ms`. const LATENCY_BUCKETS_MS: [f64; 11] = [ 5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0, 1000.0, 2500.0, 5000.0, 10000.0, @@ -206,6 +280,8 @@ pub fn try_install(port: u16, gauge_idle_timeout_secs: u64) -> Result<(), Metric .map_err(|_error| MetricsInstallError::RecorderConflict)?; describe_readiness_metrics(); describe_db_pool_metrics(); + describe_auth_metrics(); + initialize_auth_metric_series(); tokio::spawn(exporter); Ok(()) } @@ -332,6 +408,106 @@ pub fn record_db_pool_metrics(input: DbPoolMetricsInput) { } } +/// Register the bounded WebSocket authentication metric contract. +pub(crate) fn describe_auth_metrics() { + metrics::describe_counter!( + "buzz_auth_attempts_total", + "NIP-42 challenge lifecycles issued to WebSocket clients by bounded method" + ); + metrics::describe_counter!( + "buzz_auth_outcomes_total", + "WebSocket authentication attempts by bounded method and terminal outcome" + ); + metrics::describe_histogram!( + "buzz_auth_duration_seconds", + metrics::Unit::Seconds, + "WebSocket authentication attempt duration by bounded method and terminal outcome" + ); + metrics::describe_gauge!( + "buzz_ws_authenticated_connections_active", + "Current WebSocket connections that completed authentication on this pod" + ); + metrics::describe_counter!( + "buzz_auth_post_terminal_frames_total", + "AUTH frames received after the issued challenge reached a terminal, by bounded state" + ); +} + +/// Publish zero-valued authentication series at process start. +/// +/// Counters are never idle-evicted by the configured recorder. The active +/// gauge is refreshed alongside the other lifecycle-relative gauges in +/// `main.rs`, so a quiet healthy pod remains distinguishable from a missing +/// scrape target without racing its increments and decrements. +pub(crate) fn initialize_auth_metric_series() { + metrics::counter!( + "buzz_auth_attempts_total", + "method" => AUTH_METHOD_NIP42 + ) + .increment(0); + for outcome in AuthOutcome::ALL { + metrics::counter!( + "buzz_auth_outcomes_total", + "method" => AUTH_METHOD_NIP42, + "outcome" => outcome.as_str() + ) + .increment(0); + // Register the fixed-label histogram without recording a fake sample. + // Prometheus then exports zero buckets, `_sum`, and `_count` at boot. + let _ = metrics::histogram!( + "buzz_auth_duration_seconds", + "method" => AUTH_METHOD_NIP42, + "outcome" => outcome.as_str() + ); + } + for state in AuthPostTerminalState::ALL { + metrics::counter!( + "buzz_auth_post_terminal_frames_total", + "state" => state.as_str() + ) + .increment(0); + } + metrics::gauge!("buzz_ws_authenticated_connections_active").set(0.0); +} + +/// Record one issued NIP-42 challenge after it enters the connection writer. +/// +/// Before this lifecycle contract, `buzz_auth_attempts_total` incremented only +/// when an AUTH frame reached the pending handler. It now uses one consistent, +/// recovery-oriented unit: challenges successfully queued to the connection. +pub(crate) fn record_auth_attempt_started() { + metrics::counter!( + "buzz_auth_attempts_total", + "method" => AUTH_METHOD_NIP42 + ) + .increment(1); +} + +/// Record exactly one bounded terminal for an authentication attempt. +pub(crate) fn record_auth_outcome(outcome: AuthOutcome, duration: Duration) { + metrics::counter!( + "buzz_auth_outcomes_total", + "method" => AUTH_METHOD_NIP42, + "outcome" => outcome.as_str() + ) + .increment(1); + metrics::histogram!( + "buzz_auth_duration_seconds", + "method" => AUTH_METHOD_NIP42, + "outcome" => outcome.as_str() + ) + .record(duration.as_secs_f64()); +} + +/// Record protocol noise after the issued challenge already terminalized. +pub(crate) fn record_post_terminal_auth_frame(state: AuthPostTerminalState) { + metrics::counter!( + "buzz_auth_post_terminal_frames_total", + "state" => state.as_str() + ) + .increment(1); +} + #[cfg(test)] pub(crate) fn readiness_test_recorder() -> ( metrics_exporter_prometheus::PrometheusRecorder, @@ -404,6 +580,7 @@ pub async fn track_metrics(req: Request, next: Next) -> Response { #[cfg(test)] mod contract_tests { use std::collections::BTreeSet; + use std::time::Duration; const OUTCOMES: [&str; 4] = ["success", "timeout", "error", "cancelled"]; @@ -719,6 +896,109 @@ mod contract_tests { assert!(!line.contains("result=")); } } + + #[test] + fn production_builder_exports_bounded_auth_contract_and_stable_zeros() { + let (recorder, handle) = super::readiness_test_recorder(); + metrics::with_local_recorder(&recorder, || { + super::describe_auth_metrics(); + super::initialize_auth_metric_series(); + }); + + let stable_zero_scrape = handle.render(); + assert!(stable_zero_scrape.contains("buzz_auth_attempts_total{method=\"nip42\"} 0")); + assert!(stable_zero_scrape.contains("buzz_ws_authenticated_connections_active 0")); + let zero_outcomes = stable_zero_scrape + .lines() + .filter(|line| line.starts_with("buzz_auth_outcomes_total{") && line.ends_with(" 0")) + .collect::>(); + assert_eq!( + zero_outcomes.len(), + super::AuthOutcome::ALL.len(), + "every bounded outcome must exist before the first attempt:\n{stable_zero_scrape}" + ); + let zero_duration_counts = stable_zero_scrape + .lines() + .filter(|line| { + line.starts_with("buzz_auth_duration_seconds_count{") && line.ends_with(" 0") + }) + .count(); + let zero_duration_sums = stable_zero_scrape + .lines() + .filter(|line| { + line.starts_with("buzz_auth_duration_seconds_sum{") && line.ends_with(" 0") + }) + .count(); + let zero_duration_buckets = stable_zero_scrape + .lines() + .filter(|line| { + line.starts_with("buzz_auth_duration_seconds_bucket{") && line.ends_with(" 0") + }) + .count(); + assert_eq!(zero_duration_counts, super::AuthOutcome::ALL.len()); + assert_eq!(zero_duration_sums, super::AuthOutcome::ALL.len()); + assert_eq!( + zero_duration_buckets, + super::AuthOutcome::ALL.len() * 11, + "every fixed outcome must export all zero buckets before traffic:\n{stable_zero_scrape}" + ); + + metrics::with_local_recorder(&recorder, || { + super::record_auth_attempt_started(); + for outcome in super::AuthOutcome::ALL { + super::record_auth_outcome(outcome, Duration::from_millis(20)); + } + for state in super::AuthPostTerminalState::ALL { + super::record_post_terminal_auth_frame(state); + } + let active = metrics::gauge!("buzz_ws_authenticated_connections_active"); + active.increment(1.0); + active.decrement(1.0); + }); + + let scrape = handle.render(); + assert!(scrape.contains("# TYPE buzz_auth_attempts_total counter")); + assert!(scrape.contains("# TYPE buzz_auth_outcomes_total counter")); + assert!(scrape.contains("# TYPE buzz_auth_duration_seconds histogram")); + assert!(scrape.contains("# TYPE buzz_ws_authenticated_connections_active gauge")); + + let raw_series = scrape + .lines() + .filter(|line| { + line.starts_with("buzz_auth_attempts_total{") + || line.starts_with("buzz_auth_outcomes_total{") + || line.starts_with("buzz_auth_duration_seconds") + || line.starts_with("buzz_auth_post_terminal_frames_total{") + || line.starts_with("buzz_ws_authenticated_connections_active ") + }) + .collect::>(); + // 1 challenge + 11 outcomes + 2 post-terminal states + 1 active gauge +, for each outcome, + // 11 histogram buckets (including +Inf), sum, and count. + assert_eq!(raw_series.len(), 158, "unexpected raw scrape:\n{scrape}"); + + for line in raw_series { + let keys = label_keys(line); + if line.starts_with("buzz_auth_attempts_total{") { + assert_eq!(keys, BTreeSet::from(["method"])); + } else if line.starts_with("buzz_auth_outcomes_total{") { + assert_eq!(keys, BTreeSet::from(["method", "outcome"])); + } else if line.starts_with("buzz_auth_post_terminal_frames_total{") { + assert_eq!(keys, BTreeSet::from(["state"])); + } else if line.starts_with("buzz_auth_duration_seconds_bucket{") { + assert_eq!(keys, BTreeSet::from(["le", "method", "outcome"])); + } else if line.starts_with("buzz_auth_duration_seconds") { + assert_eq!(keys, BTreeSet::from(["method", "outcome"])); + } else { + assert!(keys.is_empty(), "active gauge must not add labels: {line}"); + } + assert!(!line.contains("pubkey=")); + assert!(!line.contains("community=")); + assert!(!line.contains("connection=")); + assert!(!line.contains("pod=")); + assert!(!line.contains("version=")); + assert!(!line.contains("error=")); + } + } } #[cfg(test)] diff --git a/crates/buzz-relay/src/rejection.rs b/crates/buzz-relay/src/rejection.rs index 96b8074e55..733c427f95 100644 --- a/crates/buzz-relay/src/rejection.rs +++ b/crates/buzz-relay/src/rejection.rs @@ -57,8 +57,7 @@ pub(crate) async fn enforce_ws_admission( } let (pubkey, is_agent) = { - let auth = conn.auth_state.read().await; - match &*auth { + match conn.auth_state_snapshot() { AuthState::Authenticated(ctx) => (ctx.pubkey, ctx.agent_owner_pubkey.is_some()), _ => return true, } diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 95372d5bc3..98864df49e 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1417,7 +1417,7 @@ pub(crate) mod tests { use super::*; use crate::connection::{AuthState, ConnectionState}; use std::collections::HashMap; - use tokio::sync::{Mutex, RwLock}; + use tokio::sync::Mutex; /// Helper: create a ConnectionManager with one registered connection. /// Returns (manager, conn_id, receiver, ctrl_receiver, cancel, @@ -1460,6 +1460,37 @@ pub(crate) mod tests { config.require_relay_membership = false; config.redis_url = "redis://127.0.0.1:1".to_string(); let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool"); + build_test_state(config, pool).await + } + + /// The same test state with an explicit database target. This lets handler + /// tests deterministically exercise fail-closed database seams without + /// depending on whether a developer has the normal test database running. + pub(crate) async fn test_state_with_database_url(database_url: &str) -> Arc { + 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(); + config.database_url = database_url.to_owned(); + config.read_database_url = None; + let pool = sqlx::postgres::PgPoolOptions::new() + .acquire_timeout(std::time::Duration::from_millis(100)) + .connect_lazy(&config.database_url) + .expect("lazy pg pool"); + build_test_state(config, pool).await + } + + /// Build test state around a caller-owned writer pool. Production-path + /// lifecycle tests use this to hold the sole connection as a deterministic + /// barrier while AUTH waits in the real database acquisition path. + pub(crate) async fn test_state_with_database_pool(pool: sqlx::PgPool) -> Arc { + 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(); + config.read_database_url = None; + build_test_state(config, pool).await + } + + async fn build_test_state(config: crate::config::Config, pool: sqlx::PgPool) -> Arc { let db = buzz_db::Db::from_pool(pool.clone()); let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) .create_pool(Some(deadpool_redis::Runtime::Tokio1)) @@ -1688,7 +1719,7 @@ pub(crate) mod tests { "test.local".to_string(), ), remote_addr: "127.0.0.1:1234".parse().unwrap(), - auth_state: RwLock::new(AuthState::Failed), + auth_state: std::sync::Mutex::new(AuthState::Failed), subscriptions: Arc::new(Mutex::new(HashMap::new())), send_tx: tx.clone(), ctrl_tx, diff --git a/crates/buzz-relay/tests/boot_lifecycle.rs b/crates/buzz-relay/tests/boot_lifecycle.rs index 29fcf991f8..17762bbdea 100644 --- a/crates/buzz-relay/tests/boot_lifecycle.rs +++ b/crates/buzz-relay/tests/boot_lifecycle.rs @@ -196,6 +196,50 @@ fn assert_no_startup_lifecycle_metrics(scrape: &str) { } } +fn assert_auth_metrics_have_stable_boot_zeros(scrape: &str) { + const OUTCOMES: [&str; 11] = [ + "success", + "invalid", + "banned", + "ban_check_error", + "allowlist_check_error", + "allowlist_denied", + "relay_membership_check_error", + "not_relay_member", + "timeout", + "disconnect", + "shutdown", + ]; + + assert!(scrape.contains("# TYPE buzz_auth_attempts_total counter")); + assert!(scrape.contains("buzz_auth_attempts_total{method=\"nip42\"} 0")); + assert!(scrape.contains("# TYPE buzz_auth_outcomes_total counter")); + assert!(scrape.contains("# TYPE buzz_auth_duration_seconds histogram")); + assert!(scrape.contains("# TYPE buzz_auth_post_terminal_frames_total counter")); + assert!(scrape.contains("buzz_auth_post_terminal_frames_total{state=\"authenticated\"} 0")); + assert!(scrape.contains("buzz_auth_post_terminal_frames_total{state=\"failed\"} 0")); + assert!(scrape.contains("buzz_ws_authenticated_connections_active 0")); + + for outcome in OUTCOMES { + assert!(scrape.contains(&format!( + "buzz_auth_outcomes_total{{method=\"nip42\",outcome=\"{outcome}\"}} 0" + ))); + assert!(scrape.contains(&format!( + "buzz_auth_duration_seconds_count{{method=\"nip42\",outcome=\"{outcome}\"}} 0" + ))); + assert!(scrape.contains(&format!( + "buzz_auth_duration_seconds_sum{{method=\"nip42\",outcome=\"{outcome}\"}} 0" + ))); + assert!(scrape.lines().any(|line| { + line.starts_with("buzz_auth_duration_seconds_bucket{") + && line.contains("le=\"+Inf\"") + && line.contains("method=\"nip42\"") + && line.contains(&format!("outcome=\"{outcome}\"")) + && line.ends_with(" 0") + })); + } +} + fn lifecycle_events(output: &Output) -> Vec { let mut events: Vec = output .stdout @@ -444,6 +488,7 @@ fn successful_main_emits_complete_lifecycle_without_startup_metrics() { ]); let scrape = wait_for_relay_metrics(&mut process, port); assert_no_startup_lifecycle_metrics(&scrape); + assert_auth_metrics_have_stable_boot_zeros(&scrape); let output = process.terminate(); let events = lifecycle_events(&output);