diff --git a/Justfile b/Justfile index b7914bfb8a..16146bd7cd 100644 --- a/Justfile +++ b/Justfile @@ -560,7 +560,7 @@ test-unit: # cross-node fan-out loss handling; its stalled-delivery case fakes a # database that never answers on loopback, and none use DATABASE_URL. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::media_access::tests::/) + test(/^handlers::media_access::tests::/) + test(/^subscription::tests::/) + test(/^handlers::multinode_fanout::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::dispatcher_counts_each_req_and_count_frame_once) + test(=handlers::req::tests::unqueued_eose_logs_aborted_not_served) + test(=api::bridge::postgres_tests::reject_code_keeps_only_a_known_prefix) + test(=api::bridge::postgres_tests::nip98_replay_check_fails_closed_when_guard_errors) + test(=api::bridge::postgres_tests::verify_bridge_auth_rejects_nip98_event_signed_for_wrong_communitys_host) + test(=api::bridge::postgres_tests::verify_bridge_auth_can_require_payload_tag_for_json_body_endpoints) + test(=api::bridge::postgres_tests::verify_bridge_auth_accepts_nip98_event_signed_for_matching_host) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_verifies_with_matching_query) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_rejected_against_bare_path) + test(=api::bridge::postgres_tests::moderation_read_audit_query_bearing_nip98_event_verifies) + test(=api::bridge::postgres_tests::moderation_read_restricted_no_query_still_verifies) + test(=connection::tests::aborting_lifecycle_owner_is_drop_safe) + test(=connection::tests::shutdown_terminalizes_auth_stalled_on_database_before_late_success) + test(=connection::terminal_close_delivery_tests::upgrade_would_deny_line_without_auth_logs_absent_at_close) + test(=handlers::event::tests::observer_owner_cache_is_scoped_to_community) + test(=state::tests::shared_counter_between_direct_and_fanout) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::saturated_handler_write_line_records_whether_its_ok_was_queued) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(=api::bridge::moderation_refusal_tests::moderation_lookup_failure_is_unavailable) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_policy::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::db_read_error_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^api::git::store::tests::/) + test(/^api::git::store::fault_probe::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::failed_search_closes_without_eose_or_trying_later_filters) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::count_fallback_fetches_one_extra_candidate) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^api::media_access::tests::/) + test(/^handlers::media_access::tests::/) + test(/^subscription::tests::/) + test(/^handlers::multinode_fanout::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::dispatcher_counts_each_req_and_count_frame_once) + test(=handlers::req::tests::unqueued_eose_logs_aborted_not_served) + test(=api::bridge::postgres_tests::reject_code_keeps_only_a_known_prefix) + test(=api::bridge::postgres_tests::nip98_replay_check_fails_closed_when_guard_errors) + test(=api::bridge::postgres_tests::verify_bridge_auth_rejects_nip98_event_signed_for_wrong_communitys_host) + test(=api::bridge::postgres_tests::verify_bridge_auth_can_require_payload_tag_for_json_body_endpoints) + test(=api::bridge::postgres_tests::verify_bridge_auth_accepts_nip98_event_signed_for_matching_host) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_verifies_with_matching_query) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_rejected_against_bare_path) + test(=api::bridge::postgres_tests::moderation_read_audit_query_bearing_nip98_event_verifies) + test(=api::bridge::postgres_tests::moderation_read_restricted_no_query_still_verifies) + test(=connection::tests::aborting_lifecycle_owner_is_drop_safe) + test(=connection::tests::shutdown_terminalizes_auth_stalled_on_database_before_late_success) + test(=connection::terminal_close_delivery_tests::upgrade_would_deny_line_without_auth_logs_absent_at_close) + test(=handlers::event::tests::observer_owner_cache_is_scoped_to_community) + test(=state::tests::shared_counter_between_direct_and_fanout) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::missed_pongs_close_with_their_reason_and_count) + test(=connection::tests::client_reply_to_a_jittered_restart_closes_as_shutdown) + test(=connection::tests::client_close_racing_a_policy_disconnect_keeps_the_policy_reason) + test(=connection::tests::server_close_reason_ranks_shutdown_policy_then_cancellation) + test(=connection::tests::auth_timeout_after_a_rejected_auth_closes_as_auth_failed) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::saturated_handler_write_line_records_whether_its_ok_was_queued) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(=api::bridge::moderation_refusal_tests::moderation_lookup_failure_is_unavailable) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_policy::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::db_read_error_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^api::git::store::tests::/) + test(/^api::git::store::fault_probe::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::failed_search_closes_without_eose_or_trying_later_filters) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::count_fallback_fetches_one_extra_candidate) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' # Note on audio::join::tests scope: the full suite is infra-free (no DB, # no Redis). The infra-free audio/FI regression witnesses — bootstrap # ordering barrier, CommitConfirmed arm, pending-close invisibility, diff --git a/crates/buzz-relay/src/access.rs b/crates/buzz-relay/src/access.rs index 03baa420c0..93c3fc0896 100644 --- a/crates/buzz-relay/src/access.rs +++ b/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>, refused: std::sync::OnceLock<&'static str>, + close_reason: std::sync::OnceLock<&'static str>, pub(crate) queued: std::sync::Arc, pub(crate) refusals: std::sync::Arc, } @@ -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] diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 8b32bb4823..ace9792e1a 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/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( } } +/// 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, missed_pongs: Arc, cancel: CancellationToken, + close: Arc, ) { 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, missed_pongs: Arc, cancel: CancellationToken, + close: &crate::access::CloseLine, ) -> ( futures_util::stream::SplitStream, Option>, @@ -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, + reply: bool, + server_close: impl FnOnce(Arc, &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:?}"); } } diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 4e73e9f5f9..ecbc7d9e0d 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/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" diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index d841a0c845..c6c5bf43ca 100644 --- a/crates/buzz-relay/src/state.rs +++ b/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 { diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index b49a0ca444..8c93e9dd2d 100755 --- a/scripts/run-tests.sh +++ b/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 \