diff --git a/crates/client-api/src/routes/subscribe.rs b/crates/client-api/src/routes/subscribe.rs index 2e4b1bd62ad..d5cdf15235b 100644 --- a/crates/client-api/src/routes/subscribe.rs +++ b/crates/client-api/src/routes/subscribe.rs @@ -4,7 +4,7 @@ use std::num::NonZeroUsize; use std::panic; use std::pin::{pin, Pin}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; use std::time::Duration; @@ -34,7 +34,9 @@ use spacetimedb::host::module_host::ClientConnectedError; use spacetimedb::host::NoSuchModule; use spacetimedb::subscription::row_list_builder_pool::BsatnRowListBuilderPool; use spacetimedb::util::spawn_rayon; -use spacetimedb::worker_metrics::WORKER_METRICS; +use spacetimedb::worker_metrics::{ + record_client_rejection, ClientDisconnectCause, ClientDisconnectRecorder, ClientRejectCause, WORKER_METRICS, +}; use spacetimedb::Identity; use spacetimedb_client_api_messages::websocket::v1 as ws_v1; use spacetimedb_client_api_messages::websocket::v2 as ws_v2; @@ -145,7 +147,13 @@ where } let db_identity = name_or_identity.resolve(&ctx).await?; - let sql_auth = ctx.authorize_sql(auth.claims.identity, db_identity).await?; + let sql_auth = match ctx.authorize_sql(auth.claims.identity, db_identity).await { + Ok(sql_auth) => sql_auth, + Err(err) => { + record_client_rejection(db_identity, ClientRejectCause::AuthorizationFailed); + return Err(err.into()); + } + }; #[derive(Clone, Copy)] struct NegotiatedProtocol { @@ -184,7 +192,13 @@ where ), ]); - let negotiated = protocol.ok_or((StatusCode::BAD_REQUEST, "no valid protocol selected"))?; + let negotiated = match protocol { + Some(protocol) => protocol, + None => { + record_client_rejection(db_identity, ClientRejectCause::InvalidProtocol); + Err((StatusCode::BAD_REQUEST, "no valid protocol selected"))? + } + }; let client_config = ClientConfig { protocol: negotiated.protocol, version: negotiated.version, @@ -225,6 +239,7 @@ where let ws = match ws_upgrade.upgrade(ws_config).await { Ok(ws) => ws, Err(err) => { + record_client_rejection(db_identity, ClientRejectCause::WebsocketUpgradeError); log::error!("websocket: WebSocket init error: {err}"); return; } @@ -251,14 +266,24 @@ where log::debug!("websocket: client_connected returned Ok for {client_log_string}"); connected } - Err(e @ (ClientConnectedError::Rejected(_) | ClientConnectedError::OutOfEnergy)) => { - log::info!( - "websocket: Rejecting connection for {client_log_string} due to error from client_connected reducer: {e}" - ); - return; - } - Err(e @ (ClientConnectedError::DBError(_) | ClientConnectedError::ReducerCall(_))) => { - log::warn!("websocket: ModuleHost died while {client_log_string} was connecting: {e:#}"); + Err(e) => { + let cause = match &e { + ClientConnectedError::Rejected(_) => { + log::info!("websocket: Rejecting connection for {client_log_string} due to rejection from client_connected reducer: {e}"); + ClientRejectCause::ClientConnectedRejected + } + ClientConnectedError::OutOfEnergy => { + log::info!("websocket: Rejecting connection for {client_log_string} due to out of energy error from client_connected reducer: {e}"); + ClientRejectCause::OutOfEnergy + } + ClientConnectedError::DBError(_) | ClientConnectedError::ReducerCall(_) => { + log::warn!( + "websocket: Rejecting connection for {client_log_string} due to error running client_connected reducer: {e:#}" + ); + ClientRejectCause::ClientConnectedError + } + }; + record_client_rejection(db_identity, cause); return; } }; @@ -315,18 +340,29 @@ struct ActorState { pub client_id: ClientActorId, pub database: Identity, config: WebSocketOptions, + disconnect_recorder: Option, closed: AtomicBool, got_pong: AtomicBool, + /// When the last `Ping` frame was written to the socket. + /// Taken when the corresponding `Pong` arrives, to observe the roundtrip time. + last_ping_sent: Mutex>, } impl ActorState { - pub fn new(database: Identity, client_id: ClientActorId, config: WebSocketOptions) -> Self { + pub fn new( + database: Identity, + client_id: ClientActorId, + config: WebSocketOptions, + disconnect_recorder: Option, + ) -> Self { Self { database, client_id, config, + disconnect_recorder, closed: AtomicBool::new(false), got_pong: AtomicBool::new(true), + last_ping_sent: Mutex::new(None), } } @@ -346,9 +382,34 @@ impl ActorState { self.got_pong.swap(false, Ordering::Relaxed) } + /// Record that a `Ping` frame was written to the socket. + fn record_ping_sent(&self) { + *self.last_ping_sent.lock().unwrap() = Some(Instant::now()); + } + + /// Record that the client's `Pong` was processed, + /// observing the ping-pong roundtrip time. + /// + /// This time includes any queueing of the `Ping` behind other outgoing + /// data in the TCP stream. + fn record_pong(&self) { + if let Some(sent) = self.last_ping_sent.lock().unwrap().take() { + WORKER_METRICS + .websocket_pong_rtt + .with_label_values(&self.database) + .observe(sent.elapsed().as_secs_f64()); + } + } + pub fn next_idle_deadline(&self) -> Instant { Instant::now() + self.config.idle_timeout } + + pub fn record_disconnect(&self, cause: ClientDisconnectCause) -> bool { + self.disconnect_recorder + .as_ref() + .is_some_and(|recorder| recorder.record(cause)) + } } /// Configuration for WebSocket connections. @@ -452,7 +513,12 @@ async fn ws_client_actor_inner( let database = client.module().info().database_identity; let client_id = client.id; let client_closed_metric = WORKER_METRICS.ws_clients_closed_connection.with_label_values(&database); - let state = Arc::new(ActorState::new(database, client_id, config)); + let state = Arc::new(ActorState::new( + database, + client_id, + config, + client.disconnect_recorder(), + )); // Channel for [`UnorderedWsMessage`]s. let (unordered_tx, unordered_rx) = mpsc::unbounded_channel(); @@ -673,6 +739,7 @@ async fn ws_main_loop( .ws_clients_idle_timed_out .with_label_values(&state.database) .inc(); + state.record_disconnect(ClientDisconnectCause::IdleTimeout); break; }, @@ -682,6 +749,7 @@ async fn ws_main_loop( // Branch is disabled if we already sent a close frame. res = &mut watch_hotswap, if !closed => { if let Err(NoSuchModule) = res { + state.record_disconnect(ClientDisconnectCause::ModuleExited); let close = CloseFrame { code: CloseCode::Away, reason: "module exited".into() @@ -795,6 +863,7 @@ async fn ws_recv_task( continue; } log::debug!("Client caused error: {e}"); + state.record_disconnect(ClientDisconnectCause::ClientMessageError); let close = CloseFrame { code: CloseCode::Error, reason: format!("{e:#}").into(), @@ -861,6 +930,7 @@ fn ws_recv_loop( loop { let Some(res) = next_message(&state, &mut ws).await else { log::trace!("recv stream exhausted"); + state.record_disconnect(ClientDisconnectCause::WebsocketStreamEnded); break; }; match res { @@ -894,6 +964,7 @@ fn ws_recv_loop( | WsError::Url(_) | WsError::Http(_) | WsError::HttpFormat(_)) => { + state.record_disconnect(ClientDisconnectCause::WebsocketReceiveError); log::warn!("Websocket receive error: {e}"); break; } @@ -946,6 +1017,7 @@ fn ws_recv_queue( mpsc::error::TrySendError::Full(item) => { let client_id = state.client_id; log::warn!("Client {client_id} exceeded incoming_queue_length limit of {max_incoming_queue_length} requests"); + state.record_disconnect(ClientDisconnectCause::IncomingQueueFull); // If we can't send close (send task already terminated): // // - Let downstream handlers know that we're closing, @@ -1036,6 +1108,7 @@ fn ws_client_message_handler( ClientMessage::Pong(_bytes) => { log::trace!("Received pong from client {}", state.client_id); state.set_ponged(); + state.record_pong(); }, ClientMessage::Close(close_frame) => { log::trace!("Received Close frame from client {}: {:?}", state.client_id, close_frame); @@ -1043,6 +1116,7 @@ fn ws_client_message_handler( // This is the client telling us they want to close. if !was_closed { client_closed_metric.inc(); + state.record_disconnect(ClientDisconnectCause::ClientClose); } } } @@ -1198,6 +1272,7 @@ async fn ws_send_loop_inner( while let Ok(frame) = frames_rx.try_recv() { let eof = frame.header().is_final; if let Err(e) = ws.feed(WsMessage::Frame(frame)).await { + state.record_disconnect(ClientDisconnectCause::WebsocketSendError); log::warn!("error sending frame: {e:#}"); break 'outer; } @@ -1209,6 +1284,7 @@ async fn ws_send_loop_inner( // Then send the close frame. log::trace!("sending close frame"); if let Err(e) = ws.send(WsMessage::Close(Some(close_frame))).await { + state.record_disconnect(ClientDisconnectCause::WebsocketSendError); log::warn!("error sending close frame: {e:#}"); break; } @@ -1230,9 +1306,11 @@ async fn ws_send_loop_inner( UnorderedWsMessage::Ping(bytes) => { log::trace!("sending ping"); if let Err(e) = ws.feed(WsMessage::Ping(bytes)).await { + state.record_disconnect(ClientDisconnectCause::WebsocketSendError); log::warn!("error sending ping: {e:#}"); break; } + state.record_ping_sent(); }, UnorderedWsMessage::Error(err) => { log::trace!("encoding execution error"); @@ -1259,6 +1337,7 @@ async fn ws_send_loop_inner( log::trace!("sending batch of {n} frames"); for frame in frames_batch.drain(..n) { if let Err(e) = ws.feed(WsMessage::Frame(frame)).await { + state.record_disconnect(ClientDisconnectCause::WebsocketSendError); log::warn!("error sending frame: {e:#}"); break 'outer; } @@ -1282,6 +1361,7 @@ async fn ws_send_loop_inner( } if let Err(e) = ws.flush().await { + state.record_disconnect(ClientDisconnectCause::WebsocketSendError); log::warn!("error flushing websocket: {e}"); break; } @@ -1955,7 +2035,56 @@ mod tests { } fn dummy_actor_state_with_config(config: WebSocketOptions) -> ActorState { - ActorState::new(Identity::ZERO, dummy_client_id(), config) + ActorState::new(Identity::ZERO, dummy_client_id(), config, None) + } + + fn test_database(byte: u8) -> Identity { + let mut bytes = [0; 32]; + bytes[31] = byte; + Identity::from_be_byte_array(bytes) + } + + fn actor_state_with_disconnect_recorder(byte: u8, config: WebSocketOptions) -> ActorState { + let database = test_database(byte); + ActorState::new( + database, + dummy_client_id(), + config, + Some(ClientDisconnectRecorder::new(database)), + ) + } + + fn disconnect_count(database: Identity, cause: ClientDisconnectCause) -> u64 { + WORKER_METRICS + .ws_client_disconnections + .with_label_values(&database, cause.as_str()) + .get() + } + + fn rejection_count(database: Identity, cause: ClientRejectCause) -> u64 { + WORKER_METRICS + .ws_client_rejections + .with_label_values(&database, cause.as_str()) + .get() + } + + fn assert_disconnect_count_incremented(database: Identity, cause: ClientDisconnectCause, before: u64) { + assert_eq!(disconnect_count(database, cause), before + 1); + } + + fn assert_rejection_count_incremented(database: Identity, cause: ClientRejectCause, before: u64) { + assert_eq!(rejection_count(database, cause), before + 1); + } + + #[test] + fn record_client_rejection_increments_rejection_metric() { + let database = test_database(100); + + for cause in ClientRejectCause::ALL { + let before = rejection_count(database, cause); + record_client_rejection(database, cause); + assert_rejection_count_incremented(database, cause, before); + } } #[tokio::test(start_paused = true)] // see [NOTE: start_paused] @@ -1982,22 +2111,25 @@ mod tests { #[tokio::test] async fn recv_loop_terminates_when_input_exhausted() { - let state = Arc::new(dummy_actor_state()); + let state = Arc::new(actor_state_with_disconnect_recorder(1, <_>::default())); + let before = disconnect_count(state.database, ClientDisconnectCause::WebsocketStreamEnded); let (idle_tx, _idle_rx) = watch::channel(Instant::now() + state.config.idle_timeout); let input = stream::iter(vec![Ok(WsMessage::Ping(Bytes::new()))]); pin_mut!(input); - let recv_loop = ws_recv_loop(state, idle_tx, input); + let recv_loop = ws_recv_loop(state.clone(), idle_tx, input); pin_mut!(recv_loop); assert_matches!(recv_loop.next().await, Some(ClientMessage::Ping(_))); assert_matches!(recv_loop.next().await, None); + assert_disconnect_count_incremented(state.database, ClientDisconnectCause::WebsocketStreamEnded, before); } #[tokio::test] async fn recv_loop_terminates_when_input_yields_err() { - let state = Arc::new(dummy_actor_state()); + let state = Arc::new(actor_state_with_disconnect_recorder(2, <_>::default())); + let before = disconnect_count(state.database, ClientDisconnectCause::WebsocketReceiveError); let (idle_tx, _idle_rx) = watch::channel(Instant::now() + state.config.idle_timeout); let input = stream::iter(vec![ @@ -2007,11 +2139,12 @@ mod tests { ]); pin_mut!(input); - let recv_loop = ws_recv_loop(state, idle_tx, input); + let recv_loop = ws_recv_loop(state.clone(), idle_tx, input); pin_mut!(recv_loop); assert_matches!(recv_loop.next().await, Some(ClientMessage::Ping(_))); assert_matches!(recv_loop.next().await, None); + assert_disconnect_count_incremented(state.database, ClientDisconnectCause::WebsocketReceiveError, before); } #[tokio::test] @@ -2098,7 +2231,8 @@ mod tests { #[tokio::test] async fn client_message_handler_updates_pong_and_closed_states_and_metric() { - let state = Arc::new(dummy_actor_state()); + let state = Arc::new(actor_state_with_disconnect_recorder(3, <_>::default())); + let before = disconnect_count(state.database, ClientDisconnectCause::ClientClose); state.reset_ponged(); let metric = IntGauge::new("bleep", "unhelpful").unwrap(); @@ -2109,6 +2243,31 @@ mod tests { assert!(state.closed()); assert!(state.reset_ponged()); assert_eq!(metric.get(), 1); + assert_disconnect_count_incremented(state.database, ClientDisconnectCause::ClientClose, before); + } + + #[tokio::test] + async fn recv_task_records_client_message_error_disconnect() { + let state = Arc::new(actor_state_with_disconnect_recorder(7, <_>::default())); + let before = disconnect_count(state.database, ClientDisconnectCause::ClientMessageError); + let (idle_tx, _idle_rx) = watch::channel(state.next_idle_deadline()); + let metric = IntGauge::new("bleep", "unhelpful").unwrap(); + let (unordered_tx, mut unordered_rx) = mpsc::unbounded_channel(); + let input = stream::iter([Ok(WsMessage::text("not useful"))]); + + ws_recv_task( + state.clone(), + idle_tx, + metric, + |_data, _timer| future::ready(Err(MessageHandleError::UnsupportedVersion("test"))), + unordered_tx, + input, + WsVersion::V2, + ) + .await; + + assert_matches!(unordered_rx.recv().await, Some(UnorderedWsMessage::Close(_))); + assert_disconnect_count_incremented(state.database, ClientDisconnectCause::ClientMessageError, before); } #[tokio::test] @@ -2189,8 +2348,9 @@ mod tests { ))), ]; - for message in input { - let state = Arc::new(dummy_actor_state()); + for (i, message) in input.into_iter().enumerate() { + let state = Arc::new(actor_state_with_disconnect_recorder(20 + i as u8, <_>::default())); + let before = disconnect_count(state.database, ClientDisconnectCause::WebsocketSendError); let (messages_tx, messages_rx) = mpsc::channel(64); let (unordered_tx, unordered_rx) = mpsc::unbounded_channel(); @@ -2209,6 +2369,7 @@ mod tests { Either::Right(message) => messages_tx.send(message).await.unwrap(), } send_loop.await; + assert_disconnect_count_incremented(state.database, ClientDisconnectCause::WebsocketSendError, before); } } @@ -2238,8 +2399,9 @@ mod tests { ))), ]; - for message in input { - let state = Arc::new(dummy_actor_state()); + for (i, message) in input.into_iter().enumerate() { + let state = Arc::new(actor_state_with_disconnect_recorder(30 + i as u8, <_>::default())); + let before = disconnect_count(state.database, ClientDisconnectCause::WebsocketSendError); let (messages_tx, messages_rx) = mpsc::channel(64); let (unordered_tx, unordered_rx) = mpsc::unbounded_channel(); @@ -2258,6 +2420,7 @@ mod tests { Either::Right(message) => messages_tx.send(message).await.unwrap(), } send_loop.await; + assert_disconnect_count_incremented(state.database, ClientDisconnectCause::WebsocketSendError, before); } } @@ -2286,10 +2449,14 @@ mod tests { #[tokio::test(start_paused = true)] // see [NOTE: start_paused] async fn main_loop_terminates_on_idle_timeout() { - let state = Arc::new(dummy_actor_state_with_config(WebSocketOptions { - idle_timeout: Duration::from_millis(10), - ..<_>::default() - })); + let state = Arc::new(actor_state_with_disconnect_recorder( + 4, + WebSocketOptions { + idle_timeout: Duration::from_millis(10), + ..<_>::default() + }, + )); + let before = disconnect_count(state.database, ClientDisconnectCause::IdleTimeout); let (idle_tx, idle_rx) = watch::channel(state.next_idle_deadline()); let start = Instant::now(); @@ -2320,6 +2487,7 @@ mod tests { let elapsed = start.elapsed(); assert!(elapsed >= timeout); assert!(elapsed < timeout + Duration::from_millis(10)); + assert_disconnect_count_incremented(state.database, ClientDisconnectCause::IdleTimeout, before); } #[tokio::test(start_paused = true)] // see [NOTE: start_paused] @@ -2380,7 +2548,8 @@ mod tests { #[tokio::test(start_paused = true)] // see [NOTE: start_paused] async fn main_loop_terminates_when_module_exits() { - let state = Arc::new(dummy_actor_state()); + let state = Arc::new(actor_state_with_disconnect_recorder(5, <_>::default())); + let before = disconnect_count(state.database, ClientDisconnectCause::ModuleExited); let (_idle_tx, idle_rx) = watch::channel(state.next_idle_deadline()); let unordered_tx = { @@ -2393,6 +2562,7 @@ mod tests { }; let start = tokio::time::Instant::now(); + let state_for_loop = state.clone(); tokio::spawn(async move { let hotswap = || async { sleep(Duration::from_millis(5)).await; @@ -2400,13 +2570,13 @@ mod tests { }; ws_main_loop( - state.clone(), + state_for_loop.clone(), hotswap, ws_idle_timer(idle_rx), // Pretend we received a close immediately after sending one. tokio::spawn(async move { loop { - if state.closed() { + if state_for_loop.closed() { break; } sleep(Duration::from_millis(1)).await @@ -2429,20 +2599,25 @@ mod tests { elapsed < Duration::from_millis(10), "main loop should shut down shortly after module is shut down" ); + assert_disconnect_count_incremented(state.database, ClientDisconnectCause::ModuleExited, before); } #[tokio::test] async fn recv_queue_sends_close_when_at_capacity() { - let state = Arc::new(dummy_actor_state_with_config(WebSocketOptions { - incoming_queue_length: 10.try_into().unwrap(), - ..<_>::default() - })); + let state = Arc::new(actor_state_with_disconnect_recorder( + 6, + WebSocketOptions { + incoming_queue_length: 10.try_into().unwrap(), + ..<_>::default() + }, + )); + let before = disconnect_count(state.database, ClientDisconnectCause::IncomingQueueFull); let (unordered_tx, mut unordered_rx) = mpsc::unbounded_channel(); let input = stream::iter((0..20).map(|i| Ok(WsMessage::text(format!("message {i}"))))); let metric = IntGauge::new("bleep", "unhelpful").unwrap(); - let received = ws_recv_queue(state, unordered_tx, metric.clone(), input) + let received = ws_recv_queue(state.clone(), unordered_tx, metric.clone(), input) .collect::>() .await; @@ -2451,6 +2626,7 @@ mod tests { assert_eq!(metric.get(), 0); // Should have received all of the input. assert_eq!(received.len(), 20); + assert_disconnect_count_incremented(state.database, ClientDisconnectCause::IncomingQueueFull, before); } #[tokio::test] diff --git a/crates/core/src/client/client_connection.rs b/crates/core/src/client/client_connection.rs index 85c7d97a620..f33e25a8fc6 100644 --- a/crates/core/src/client/client_connection.rs +++ b/crates/core/src/client/client_connection.rs @@ -14,7 +14,9 @@ use crate::host::module_host::{ClientConnectedError, ProcedureResultTarget}; use crate::host::{FunctionArgs, ModuleHost, NoSuchModule, ReducerCallError}; use crate::subscription::module_subscription_manager::BroadcastError; use crate::util::prometheus_handle::IntGaugeExt; -use crate::worker_metrics::WORKER_METRICS; +use crate::worker_metrics::{ + record_client_rejection, ClientDisconnectCause, ClientDisconnectRecorder, ClientRejectCause, WORKER_METRICS, +}; use bytes::Bytes; use bytestring::ByteString; use derive_more::From; @@ -314,6 +316,7 @@ pub struct ClientConnectionMetrics { pub websocket_request_msg_size: Histogram, pub websocket_requests: IntCounter, pub outgoing_queue_disconnects: IntCounter, + pub disconnect_recorder: ClientDisconnectRecorder, /// The `total_outgoing_queue_length` metric labeled with this database's `Identity`, /// which we'll increment whenever sending a message. @@ -340,11 +343,13 @@ impl ClientConnectionMetrics { let sendtx_queue_size = WORKER_METRICS .total_outgoing_queue_length .with_label_values(&database_identity); + let disconnect_recorder = ClientDisconnectRecorder::new(database_identity); Self { websocket_request_msg_size, websocket_requests, outgoing_queue_disconnects, + disconnect_recorder, sendtx_queue_size, } } @@ -449,6 +454,9 @@ impl ClientConnectionSender { ); if let Some(metrics) = &self.metrics { metrics.outgoing_queue_disconnects.inc(); + metrics + .disconnect_recorder + .record(ClientDisconnectCause::OutgoingQueueFull); } self.abort_handle.abort(); self.cancelled.store(true, Ordering::Relaxed); @@ -866,22 +874,30 @@ impl ClientConnection { let module_info = module.info.clone(); let database_identity = module_info.database_identity; let client_identity = id.identity; + let metrics = ClientConnectionMetrics::new(database_identity, config.protocol); + let actor_disconnect_recorder = metrics.disconnect_recorder.clone(); let abort_handle = tokio::spawn(async move { - let Ok(fut) = fut_rx.await else { return }; + let Ok(fut) = fut_rx.await else { + log::warn!("websocket connection aborted for client identity `{client_identity}` and database identity `{database_identity}` before actor startup"); + record_client_rejection(database_identity, ClientRejectCause::ActorStartupFailure); + return }; let _gauge_guard = module_info.metrics.connected_clients.inc_scope(); module_info.metrics.ws_clients_spawned.inc(); scopeguard::defer! { let database_identity = module_info.database_identity; - log::warn!("websocket connection aborted for client identity `{client_identity}` and database identity `{database_identity}`"); module_info.metrics.ws_clients_aborted.inc(); + // This is always called for to make sure `ws_clients_aborted` is incremented, but we only want to log a warning here + // if we haven't already recorded a cause for this disconnection. + if actor_disconnect_recorder.record(ClientDisconnectCause::Unknown) { + log::warn!("websocket connection aborted for client identity `{client_identity}` and database identity `{database_identity}`"); + } }; fut.await }) .abort_handle(); - let metrics = ClientConnectionMetrics::new(database_identity, config.protocol); let receiver = ClientConnectionReceiver::new( config.confirmed_reads, MeteredReceiver::with_gauge(sendrx, metrics.sendtx_queue_size.clone()), @@ -943,6 +959,13 @@ impl ClientConnection { self.sender.clone() } + pub fn disconnect_recorder(&self) -> Option { + self.sender + .metrics + .as_ref() + .map(|metrics| metrics.disconnect_recorder.clone()) + } + /// Get the [`ModuleHost`] for this connection. /// /// Note that modules can be hotswapped, in which case the returned handle @@ -1477,4 +1500,48 @@ mod tests { offset.mark_durable_at(3); assert_received_update(receiver.recv()).await; } + + #[tokio::test] + async fn client_connection_sender_records_outgoing_queue_full_disconnect() { + let database_identity = Identity::from_be_byte_array([7; 32]); + let cause = ClientDisconnectCause::OutgoingQueueFull; + let before = WORKER_METRICS + .ws_client_disconnections + .with_label_values(&database_identity, cause.as_str()) + .get(); + + let (sendtx, _rx) = mpsc::channel(1); + let abort_handle = tokio::spawn(std::future::pending::<()>()).abort_handle(); + let sender = ClientConnectionSender { + id: ClientActorId::for_test(Identity::ZERO), + auth: ConnectionAuthCtx::try_from(SpacetimeIdentityClaims { + identity: Identity::ZERO, + subject: "".into(), + issuer: "".into(), + audience: [].into(), + iat: SystemTime::now(), + exp: None, + extra: None, + }) + .unwrap(), + config: ClientConfig::for_test(), + sendtx, + abort_handle, + cancelled: AtomicBool::new(false), + metrics: Some(ClientConnectionMetrics::new(database_identity, Protocol::Binary)), + }; + + sender.send_message(None, empty_tx_update()).unwrap(); + assert_matches!( + sender.send_message(None, empty_tx_update()), + Err(ClientSendError::Cancelled) + ); + assert_eq!( + WORKER_METRICS + .ws_client_disconnections + .with_label_values(&database_identity, cause.as_str()) + .get(), + before + 1 + ); + } } diff --git a/crates/core/src/worker_metrics/mod.rs b/crates/core/src/worker_metrics/mod.rs index e99bd196b63..c4cfb407187 100644 --- a/crates/core/src/worker_metrics/mod.rs +++ b/crates/core/src/worker_metrics/mod.rs @@ -8,12 +8,163 @@ use spacetimedb_lib::Identity; use spacetimedb_metrics::metrics_group; use spacetimedb_sats::memory_usage::MemoryUsage; use spacetimedb_table::page_pool::PagePool; -use std::{sync::Once, time::Duration}; +use std::{ + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, Once, + }, + time::Duration, +}; use tokio::{spawn, time::sleep}; // Used as a metrics label value, so private billing code needs access to it. pub use crate::host::v8::JsWorkerKind; +/// Causes for websocket client connection endings. +/// +/// These values are metric labels, so changes may require changes to dashboards and alerting rules. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)] +pub enum ClientDisconnectCause { + /// The client sent a websocket close frame. + ClientClose, + /// The accepted client stopped sending data, including keepalive pongs, until the idle timeout elapsed. + IdleTimeout, + /// The accepted client sent messages faster than the server could process and filled its incoming queue. + IncomingQueueFull, + /// The accepted client could not receive server messages quickly enough and filled its outgoing queue. + OutgoingQueueFull, + /// The database module exited and the server closed the websocket. + ModuleExited, + /// The accepted client sent a syntactically or semantically invalid websocket request. + ClientMessageError, + /// The websocket receive stream returned an error. + WebsocketReceiveError, + /// The server failed while sending or flushing websocket data. + WebsocketSendError, + /// The websocket receive stream ended without a more specific cause. + WebsocketStreamEnded, + /// The accepted websocket actor ended without a more specific recorded cause. + Unknown, +} + +impl ClientDisconnectCause { + pub const ALL: [Self; 10] = [ + Self::ClientClose, + Self::IdleTimeout, + Self::IncomingQueueFull, + Self::OutgoingQueueFull, + Self::ModuleExited, + Self::ClientMessageError, + Self::WebsocketReceiveError, + Self::WebsocketSendError, + Self::WebsocketStreamEnded, + Self::Unknown, + ]; + + pub const fn as_str(self) -> &'static str { + match self { + Self::ClientClose => "client_close", + Self::IdleTimeout => "idle_timeout", + Self::IncomingQueueFull => "incoming_queue_full", + Self::OutgoingQueueFull => "outgoing_queue_full", + Self::ModuleExited => "module_exited", + Self::ClientMessageError => "client_message_error", + Self::WebsocketReceiveError => "websocket_receive_error", + Self::WebsocketSendError => "websocket_send_error", + Self::WebsocketStreamEnded => "websocket_stream_ended", + Self::Unknown => "unknown", + } + } +} + +/// Causes for websocket client connection attempts rejected before being counted as connected. +/// +/// These values are metric labels, so changes may require changes to dashboards and alerting rules. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)] +pub enum ClientRejectCause { + /// The subscribe request did not negotiate a supported websocket protocol. + InvalidProtocol, + /// SQL authorization failed after the request was associated with a database. + AuthorizationFailed, + /// The HTTP upgrade to a websocket failed after the request was associated with a database. + WebsocketUpgradeError, + /// The module's client-connected lifecycle reducer rejected the connection. + ClientConnectedRejected, + /// The module rejected the connection because the caller was out of energy. + OutOfEnergy, + /// The client-connected lifecycle reducer or module host failed while accepting the connection. + ClientConnectedError, + /// The client actor function failed before completing setup. + ActorStartupFailure, +} + +impl ClientRejectCause { + pub const ALL: [Self; 7] = [ + Self::InvalidProtocol, + Self::AuthorizationFailed, + Self::WebsocketUpgradeError, + Self::ClientConnectedRejected, + Self::OutOfEnergy, + Self::ClientConnectedError, + Self::ActorStartupFailure, + ]; + + pub const fn as_str(self) -> &'static str { + match self { + Self::InvalidProtocol => "invalid_protocol", + Self::AuthorizationFailed => "authorization_failed", + Self::WebsocketUpgradeError => "websocket_upgrade_error", + Self::ClientConnectedRejected => "client_connected_rejected", + Self::OutOfEnergy => "out_of_energy", + Self::ClientConnectedError => "client_connected_error", + Self::ActorStartupFailure => "actor_startup_failure", + } + } +} + +pub fn record_client_disconnect(database_identity: Identity, cause: ClientDisconnectCause) { + WORKER_METRICS + .ws_client_disconnections + .with_label_values(&database_identity, cause.as_str()) + .inc(); +} + +pub fn record_client_rejection(database_identity: Identity, cause: ClientRejectCause) { + WORKER_METRICS + .ws_client_rejections + .with_label_values(&database_identity, cause.as_str()) + .inc(); +} + +/// Records at most one disconnect cause for a single accepted websocket client. +#[derive(Clone, Debug)] +pub struct ClientDisconnectRecorder { + database_identity: Identity, + recorded: Arc, +} + +impl ClientDisconnectRecorder { + pub fn new(database_identity: Identity) -> Self { + Self { + database_identity, + recorded: Arc::new(AtomicBool::new(false)), + } + } + + pub fn record(&self, cause: ClientDisconnectCause) -> bool { + if self + .recorded + .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + record_client_disconnect(self.database_identity, cause); + true + } else { + false + } + } +} + metrics_group!( pub struct WorkerMetrics { #[name = spacetime_worker_connected_clients] @@ -41,6 +192,25 @@ metrics_group!( #[labels(database_identity: Identity)] pub ws_clients_idle_timed_out: IntCounterVec, + // Compatibility counters above continue to be emitted for existing dashboards. + // Accepted-client disconnection `cause` label values are: + // client_close, idle_timeout, incoming_queue_full, outgoing_queue_full, + // module_exited, client_message_error, websocket_receive_error, + // websocket_send_error, websocket_stream_ended, unknown. + #[name = spacetime_worker_ws_client_disconnections_total] + #[help = "The cumulative number of accepted websocket client disconnections by cause. Cause values are documented by ClientDisconnectCause::ALL."] + #[labels(database_identity: Identity, cause: str)] + pub ws_client_disconnections: IntCounterVec, + + // Pre-connected rejection `cause` label values are: + // invalid_protocol, authorization_failed, websocket_upgrade_error, + // client_connected_rejected, out_of_energy, client_connected_error, + // actor_startup_failure. + #[name = spacetime_worker_ws_client_rejections_total] + #[help = "The cumulative number of websocket client attempts rejected before being counted as connected. Cause values are documented by ClientRejectCause::ALL."] + #[labels(database_identity: Identity, cause: str)] + pub ws_client_rejections: IntCounterVec, + #[name = spacetime_websocket_requests_total] #[help = "The cumulative number of websocket request messages"] #[labels(database_identity: Identity, protocol: str)] @@ -247,6 +417,12 @@ metrics_group!( #[buckets(0.001, 0.01, 0.05, 0.1, 0.25, 0.5, 1.0)] pub websocket_serialize_secs: HistogramVec, + #[name = spacetime_websocket_pong_rtt_secs] + #[help = "Time from writing a Ping frame to the socket until the client's Pong is processed"] + #[labels(db: Identity)] + #[buckets(0.005, 0.025, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0)] + pub websocket_pong_rtt: HistogramVec, + #[name = spacetime_worker_instance_operation_queue_length] #[help = "Length of the wait queue for access to a module instance."] #[labels(database_identity: Identity)] @@ -625,3 +801,16 @@ pub fn spawn_tokio_stats(node_id: String, rt_id: String, rt: tokio::runtime::Han }); }); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn client_disconnect_recorder_records_only_first_cause() { + let recorder = ClientDisconnectRecorder::new(Identity::ZERO); + + assert!(recorder.record(ClientDisconnectCause::ClientClose)); + assert!(!recorder.record(ClientDisconnectCause::Unknown)); + } +}