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