metrics: disambiguate websocket receive disconnects (#5547)

# Description of Changes

Adds specific `ClientDisconnectCause` variants for each websocket
receive `WsError` variant, and records those specific causes in the
subscribe route instead of collapsing all receive failures into
`websocket_receive_error`.

The legacy `WebsocketReceiveError` enum variant and metric label are
retained, but the receive loop now emits the more specific labels.

# API and ABI breaking changes

This adds public Rust enum variants and new metric label values.
Existing variants are retained.

# Expected complexity level and risk

1

This is an observability-only change. Websocket control flow is
unchanged.

# Testing

- [x] `cargo fmt --all -- --check`
- [x] `cargo test -p spacetimedb-core
client_disconnect_cause_labels_are_unique`
- [x] `cargo test -p spacetimedb-client-api
recv_loop_terminates_when_input_yields_err`

Co-authored-by: clockwork-labs-bot <clockwork-labs-bot@users.noreply.github.com>
Co-authored-by: Tyler Cloutier <cloutiertyler@users.noreply.github.com>

Author: clockwork-labs-bot

crates/client-api/src/routes/subscribe.rs

@@ -892,6 +892,23 @@ fn ws_recv_loop(
     idle_tx: watch::Sender<Instant>,
     mut ws: impl Stream<Item = Result<WsMessage, WsError>> + Unpin,
 ) -> impl Stream<Item = ClientMessage> {
+    fn receive_error_cause(error: &WsError) -> ClientDisconnectCause {
+        match error {
+            WsError::ConnectionClosed => ClientDisconnectCause::WebsocketReceiveConnectionClosed,
+            WsError::AlreadyClosed => ClientDisconnectCause::WebsocketReceiveAlreadyClosed,
+            WsError::Io(_) => ClientDisconnectCause::WebsocketReceiveIo,
+            WsError::Tls(_) => ClientDisconnectCause::WebsocketReceiveTls,
+            WsError::Capacity(_) => ClientDisconnectCause::WebsocketReceiveCapacity,
+            WsError::Protocol(_) => ClientDisconnectCause::WebsocketReceiveProtocol,
+            WsError::WriteBufferFull(_) => ClientDisconnectCause::WebsocketReceiveWriteBufferFull,
+            WsError::Utf8(_) => ClientDisconnectCause::WebsocketReceiveUtf8,
+            WsError::AttackAttempt => ClientDisconnectCause::WebsocketReceiveAttackAttempt,
+            WsError::Url(_) => ClientDisconnectCause::WebsocketReceiveUrl,
+            WsError::Http(_) => ClientDisconnectCause::WebsocketReceiveHttp,
+            WsError::HttpFormat(_) => ClientDisconnectCause::WebsocketReceiveHttpFormat,
+        }
+    }
+
     // Get the next message from `ws`, or `None` if the stream is exhausted.
     //
     // If `state.closed`, `ws` is drained until it either yields an `Err`, is
@@ -948,27 +965,13 @@ fn ws_recv_loop(
                     log::trace!("message received while already closed");
                 }
                 // None of the error cases can be meaningfully recovered from
-                // (and some can't even occur on the `ws` stream).
-                // Exit here but spell out an exhaustive match
-                // in order to bring any future library changes to our attention.
-                Err(e) => match e {
-                    e @ (WsError::ConnectionClosed
-                    | WsError::AlreadyClosed
-                    | WsError::Io(_)
-                    | WsError::Tls(_)
-                    | WsError::Capacity(_)
-                    | WsError::Protocol(_)
-                    | WsError::WriteBufferFull(_)
-                    | WsError::Utf8(_)
-                    | WsError::AttackAttempt
-                    | WsError::Url(_)
-                    | WsError::Http(_)
-                    | WsError::HttpFormat(_)) => {
-                        state.record_disconnect(ClientDisconnectCause::WebsocketReceiveError);
-                        log::warn!("Websocket receive error: {e}");
-                        break;
-                    }
-                },
+                // (and some can't even occur on the `ws` stream), so record the
+                // specific receive error cause and terminate the stream.
+                Err(e) => {
+                    state.record_disconnect(receive_error_cause(&e));
+                    log::warn!("Websocket receive error: {e}");
+                    break;
+                }
             }
         }
     }
@@ -2129,7 +2132,8 @@ mod tests {
     #[tokio::test]
     async fn recv_loop_terminates_when_input_yields_err() {
         let state = Arc::new(actor_state_with_disconnect_recorder(2, <_>::default()));
-        let before = disconnect_count(state.database, ClientDisconnectCause::WebsocketReceiveError);
+        let cause = ClientDisconnectCause::WebsocketReceiveConnectionClosed;
+        let before = disconnect_count(state.database, cause);
         let (idle_tx, _idle_rx) = watch::channel(Instant::now() + state.config.idle_timeout);
 
         let input = stream::iter(vec![
@@ -2144,7 +2148,7 @@ mod tests {
 
         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);
+        assert_disconnect_count_incremented(state.database, cause, before);
     }
 
     #[tokio::test]

crates/core/src/worker_metrics/mod.rs

@@ -39,6 +39,30 @@ pub enum ClientDisconnectCause {
     ClientMessageError,
     /// The websocket receive stream returned an error.
     WebsocketReceiveError,
+    /// The websocket receive stream reported that the connection was already closed.
+    WebsocketReceiveConnectionClosed,
+    /// The websocket receive stream was polled after it had already closed.
+    WebsocketReceiveAlreadyClosed,
+    /// The websocket receive stream returned an IO error.
+    WebsocketReceiveIo,
+    /// The websocket receive stream returned a TLS error.
+    WebsocketReceiveTls,
+    /// The websocket receive stream returned a capacity error.
+    WebsocketReceiveCapacity,
+    /// The websocket receive stream returned a protocol error.
+    WebsocketReceiveProtocol,
+    /// The websocket receive stream reported a full write buffer while receiving.
+    WebsocketReceiveWriteBufferFull,
+    /// The websocket receive stream returned a UTF-8 error.
+    WebsocketReceiveUtf8,
+    /// The websocket receive stream detected an attack attempt.
+    WebsocketReceiveAttackAttempt,
+    /// The websocket receive stream returned a URL error.
+    WebsocketReceiveUrl,
+    /// The websocket receive stream returned an HTTP error.
+    WebsocketReceiveHttp,
+    /// The websocket receive stream returned an HTTP format error.
+    WebsocketReceiveHttpFormat,
     /// The server failed while sending or flushing websocket data.
     WebsocketSendError,
     /// The websocket receive stream ended without a more specific cause.
@@ -48,14 +72,26 @@ pub enum ClientDisconnectCause {
 }
 
 impl ClientDisconnectCause {
-    pub const ALL: [Self; 10] = [
+    pub const ALL: [Self; 22] = [
         Self::ClientClose,
         Self::IdleTimeout,
         Self::IncomingQueueFull,
         Self::OutgoingQueueFull,
         Self::ModuleExited,
         Self::ClientMessageError,
         Self::WebsocketReceiveError,
+        Self::WebsocketReceiveConnectionClosed,
+        Self::WebsocketReceiveAlreadyClosed,
+        Self::WebsocketReceiveIo,
+        Self::WebsocketReceiveTls,
+        Self::WebsocketReceiveCapacity,
+        Self::WebsocketReceiveProtocol,
+        Self::WebsocketReceiveWriteBufferFull,
+        Self::WebsocketReceiveUtf8,
+        Self::WebsocketReceiveAttackAttempt,
+        Self::WebsocketReceiveUrl,
+        Self::WebsocketReceiveHttp,
+        Self::WebsocketReceiveHttpFormat,
         Self::WebsocketSendError,
         Self::WebsocketStreamEnded,
         Self::Unknown,
@@ -70,6 +106,18 @@ impl ClientDisconnectCause {
             Self::ModuleExited => "module_exited",
             Self::ClientMessageError => "client_message_error",
             Self::WebsocketReceiveError => "websocket_receive_error",
+            Self::WebsocketReceiveConnectionClosed => "websocket_receive_connection_closed",
+            Self::WebsocketReceiveAlreadyClosed => "websocket_receive_already_closed",
+            Self::WebsocketReceiveIo => "websocket_receive_io",
+            Self::WebsocketReceiveTls => "websocket_receive_tls",
+            Self::WebsocketReceiveCapacity => "websocket_receive_capacity",
+            Self::WebsocketReceiveProtocol => "websocket_receive_protocol",
+            Self::WebsocketReceiveWriteBufferFull => "websocket_receive_write_buffer_full",
+            Self::WebsocketReceiveUtf8 => "websocket_receive_utf8",
+            Self::WebsocketReceiveAttackAttempt => "websocket_receive_attack_attempt",
+            Self::WebsocketReceiveUrl => "websocket_receive_url",
+            Self::WebsocketReceiveHttp => "websocket_receive_http",
+            Self::WebsocketReceiveHttpFormat => "websocket_receive_http_format",
             Self::WebsocketSendError => "websocket_send_error",
             Self::WebsocketStreamEnded => "websocket_stream_ended",
             Self::Unknown => "unknown",
@@ -196,7 +244,13 @@ metrics_group!(
         // 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.
+        // websocket_receive_connection_closed, websocket_receive_already_closed,
+        // websocket_receive_io, websocket_receive_tls, websocket_receive_capacity,
+        // websocket_receive_protocol, websocket_receive_write_buffer_full,
+        // websocket_receive_utf8, websocket_receive_attack_attempt,
+        // websocket_receive_url, websocket_receive_http,
+        // websocket_receive_http_format, 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)]
@@ -805,6 +859,15 @@ pub fn spawn_tokio_stats(node_id: String, rt_id: String, rt: tokio::runtime::Han
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::collections::HashSet;
+
+    #[test]
+    fn client_disconnect_cause_labels_are_unique() {
+        let mut labels = HashSet::new();
+        for cause in ClientDisconnectCause::ALL {
+            assert!(labels.insert(cause.as_str()), "duplicate label for {cause:?}");
+        }
+    }
 
     #[test]
     fn client_disconnect_recorder_records_only_first_cause() {