Add client_connection_outgoing_queue_length metric

Like the previous metric added in this branch, it's per-client,
so we'll use it for testing, but likely not merge it into master.
I'll follow up in a separate PR with a version that's per-database instead.

Author: gefjon

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

@@ -237,12 +237,22 @@ async fn ws_client_actor_inner(
             .remove_label_values(&addr, &client_identity, &connection_id) {
                 log::error!("Failed to `remove_label_values` for `client_connection_incoming_queue_length`: {e:?}");
             };
+
+        if let Err(e) = WORKER_METRICS
+            .client_connection_outgoing_queue_length
+            .remove_label_values(&addr, &client_identity, &connection_id) {
+                log::error!("Failed to `remove_label_values` for `client_connection_outgoing_queue_length`: {e:?}");
+            }
     );
 
-    let queue_length_metric = WORKER_METRICS
+    let incoming_queue_length_metric = WORKER_METRICS
         .client_connection_incoming_queue_length
         .with_label_values(&addr, &client_identity, &connection_id);
 
+    let outgoing_queue_length_metric = WORKER_METRICS
+        .client_connection_outgoing_queue_length
+        .with_label_values(&addr, &client_identity, &connection_id);
+
     loop {
         rx_buf.clear();
         enum Item {
@@ -251,7 +261,7 @@ async fn ws_client_actor_inner(
         }
         if let MaybeDone::Gone = *current_message {
             if let Some((message, timer)) = message_queue.pop_front() {
-                queue_length_metric.dec();
+                incoming_queue_length_metric.dec();
                 let client = client.clone();
                 let fut = async move { client.handle_message(message, timer).await };
                 current_message.set(MaybeDone::Future(fut));
@@ -285,6 +295,7 @@ async fn ws_client_actor_inner(
             // If we have an outgoing message to send, send it off.
             // No incoming `message` to handle, so `continue`.
             Some(n) = sendrx.recv_many(&mut rx_buf, 32).map(|n| (n != 0).then_some(n)) => {
+                outgoing_queue_length_metric.sub(n as _);
                 if closed {
                     // TODO: this isn't great. when we receive a close request from the peer,
                     //       tungstenite doesn't let us send any new messages on the socket,
@@ -378,7 +389,7 @@ async fn ws_client_actor_inner(
         match message {
             Item::Message(ClientMessage::Message(message)) => {
                 let timer = Instant::now();
-                queue_length_metric.inc();
+                incoming_queue_length_metric.inc();
                 message_queue.push_back((message, timer))
             }
             Item::HandleResult(res) => {

crates/core/src/client/client_connection.rs

@@ -16,6 +16,7 @@ use bytes::Bytes;
 use bytestring::ByteString;
 use derive_more::From;
 use futures::prelude::*;
+use prometheus::IntGauge;
 use spacetimedb_client_api_messages::websocket::{
     BsatnFormat, CallReducerFlags, Compression, FormatSwitch, JsonFormat, SubscribeMulti, SubscribeSingle, Unsubscribe,
     UnsubscribeMulti, WebsocketFormat,
@@ -69,6 +70,7 @@ pub struct ClientConnectionSender {
     sendtx: mpsc::Sender<SerializableMessage>,
     abort_handle: AbortHandle,
     cancelled: AtomicBool,
+    sendtx_queue_size_metric: Option<IntGauge>,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -87,13 +89,15 @@ impl ClientConnectionSender {
             Ok(h) => h.spawn(async {}).abort_handle(),
             Err(_) => tokio::runtime::Runtime::new().unwrap().spawn(async {}).abort_handle(),
         };
+
         (
             Self {
                 id,
                 config,
                 sendtx,
                 abort_handle,
                 cancelled: AtomicBool::new(false),
+                sendtx_queue_size_metric: None,
             },
             rx,
         )
@@ -111,6 +115,11 @@ impl ClientConnectionSender {
         if self.cancelled.load(Relaxed) {
             return Err(ClientSendError::Cancelled);
         }
+
+        if let Some(metric) = &self.sendtx_queue_size_metric {
+            metric.inc();
+        }
+
         self.sendtx.try_send(message).map_err(|e| match e {
             mpsc::error::TrySendError::Full(_) => {
                 // we've hit CLIENT_CHANNEL_CAPACITY messages backed up in
@@ -216,12 +225,17 @@ impl ClientConnection {
         })
         .abort_handle();
 
+        let sendtx_queue_size_metric = WORKER_METRICS
+            .client_connection_outgoing_queue_length
+            .with_label_values(&db, &id.identity, &id.connection_id);
+
         let sender = Arc::new(ClientConnectionSender {
             id,
             config,
             sendtx,
             abort_handle,
             cancelled: AtomicBool::new(false),
+            sendtx_queue_size_metric: Some(sendtx_queue_size_metric),
         });
         let this = Self {
             sender,

crates/core/src/worker_metrics/mod.rs

@@ -262,6 +262,11 @@ metrics_group!(
         #[help = "The number of client -> server WebSocket messages waiting in a client connection's incoming queue"]
         #[labels(db: Identity, client_identity: Identity, connection_id: ConnectionId)]
         pub client_connection_incoming_queue_length: IntGaugeVec,
+
+        #[name = spacetime_client_connection_outgoing_queue_length]
+        #[help = "The number of server -> client WebSocket messages waiting in a client connection's outgoing queue"]
+        #[labels(db: Identity, client_identity: Identity, connection_id: ConnectionId)]
+        pub client_connection_outgoing_queue_length: IntGaugeVec,
     }
 );