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 <rarora@squareup.com>

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<T> {
+    Allowed(T),
+    Denied,
+    DependencyError,
+}
+
+fn classify_allowlist<E>(result: Result<bool, E>) -> PolicyCheck<()> {
+    match result {
+        Ok(true) => PolicyCheck::Allowed(()),
+        Ok(false) => PolicyCheck::Denied,
+        Err(_) => PolicyCheck::DependencyError,
+    }
+}
+
+fn classify_relay_membership(
+    result: Result<crate::api::relay_members::MembershipDecision, String>,
+) -> PolicyCheck<Option<nostr::PublicKey>> {
+    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<String> {
 pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state: Arc<AppState>) {
     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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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<metrics::Unit>,
+        Option<metrics::SharedString>,
+        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::<Vec<_>>();
+                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::<bool, _>("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
+        );
+    }
 }

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())
             }

crates/buzz-relay/src/handlers/event.rs

@@ -632,8 +632,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
     .increment(1);
 
     let (conn_id, pubkey_bytes, auth_pubkey, scopes, channel_ids) = {
-        let auth = conn.auth_state.read().await;
-        match &*auth {
+        match conn.auth_state_snapshot() {
             AuthState::Authenticated(ctx) => (
                 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![],

crates/buzz-relay/src/handlers/req.rs

@@ -57,8 +57,7 @@ pub async fn handle_req(
     state: Arc<AppState>,
 ) {
     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"));

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::<std::collections::HashMap<_, _>>();
 
         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));
     }
 

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::<Vec<_>>();
+        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::<Vec<_>>();
+        // 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)]

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,
         }

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<AppState> {
+        let mut config = crate::config::Config::from_env().expect("default config loads");
+        config.require_relay_membership = false;
+        config.redis_url = "redis://127.0.0.1:1".to_string();
+        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<AppState> {
+        let mut config = crate::config::Config::from_env().expect("default config loads");
+        config.require_relay_membership = false;
+        config.redis_url = "redis://127.0.0.1:1".to_string();
+        config.read_database_url = None;
+        build_test_state(config, pool).await
+    }
+
+    async fn build_test_state(config: crate::config::Config, pool: sqlx::PgPool) -> Arc<AppState> {
         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,

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<Value> {
     let mut events: Vec<Value> = 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);