fix(hermes): observe WS latency histograms and satisfy clippy/fmt

Co-Authored-By: Tejas Badadare <[email protected]>

Author: devin-ai-integration[bot]

apps/hermes/server/src/api/rest/v2/sse.rs

@@ -1,4 +1,5 @@
 use {
+    crate::state::metrics::Metrics,
     crate::{
         api::{
             rest::{validate_price_ids, RestError},
@@ -16,19 +17,17 @@ use {
         response::sse::{Event, KeepAlive, Sse},
     },
     futures::Stream,
+    prometheus_client::metrics::histogram::Histogram,
     pyth_sdk::PriceIdentifier,
     serde::Deserialize,
     serde_qs::axum::QsQuery,
+    std::sync::Arc,
     std::{convert::Infallible, time::Duration},
     tokio::{sync::broadcast, time::Instant},
     tokio_stream::{wrappers::BroadcastStream, StreamExt as _},
-    prometheus_client::metrics::histogram::Histogram,
     utoipa::IntoParams,
-    std::sync::Arc,
-    crate::state::metrics::Metrics,
 };
 
-
 pub struct SseMetrics {
     pub receive_to_sse_send_latency: Histogram,
 }
@@ -40,15 +39,17 @@ impl SseMetrics {
         S: Send + Sync + 'static,
     {
         let receive_to_sse_send_latency = Histogram::new(
-            [0.1, 0.2, 0.3, 0.4, 0.5, 0.7, 1.0, 1.3, 1.7, 2.0, 3.0, 5.0, 10.0, 20.0].into_iter(),
+            [
+                0.1, 0.2, 0.3, 0.4, 0.5, 0.7, 1.0, 1.3, 1.7, 2.0, 3.0, 5.0, 10.0, 20.0,
+            ]
+            .into_iter(),
         );
 
         let new = Self {
             receive_to_sse_send_latency: receive_to_sse_send_latency.clone(),
         };
 
         {
-
             tokio::spawn(async move {
                 Metrics::register(
                     &*state,
@@ -66,7 +67,6 @@ impl SseMetrics {
     }
 }
 
-
 // Constants
 const MAX_CONNECTION_DURATION: Duration = Duration::from_secs(24 * 60 * 60); // 24 hours
 
@@ -242,7 +242,6 @@ where
 
     let mut parsed_price_updates: Vec<ParsedPriceUpdate> = price_feeds_with_update_data
         .price_feeds
-
         .into_iter()
         .map(|price_feed| price_feed.into())
         .collect();

apps/hermes/server/src/api/ws.rs

@@ -96,10 +96,16 @@ impl WsMetrics {
         S: Send + Sync + 'static,
     {
         let publish_to_receive_latency = Histogram::new(
-            [0.1, 0.2, 0.3, 0.4, 0.5, 0.7, 1.0, 1.3, 1.7, 2.0, 3.0, 5.0, 10.0, 20.0].into_iter(),
+            [
+                0.1, 0.2, 0.3, 0.4, 0.5, 0.7, 1.0, 1.3, 1.7, 2.0, 3.0, 5.0, 10.0, 20.0,
+            ]
+            .into_iter(),
         );
         let receive_to_ws_send_latency = Histogram::new(
-            [0.1, 0.2, 0.3, 0.4, 0.5, 0.7, 1.0, 1.3, 1.7, 2.0, 3.0, 5.0, 10.0, 20.0].into_iter(),
+            [
+                0.1, 0.2, 0.3, 0.4, 0.5, 0.7, 1.0, 1.3, 1.7, 2.0, 3.0, 5.0, 10.0, 20.0,
+            ]
+            .into_iter(),
         );
         let new = Self {
             interactions: Family::default(),
@@ -503,12 +509,30 @@ where
                         )
                         .await?;
                     self.sender.close().await?;
+                    if let Some(received_at) = received_at_opt {
+                        let pub_to_recv = (received_at - publish_time).max(0) as f64;
+                        self.ws_state
+                            .metrics
+                            .publish_to_receive_latency
+                            .observe(pub_to_recv);
+
+                        let now_secs = std::time::SystemTime::now()
+                            .duration_since(std::time::UNIX_EPOCH)
+                            .ok()
+                            .and_then(|d| i64::try_from(d.as_secs()).ok())
+                            .unwrap_or(received_at);
+                        let recv_to_send = (now_secs - received_at).max(0) as f64;
+                        self.ws_state
+                            .metrics
+                            .receive_to_ws_send_latency
+                            .observe(recv_to_send);
+                    }
+
                     self.closed = true;
                     return Ok(());
                 }
             }
 
-
             self.sender.feed(message.into()).await?;
 
             self.ws_state