Record why relay connections close and quiet close warnings (#9245)

Since bb-public prod moved to `RUST_LOG=error,buzz_relay=info`, two
connection-close warnings send about 1.9M lines a day to Datadog. This
moves that information to the access log and a metric.

- **`close_reason` on the access `close` line.** The first reason
recorded wins. When the server has already decided to close (shutdown, a
policy disconnect, or a cancellation), the reader defers to that
decision, so a peer's reply to the server's Close is never logged as the
cause.
- Values: `auth_timeout`, `auth_failed` (an AUTH rejected and closed
immediately, or a timeout after a rejected AUTH), `missed_pongs`,
`client_closed`, `socket_error`, `frame_too_large`,
`control_channel_full`, `shutdown`, `server_closed`, `refused` for
connections refused before they were served, and `handler_dropped` for a
handler dropped before normal cleanup.
- A community disconnect logs its own value: `community_archived`,
`community_deleted`, `auth_policy_upgraded`, `authorization_denied`,
`access_revoked` or `removed`.
- **The "NIP-42 auth timeout" and "3 missed pongs" warnings are now
`debug!`**, with the same text and fields.
- **New counter `buzz_ws_missed_pong_disconnects_total`**, next to
`buzz_ws_auth_timeouts_total`. A heartbeat tick that races a close
already decided is not counted.

Timeouts, ping cadence, the frames sent and when connections close are
unchanged. The new infra-free tests are added to the `test-unit`
selection and `scripts/run-tests.sh`.

---------

Signed-off-by: Will Pfleger <wpfleger@block.xyz>
Co-authored-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>

crates/buzz-relay/src/access.rs

@@ -1057,6 +1057,8 @@ where
 /// after a normal close, a refusal during startup, or the handler being
 /// dropped. `pubkey` is the identity it logged in as, if any, with its
 /// `method` and `pubkey_verified`; the login also wrote its own `auth` line.
+/// `close_reason` is why the connection ended: the first reason recorded wins,
+/// else `refused` for a refusal, else `handler_dropped`.
 pub(crate) struct CloseLine {
     conn_id: Uuid,
     community: CommunityId,
@@ -1065,6 +1067,7 @@ pub(crate) struct CloseLine {
     opened: std::time::Instant,
     pubkey: std::sync::Mutex<Option<PublicKey>>,
     refused: std::sync::OnceLock<&'static str>,
+    close_reason: std::sync::OnceLock<&'static str>,
     pub(crate) queued: std::sync::Arc<QueuedTotals>,
     pub(crate) refusals: std::sync::Arc<RefusalBudget>,
 }
@@ -1085,6 +1088,7 @@ impl CloseLine {
             opened: std::time::Instant::now(),
             pubkey: std::sync::Mutex::new(None),
             refused: std::sync::OnceLock::new(),
+            close_reason: std::sync::OnceLock::new(),
             queued: Default::default(),
             refusals: Default::default(),
         })
@@ -1103,6 +1107,12 @@ impl CloseLine {
     pub(crate) fn refused(&self, reason: &'static str) {
         let _ = self.refused.set(reason);
     }
+
+    /// Record why the connection ended. Only the first reason is kept, so the
+    /// cause recorded before cancelling outranks the generic one cleanup adds.
+    pub(crate) fn closed(&self, reason: &'static str) {
+        let _ = self.close_reason.set(reason);
+    }
 }
 
 impl Drop for CloseLine {
@@ -1121,6 +1131,12 @@ impl Drop for CloseLine {
             peer_addr = %self.peer_addr,
             client_ip = self.client_ip.map(tracing::field::display),
             refused = self.refused.get().copied(),
+            close_reason = self
+                .close_reason
+                .get()
+                .copied()
+                .or(self.refused.get().map(|_| "refused"))
+                .unwrap_or("handler_dropped"),
             pubkey = pubkey.as_deref(),
             method = pubkey.is_some().then_some(AuthMethod::Nip42.as_str()),
             pubkey_verified = pubkey.is_some().then_some(true),
@@ -1809,6 +1825,36 @@ pub(crate) mod tests {
         assert_eq!(RefusalBudget::close(&budget.auths), 0);
     }
 
+    /// The first recorded close reason wins; a refusal with none closes as
+    /// `refused`, and a line dropped with neither as `handler_dropped`.
+    #[test]
+    fn close_line_keeps_the_first_reason_and_defaults_refusals() {
+        let open = || {
+            CloseLine::open(
+                Uuid::nil(),
+                CommunityId::from_uuid(Uuid::nil()),
+                "127.0.0.1:1".parse().unwrap(),
+                None,
+            )
+        };
+        let ((), out) = capture(|| {
+            let close = open();
+            close.closed("auth_timeout");
+            close.closed("server_closed");
+            drop(close);
+            let refused = open();
+            refused.refused("connection_limit");
+            drop(refused);
+            drop(open());
+        });
+        let reasons: Vec<_> = out
+            .lines()
+            .iter()
+            .map(|l| l["close_reason"].clone())
+            .collect();
+        assert_eq!(reasons, ["auth_timeout", "refused", "handler_dropped"]);
+    }
+
     /// Refusals charged after `close` was written each log their own line,
     /// so every refusal is either counted on `close` or logged, never both.
     #[test]

crates/buzz-relay/src/connection.rs

@@ -200,17 +200,18 @@ impl ConnectionState {
 
     /// 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 {
+    /// Returns the `close_reason` to close with, if the connection must close.
+    fn expire_auth(&self) -> Option<&'static str> {
         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
+                Some("auth_timeout")
             }
-            AuthState::Failed => true,
-            AuthState::Authenticated(_) => false,
+            AuthState::Failed => Some("auth_failed"),
+            AuthState::Authenticated(_) => None,
         }
     }
 
@@ -653,6 +654,7 @@ async fn handle_active_connection(
         .is_err()
     {
         warn!(conn_id = %conn_id, "Failed to send AUTH challenge — client disconnected immediately");
+        close.closed("client_closed");
         return;
     }
 
@@ -697,15 +699,18 @@ async fn handle_active_connection(
         ctrl_tx,
         Arc::clone(&missed_pongs),
         heartbeat_cancel,
+        Arc::clone(&close),
     ));
 
     let auth_timeout_conn = Arc::clone(&conn);
     let auth_timeout_cancel = cancel.clone();
+    let auth_timeout_close = Arc::clone(&close);
     let auth_timeout_task = tokio::spawn(async move {
         tokio::select! {
             _ = tokio::time::sleep(AUTH_TIMEOUT) => {
-                if auth_timeout_conn.expire_auth() {
-                    warn!(
+                if let Some(reason) = auth_timeout_conn.expire_auth() {
+                    auth_timeout_close.closed(reason);
+                    debug!(
                         conn_id = %auth_timeout_conn.conn_id,
                         timeout_secs = AUTH_TIMEOUT.as_secs(),
                         "NIP-42 auth timeout — closing connection"
@@ -759,10 +764,13 @@ async fn handle_active_connection(
         Arc::clone(&state),
         Arc::clone(&missed_pongs),
         cancel.clone(),
+        &close,
     )
     .await;
 
     cancel.cancel();
+    // Whoever cancelled first recorded its own reason; this fills in the rest.
+    close.closed(server_close_reason(&state, &conn).unwrap_or("server_closed"));
     // The socket can no longer AUTH; cleanup below may block.
     conn.community_control.finish_nip_fi_line();
     let close_outcome = if state.shutting_down.load(Ordering::Acquire) {
@@ -1179,6 +1187,29 @@ async fn send_loop_inner<S>(
     }
 }
 
+/// The `close_reason` of a close the server already decided on, if any:
+/// shutdown, then a policy disconnect, then a cancellation (`auth_failed` if
+/// it rejected AUTH). The reader defers to it, so a peer's reply to the
+/// server's Close is not logged as the cause.
+pub(crate) fn server_close_reason(
+    state: &AppState,
+    conn: &ConnectionState,
+) -> Option<&'static str> {
+    if state.shutting_down.load(Ordering::Acquire) {
+        return Some("shutdown");
+    }
+    if let Some(reason) = *conn.community_control.disconnect_reason().borrow() {
+        return Some(reason.as_str());
+    }
+    conn.cancel.is_cancelled().then(|| {
+        if matches!(conn.auth_state_snapshot(), AuthState::Failed) {
+            "auth_failed"
+        } else {
+            "server_closed"
+        }
+    })
+}
+
 /// 3 missed pongs → disconnect.
 ///
 /// Sends Ping through the control channel so it isn't blocked by a full
@@ -1188,23 +1219,31 @@ async fn heartbeat_loop(
     ctrl_tx: mpsc::Sender<WsMessage>,
     missed_pongs: Arc<AtomicU8>,
     cancel: CancellationToken,
+    close: Arc<crate::access::CloseLine>,
 ) {
     let mut interval = tokio::time::interval(Duration::from_secs(30));
     loop {
         tokio::select! {
             _ = interval.tick() => {
+                // A tick racing a close already decided is not a miss.
+                if cancel.is_cancelled() {
+                    break;
+                }
                 // fetch_add returns the *previous* value before incrementing:
                 //   prev=0 → now 1 (first miss)
                 //   prev=1 → now 2 (second miss)
                 //   prev=2 → now 3 (third miss → disconnect)
                 let missed = missed_pongs.fetch_add(1, Ordering::Relaxed);
                 if missed >= 2 {
-                    warn!("3 missed pongs — closing connection");
+                    debug!("3 missed pongs — closing connection");
+                    metrics::counter!("buzz_ws_missed_pong_disconnects_total").increment(1);
+                    close.closed("missed_pongs");
                     cancel.cancel();
                     break;
                 }
                 if ctrl_tx.try_send(WsMessage::Ping(axum::body::Bytes::new())).is_err() {
                     warn!("control channel full — cannot send Ping, closing");
+                    close.closed("control_channel_full");
                     cancel.cancel();
                     break;
                 }
@@ -1225,6 +1264,7 @@ async fn recv_loop(
     state: Arc<AppState>,
     missed_pongs: Arc<AtomicU8>,
     cancel: CancellationToken,
+    close: &crate::access::CloseLine,
 ) -> (
     futures_util::stream::SplitStream<WebSocket>,
     Option<tokio::task::JoinHandle<()>>,
@@ -1247,6 +1287,7 @@ async fn recv_loop(
                                 text.len(),
                                 max_frame_bytes
                             ));
+                            close.closed("frame_too_large");
                             break;
                         }
                         trace!(len = text.len(), "frame received");
@@ -1269,6 +1310,7 @@ async fn recv_loop(
                                 bytes.len(),
                                 max_frame_bytes
                             ));
+                            close.closed("frame_too_large");
                             break;
                         }
                         // Binary frames: attempt UTF-8 decode and treat as text. Some clients
@@ -1291,15 +1333,18 @@ async fn recv_loop(
                             // Control channel full means the socket writer is
                             // completely stalled — treat as terminal.
                             warn!(conn_id = %conn.conn_id, "control channel full — cannot send Pong, closing");
+                            close.closed("control_channel_full");
                             break;
                         }
                     }
                     Some(Ok(WsMessage::Close(_))) | None => {
                         debug!("WebSocket closed by client");
+                        close.closed(server_close_reason(&state, &conn).unwrap_or("client_closed"));
                         break;
                     }
                     Some(Err(e)) => {
                         debug!("WebSocket error: {e}");
+                        close.closed(server_close_reason(&state, &conn).unwrap_or("socket_error"));
                         break;
                     }
                 }
@@ -1729,6 +1774,45 @@ pub(crate) mod tests {
         DebugValue,
     )>;
 
+    /// A connection that misses three pongs closes as `missed_pongs` and is
+    /// counted. Mutation: drop the counter or the `closed` call → RED.
+    #[test]
+    fn missed_pongs_close_with_their_reason_and_count() {
+        let recorder = metrics_util::debugging::DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let (_, out) = crate::access::tests::capture(|| {
+            metrics::with_local_recorder(&recorder, || {
+                tokio::runtime::Builder::new_current_thread()
+                    .enable_time()
+                    .start_paused(true)
+                    .build()
+                    .unwrap()
+                    .block_on(async {
+                        let close = crate::access::CloseLine::open(
+                            Uuid::nil(),
+                            buzz_core::CommunityId::from_uuid(Uuid::nil()),
+                            "127.0.0.1:1".parse().unwrap(),
+                            None,
+                        );
+                        let (ctrl_tx, _ctrl_rx) = mpsc::channel(8);
+                        let cancel = CancellationToken::new();
+                        heartbeat_loop(ctrl_tx, Arc::new(AtomicU8::new(0)), cancel.clone(), close)
+                            .await;
+                        assert!(cancel.is_cancelled());
+                    });
+            });
+        });
+        let snapshot = snapshotter.snapshot().into_vec();
+        assert_eq!(
+            counter_value(&snapshot, "buzz_ws_missed_pong_disconnects_total", None),
+            1
+        );
+        let lines = out.lines();
+        let close: Vec<_> = lines.iter().filter(|l| l["event"] == "close").collect();
+        assert_eq!(close.len(), 1, "{lines:?}");
+        assert_eq!(close[0]["close_reason"], "missed_pongs", "{lines:?}");
+    }
+
     fn counter_value(snapshot: &MetricSnapshot, name: &str, outcome: Option<&str>) -> u64 {
         snapshot
             .iter()
@@ -1867,7 +1951,12 @@ pub(crate) mod tests {
 
         crate::metrics::record_auth_attempt_started();
         let (timed_out, _rx) = test_conn_with_auth(pending_state());
-        assert!(timed_out.expire_auth());
+        assert_eq!(timed_out.expire_auth(), Some("auth_timeout"));
+        assert_eq!(
+            timed_out.expire_auth(),
+            Some("auth_failed"),
+            "a failed connection closes as auth_failed"
+        );
         assert!(timed_out
             .finish_auth_on_close(AuthOutcome::Disconnect)
             .is_none());
@@ -2035,6 +2124,149 @@ pub(crate) mod tests {
         (frames, lines)
     }
 
+    /// Serve one real active connection on `state`; once the client has its
+    /// AUTH challenge, run `server_close`, then close from the client, after
+    /// the server's own Close when `reply` is set. Returns the `close_reason`.
+    async fn close_reason_after(
+        state: Arc<AppState>,
+        reply: bool,
+        server_close: impl FnOnce(Arc<AppState>, &CommunityConnectionControl),
+    ) -> serde_json::Value {
+        use futures_util::{SinkExt as _, StreamExt as _};
+        let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4());
+        let conn_id = Uuid::new_v4();
+        let control = CommunityConnectionControl::new(CancellationToken::new());
+        let (_guard, out) = crate::access::tests::capture_scoped();
+        let (served, served_control) = (Arc::clone(&state), control.clone());
+        let app = Router::new().route(
+            "/",
+            get(move |ws: WebSocketUpgrade| {
+                let (state, control) = (Arc::clone(&served), served_control.clone());
+                async move {
+                    ws.on_upgrade(move |socket| async move {
+                        let close = crate::access::CloseLine::open(
+                            conn_id,
+                            community,
+                            "127.0.0.1:9999".parse().unwrap(),
+                            None,
+                        );
+                        handle_active_connection(
+                            socket,
+                            state,
+                            "127.0.0.1:9999".parse().unwrap(),
+                            TenantContext::resolved(community, "test.local".to_owned()),
+                            close,
+                            control,
+                            None,
+                            chrono::Utc::now(),
+                            None,
+                        )
+                        .await
+                    })
+                }
+            }),
+        );
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+        let addr = listener.local_addr().expect("local addr");
+        let server = tokio::spawn(async move { axum::serve(listener, app).await });
+        let (mut client, _) = connect_async(format!("ws://{addr}/"))
+            .await
+            .expect("connect");
+        assert!(matches!(client.next().await, Some(Ok(Message::Text(_)))));
+        while state.conn_manager.community_for_conn(conn_id).is_none() {
+            tokio::task::yield_now().await;
+        }
+        server_close(Arc::clone(&state), &control);
+        if reply {
+            while let Some(Ok(frame)) = client.next().await {
+                if frame.is_close() {
+                    break;
+                }
+            }
+        }
+        // Send the Close, or flush the reply tungstenite queued to the server's.
+        let _ = client.close(None).await;
+        let _ = client.flush().await;
+        let lines = close_lines(&out).await;
+        server.abort();
+        assert_eq!(lines.len(), 1, "{lines:?}");
+        lines[0]["close_reason"].clone()
+    }
+
+    /// A client answering the jittered restart's 1012 with its own Close
+    /// still closes as `shutdown`. Mutation: let the reader record
+    /// `client_closed` without consulting `server_close_reason` → RED.
+    #[tokio::test(flavor = "current_thread")]
+    async fn client_reply_to_a_jittered_restart_closes_as_shutdown() {
+        let state = crate::state::tests::test_state().await;
+        let reason = close_reason_after(state, true, |state, _| {
+            state.begin_shutdown();
+            tokio::spawn(async move { state.conn_manager.drain_all_jittered(1).await });
+        })
+        .await;
+        assert_eq!(reason, "shutdown");
+    }
+
+    /// A client Close that arrives as a policy disconnect cancels keeps the
+    /// policy's reason, whichever the reader sees first.
+    #[tokio::test(flavor = "current_thread")]
+    async fn client_close_racing_a_policy_disconnect_keeps_the_policy_reason() {
+        let state = crate::state::tests::test_state().await;
+        let reason = close_reason_after(state, false, |_, control| {
+            let (frame_tx, _frame_rx) = mpsc::channel(1);
+            control.deny_authorization(&frame_tx, crate::nip_fi_session::NipFiWsRoute::Root);
+        })
+        .await;
+        assert_eq!(reason, "authorization_denied");
+    }
+
+    /// The reasons the server ranks for a close it already decided on.
+    #[tokio::test(flavor = "current_thread")]
+    async fn server_close_reason_ranks_shutdown_policy_then_cancellation() {
+        let state = crate::state::tests::test_state().await;
+        let (open, _rx) = test_conn_with_auth(pending_state());
+        assert_eq!(server_close_reason(&state, &open), None);
+        open.cancel.cancel();
+        assert_eq!(server_close_reason(&state, &open), Some("server_closed"));
+        let (rejected, _rx) = test_conn_with_auth(AuthState::Failed);
+        rejected.cancel.cancel();
+        assert_eq!(server_close_reason(&state, &rejected), Some("auth_failed"));
+        let (denied, _rx) = test_conn_with_auth(AuthState::Failed);
+        denied.community_control.deny_authorization(
+            &denied.terminal_ctrl_tx,
+            crate::nip_fi_session::NipFiWsRoute::Root,
+        );
+        assert_eq!(
+            server_close_reason(&state, &denied),
+            Some("authorization_denied")
+        );
+        state.begin_shutdown();
+        assert_eq!(server_close_reason(&state, &denied), Some("shutdown"));
+    }
+
+    /// An AUTH rejected without closing leaves the socket open, and its auth
+    /// timeout then closes it as `auth_failed`.
+    #[tokio::test(flavor = "current_thread")]
+    async fn auth_timeout_after_a_rejected_auth_closes_as_auth_failed() {
+        let state = crate::state::tests::test_state().await;
+        let (conn, _rx) = test_conn_with_auth(pending_state());
+        let relay_url: RelayUrl =
+            crate::api::bridge::nip42_expected_relay_url(&state.config.relay_url, &conn.tenant)
+                .parse()
+                .expect("relay url");
+        let event = EventBuilder::auth("wrong-challenge", relay_url)
+            .sign_with_keys(&Keys::generate())
+            .expect("sign auth");
+        let frame = serde_json::json!(["AUTH", event]).to_string();
+        handle_text_message(frame, Arc::clone(&conn), state).await;
+        assert!(matches!(conn.auth_state_snapshot(), AuthState::Failed));
+        assert!(
+            !conn.cancel.is_cancelled(),
+            "a bad challenge does not close"
+        );
+        assert_eq!(conn.expire_auth(), Some("auth_failed"));
+    }
+
     /// A connection cancelled during the community-active startup check
     /// still writes its one close line, marked not admitted.
     #[tokio::test]
@@ -2134,6 +2366,7 @@ pub(crate) mod tests {
         assert_eq!(lines.len(), 1, "{lines:?}");
         assert_eq!(lines[0]["pubkey"], pubkey.to_hex(), "{}", lines[0]);
         assert!(lines[0].get("refused").is_none(), "{}", lines[0]);
+        assert_eq!(lines[0]["close_reason"], "handler_dropped", "{}", lines[0]);
     }
 
     /// Aborting the production lifecycle owner must synchronously terminalize
@@ -3171,6 +3404,7 @@ pub(crate) mod tests {
             assert_eq!(close["pubkey_verified"], true, "{close}");
             assert_eq!(close["peer_addr"], PEER, "{close}");
             assert_eq!(lines[1]["outcome"], "served", "{lines:?}");
+            assert_eq!(close["close_reason"], "client_closed", "{close}");
         }
 
         /// A refused login writes one WARN `auth` line, no success line, and a close line with no identity.
@@ -3781,6 +4015,7 @@ pub(crate) mod tests {
             let events: Vec<_> = lines.iter().map(|l| l["event"].clone()).collect();
             assert_eq!(events, ["auth", "close"], "{lines:?}");
             assert_eq!(lines[0]["outcome"], "success", "{lines:?}");
+            assert_eq!(lines[1]["close_reason"], "removed", "{lines:?}");
         }
     }
 

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

@@ -1133,6 +1133,11 @@ mod tests {
             "a login waits for a permit before it is shed"
         );
         assert!(matches!(shed.conn.auth_state_snapshot(), AuthState::Failed));
+        assert_eq!(
+            crate::connection::server_close_reason(&state, &shed.conn),
+            Some("auth_failed"),
+            "a shed login closes as auth_failed"
+        );
         assert!(
             shed.send_rx.try_recv().is_err(),
             "no OK on the data channel"

crates/buzz-relay/src/state.rs

@@ -56,6 +56,18 @@ pub(crate) enum CommunityDisconnectReason {
 }
 
 impl CommunityDisconnectReason {
+    /// The access log `close_reason` of a connection this reason closed.
+    pub(crate) fn as_str(self) -> &'static str {
+        match self {
+            Self::CommunityArchived => "community_archived",
+            Self::CommunityDeleted => "community_deleted",
+            Self::AuthPolicyUpgraded => "auth_policy_upgraded",
+            Self::AuthorizationDenied => "authorization_denied",
+            Self::AccessRevoked => "access_revoked",
+            Self::Removed => "removed",
+        }
+    }
+
     pub(crate) fn close_message(self) -> WsMessage {
         match self {
             Self::CommunityArchived => WsMessage::Close(Some(axum::extract::ws::CloseFrame {

scripts/run-tests.sh

@@ -326,6 +326,11 @@ run_unit_tests() {
       api::bridge::postgres_tests::moderation_read_audit_query_bearing_nip98_event_verifies \
       api::bridge::postgres_tests::moderation_read_restricted_no_query_still_verifies \
       connection::tests::aborting_lifecycle_owner_is_drop_safe \
+      connection::tests::missed_pongs_close_with_their_reason_and_count \
+      connection::tests::client_reply_to_a_jittered_restart_closes_as_shutdown \
+      connection::tests::client_close_racing_a_policy_disconnect_keeps_the_policy_reason \
+      connection::tests::server_close_reason_ranks_shutdown_policy_then_cancellation \
+      connection::tests::auth_timeout_after_a_rejected_auth_closes_as_auth_failed \
       connection::tests::shutdown_terminalizes_auth_stalled_on_database_before_late_success \
       connection::terminal_close_delivery_tests::upgrade_would_deny_line_without_auth_logs_absent_at_close \
       handlers::event::tests::observer_owner_cache_is_scoped_to_community \