Add metric for size of ClientMessages received by sync socket (#40977)

GitOrigin-RevId: 6cb2520286cdb58096beef354c84572545c7205c

Author: emmaling27

Cargo.lock

@@ -207,6 +207,7 @@ dependencies = [
  "rand 0.9.0",
  "serde",
  "serde_json",
+ "strum",
  "uuid",
 ]
 
@@ -536,6 +537,12 @@ dependencies = [
  "http",
 ]
 
+[[package]]
+name = "heck"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
+
 [[package]]
 name = "http"
 version = "1.3.1"
@@ -1279,6 +1286,12 @@ dependencies = [
  "untrusted",
 ]
 
+[[package]]
+name = "rustversion"
+version = "1.0.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4f3208ce4d8448b3f3e7d168a73f5e0c43a61e32930de3bceeccedb388b6bf06"
+
 [[package]]
 name = "rusty-fork"
 version = "0.3.0"
@@ -1447,6 +1460,28 @@ version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
 
+[[package]]
+name = "strum"
+version = "0.27.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f64def088c51c9510a8579e3c5d67c65349dcf755e5479ad3d010aa6454e2c32"
+dependencies = [
+ "strum_macros",
+]
+
+[[package]]
+name = "strum_macros"
+version = "0.27.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c77a8c5abcaf0f9ce05d62342b7d298c346515365c36b673df4ebe3ced01fde8"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "quote",
+ "rustversion",
+ "syn 2.0.95",
+]
+
 [[package]]
 name = "subtle"
 version = "2.5.0"

crates/convex/Cargo.oss.lock

@@ -91,6 +91,7 @@ dependencies = [
  "rand 0.9.0",
  "serde",
  "serde_json",
+ "strum",
  "uuid",
 ]
 
@@ -252,6 +253,12 @@ dependencies = [
  "http",
 ]
 
+[[package]]
+name = "heck"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
+
 [[package]]
 name = "http"
 version = "1.3.1"
@@ -508,6 +515,12 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "rustversion"
+version = "1.0.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4f3208ce4d8448b3f3e7d168a73f5e0c43a61e32930de3bceeccedb388b6bf06"
+
 [[package]]
 name = "rusty-fork"
 version = "0.3.0"
@@ -570,6 +583,28 @@ dependencies = [
  "digest",
 ]
 
+[[package]]
+name = "strum"
+version = "0.27.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f64def088c51c9510a8579e3c5d67c65349dcf755e5479ad3d010aa6454e2c32"
+dependencies = [
+ "strum_macros",
+]
+
+[[package]]
+name = "strum_macros"
+version = "0.27.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c77a8c5abcaf0f9ce05d62342b7d298c346515365c36b673df4ebe3ced01fde8"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "quote",
+ "rustversion",
+ "syn 2.0.95",
+]
+
 [[package]]
 name = "syn"
 version = "1.0.109"

crates/convex/sync_types/Cargo.oss.lock

@@ -24,6 +24,7 @@ proptest-derive = { optional = true, version = "0.5.0" }
 rand = { version = "0.9" }
 serde = { features = [ "derive" ], version = "1" }
 serde_json = { features = [ "float_roundtrip", "preserve_order" ], version = "1" }
+strum = { version = "0.27", features = [ "derive" ] }
 uuid = { features = [ "serde", "v4" ], version = "1.6" }
 
 [dev-dependencies]

crates/convex/sync_types/Cargo.oss.toml

@@ -27,6 +27,7 @@ serde_json = { workspace = true, features = [
     "float_roundtrip",
     "preserve_order",
 ] }
+strum = { workspace = true }
 uuid = { workspace = true, features = ["serde", "v4"] }
 
 [dev-dependencies]

crates/convex/sync_types/Cargo.toml

@@ -15,6 +15,7 @@ use serde::{
     Serialize,
 };
 use serde_json::Value as JsonValue;
+use strum::AsRefStr;
 use uuid::Uuid;
 
 #[cfg(any(test, feature = "testing"))]
@@ -100,7 +101,7 @@ pub enum QuerySetModification {
     Remove { query_id: QueryId },
 }
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Eq, PartialEq, AsRefStr)]
 #[cfg_attr(any(test, feature = "testing"), derive(proptest_derive::Arbitrary))]
 pub enum ClientMessage {
     Connect {

crates/convex/sync_types/src/types.rs

@@ -28,6 +28,19 @@ pub fn log_websocket_message_in() {
     log_counter(&BACKEND_WS_IN_TOTAL, 1);
 }
 
+register_convex_histogram!(
+    WS_CLIENT_MESSAGE_BYTES,
+    "Size of client message received by the sync socket",
+    &["type"]
+);
+pub fn log_websocket_client_message_bytes(bytes: usize, message_type: String) {
+    log_distribution_with_labels(
+        &WS_CLIENT_MESSAGE_BYTES,
+        bytes as f64,
+        vec![StaticMetricLabel::new("type", message_type)],
+    );
+}
+
 register_convex_counter!(BACKEND_PING_TOTAL, "Number of websocket pings sent");
 pub fn log_websocket_ping() {
     log_counter(&BACKEND_PING_TOTAL, 1);

crates/local_backend/src/subs/metrics.rs

@@ -54,6 +54,7 @@ use sync::{
     SyncWorkerConfig,
 };
 use sync_types::{
+    ClientMessage,
     IdentityVersion,
     SessionId,
 };
@@ -76,7 +77,10 @@ use metrics::{
     websocket_upgrade_timer,
 };
 
-use crate::RouterState;
+use crate::{
+    subs::metrics::log_websocket_client_message_bytes,
+    RouterState,
+};
 
 /// How often heartbeat pings are sent.
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
@@ -161,7 +165,8 @@ async fn run_sync_socket(
 
             match message {
                 Message::Text(s) => {
-                    let body = serde_json::from_str::<JsonValue>(&s)
+                    let client_message_size = s.len();
+                    let body: ClientMessage = serde_json::from_str::<JsonValue>(&s)
                         .map_err(|e| anyhow::anyhow!(e))
                         .and_then(|body| body.try_into())
                         .map_err(|e| {
@@ -170,6 +175,10 @@ async fn run_sync_socket(
                                 format!("Received Invalid JSON on websocket: {e}"),
                             ))
                         })?;
+                    log_websocket_client_message_bytes(
+                        client_message_size,
+                        body.as_ref().to_string(),
+                    );
                     log_websocket_message_in();
                     if client_tx.send((body, st.runtime.monotonic_now())).is_err() {
                         break;