rpcpool · fanatid · Jan 6, 2025 · Jan 6, 2025 · Jan 6, 2025 · Jan 6, 2025
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -18,6 +18,15 @@ The minor version will be incremented upon a breaking change and the patch versi
 
 ### Breaking
 
+## 2024-01-06
+
+- yellowstone-grpc-geyser-4.3.0
+- yellowstone-grpc-proto-4.2.0
+
+### Features
+
+- geyser: add connection slot lag metric ([#508](https://github.com/rpcpool/yellowstone-grpc/pull/508))
+
 ## 2024-12-16
 
 - yellowstone-grpc-client-simple-4.3.0

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -3,8 +3,8 @@ resolver = "2"
 members = [
     "examples/rust", # 4.3.0
     "yellowstone-grpc-client", # 4.1.0
-    "yellowstone-grpc-geyser", # 4.2.2
-    "yellowstone-grpc-proto", # 4.1.1
+    "yellowstone-grpc-geyser", # 4.3.0
+    "yellowstone-grpc-proto", # 4.2.0
 ]
 exclude = [
     "yellowstone-grpc-client-nodejs/solana-encoding-wasm", # 3.0.0
@@ -64,13 +64,12 @@ smallvec = "1.13.2"
 spl-token-2022 = "6.0.0"
 thiserror = "1.0.63"
 tokio = "1.21.2"
-tokio-stream = "0.1.11"
 tonic = "0.12.1"
 tonic-build = "0.12.1"
 tonic-health = "0.12.1"
 vergen = "9.0.0"
 yellowstone-grpc-client = { path = "yellowstone-grpc-client", version = "4.1.0" }
-yellowstone-grpc-proto = { path = "yellowstone-grpc-proto", version = "4.1.1", default-features = false }
+yellowstone-grpc-proto = { path = "yellowstone-grpc-proto", version = "4.2.0", default-features = false }
 
 [workspace.lints.clippy]
 clone_on_ref_ptr = "deny"

diff --git a/yellowstone-grpc-client-nodejs/solana-encoding-wasm/Cargo.lock b/yellowstone-grpc-client-nodejs/solana-encoding-wasm/Cargo.lock
diff --git a/yellowstone-grpc-client-nodejs/solana-encoding-wasm/Cargo.toml b/yellowstone-grpc-client-nodejs/solana-encoding-wasm/Cargo.toml
@@ -16,7 +16,7 @@ crate-type = ["cdylib"]
 serde_json = "1.0.86"
 solana-transaction-status = "~2.1.1"
 wasm-bindgen = "0.2.95"
-yellowstone-grpc-proto = { path = "../../yellowstone-grpc-proto", version = "4.1.1", default-features = false, features = ["convert"] }
+yellowstone-grpc-proto = { path = "../../yellowstone-grpc-proto", version = "4.2.0", default-features = false, features = ["convert"] }
 
 [workspace.lints.clippy]
 clone_on_ref_ptr = "deny"

diff --git a/yellowstone-grpc-geyser/Cargo.toml b/yellowstone-grpc-geyser/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "yellowstone-grpc-geyser"
-version = "4.2.2"
+version = "4.3.0"
 authors = { workspace = true }
 edition = { workspace = true }
 description = "Yellowstone gRPC Geyser Plugin"
@@ -44,7 +44,6 @@ solana-transaction-status = { workspace = true }
 spl-token-2022 = { workspace = true, features = ["no-entrypoint"] }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "fs"] }
-tokio-stream = { workspace = true }
 tonic = { workspace = true, features = ["gzip", "zstd", "tls", "tls-roots"] }
 tonic-health = { workspace = true }
 yellowstone-grpc-proto = { workspace = true, features = ["convert", "plugin"] }

diff --git a/yellowstone-grpc-geyser/config.json b/yellowstone-grpc-geyser/config.json
@@ -86,6 +86,7 @@
     }
   },
   "prometheus": {
-    "address": "0.0.0.0:8999"
+    "address": "0.0.0.0:8999",
+    "metric_connection_slot_lag": false
   }
 }
diff --git a/yellowstone-grpc-geyser/src/config.rs b/yellowstone-grpc-geyser/src/config.rs
@@ -302,6 +302,9 @@ impl ConfigGrpcCompression {
 pub struct ConfigPrometheus {
     /// Address of Prometheus service.
     pub address: SocketAddr,
+    /// Sent slots to lag metric
+    #[serde(default)]
+    pub metric_connection_slot_lag: bool,
 }
 
 #[derive(Deserialize)]

diff --git a/yellowstone-grpc-geyser/src/grpc.rs b/yellowstone-grpc-geyser/src/grpc.rs
@@ -1,10 +1,11 @@
 use {
     crate::{
         config::{ConfigGrpc, ConfigTokio},
-        metrics::{self, DebugClientMessage},
+        metrics::{self, ConnectionCloseStatus, DebugClientMessage},
         version::GrpcVersionInfo,
     },
-    anyhow::Context,
+    anyhow::Context as _,
+    futures::stream::Stream,
     log::{error, info},
     prost_types::Timestamp,
     solana_sdk::{
@@ -13,10 +14,12 @@ use {
     },
     std::{
         collections::{BTreeMap, HashMap},
+        pin::Pin,
         sync::{
             atomic::{AtomicUsize, Ordering},
             Arc,
         },
+        task::{Context, Poll},
         time::SystemTime,
     },
     tokio::{
@@ -26,7 +29,6 @@ use {
         task::spawn_blocking,
         time::{sleep, Duration, Instant},
     },
-    tokio_stream::wrappers::ReceiverStream,
     tonic::{
         service::interceptor::interceptor,
         transport::{
@@ -344,6 +346,7 @@ pub struct GrpcService {
     snapshot_rx: Mutex<Option<crossbeam_channel::Receiver<Box<Message>>>>,
     broadcast_tx: broadcast::Sender<BroadcastedMessage>,
     replay_stored_slots_tx: Option<mpsc::Sender<ReplayStoredSlotsRequest>>,
+    metric_connection_slot_lag: bool,
     debug_clients_tx: Option<mpsc::UnboundedSender<DebugClientMessage>>,
     filter_names: Arc<Mutex<FilterNames>>,
 }
@@ -353,6 +356,7 @@ impl GrpcService {
     pub async fn create(
         config_tokio: ConfigTokio,
         config: ConfigGrpc,
+        metric_connection_slot_lag: bool,
         debug_clients_tx: Option<mpsc::UnboundedSender<DebugClientMessage>>,
         is_reload: bool,
     ) -> anyhow::Result<(
@@ -442,6 +446,7 @@ impl GrpcService {
             snapshot_rx: Mutex::new(snapshot_rx),
             broadcast_tx: broadcast_tx.clone(),
             replay_stored_slots_tx,
+            metric_connection_slot_lag,
             debug_clients_tx,
             filter_names,
         })
@@ -896,6 +901,7 @@ impl GrpcService {
                                 if let Some(msg) = filter_new.get_pong_msg() {
                                     if stream_tx.send(Ok(msg)).await.is_err() {
                                         error!("client #{id}: stream closed");
+                                        metrics::connections_close_status_inc(ConnectionCloseStatus::TxClosed);
                                         break 'outer;
                                     }
                                     continue;
@@ -912,6 +918,7 @@ impl GrpcService {
                                         tokio::spawn(async move {
                                             let _ = stream_tx.send(Err(Status::internal("from_slot is not supported"))).await;
                                         });
+                                        metrics::connections_close_status_inc(ConnectionCloseStatus::FromSlot);
                                         break 'outer;
                                     };
 
@@ -922,6 +929,7 @@ impl GrpcService {
                                         tokio::spawn(async move {
                                             let _ = stream_tx.send(Err(Status::internal("failed to send from_slot request"))).await;
                                         });
+                                        metrics::connections_close_status_inc(ConnectionCloseStatus::FromSlot);
                                         break 'outer;
                                     }
 
@@ -935,13 +943,15 @@ impl GrpcService {
                                                 );
                                                 let _ = stream_tx.send(Err(Status::internal(message))).await;
                                             });
+                                            metrics::connections_close_status_inc(ConnectionCloseStatus::FromSlot);
                                             break 'outer;
                                         },
                                         Err(_error) => {
                                             error!("client #{id}: failed to get replay response");
                                             tokio::spawn(async move {
                                                 let _ = stream_tx.send(Err(Status::internal("failed to get replay response"))).await;
                                             });
+                                            metrics::connections_close_status_inc(ConnectionCloseStatus::FromSlot);
                                             break 'outer;
                                         }
                                     };
@@ -953,6 +963,7 @@ impl GrpcService {
                                                 Ok(()) => {}
                                                 Err(mpsc::error::SendError(_)) => {
                                                     error!("client #{id}: stream closed");
+                                                    metrics::connections_close_status_inc(ConnectionCloseStatus::TxClosed);
                                                     break 'outer;
                                                 }
                                             }
@@ -961,9 +972,11 @@ impl GrpcService {
                                 }
                             }
                             Some(None) => {
+                                metrics::connections_close_status_inc(ConnectionCloseStatus::RxClosed);
                                 break 'outer;
                             },
                             None => {
+                                metrics::connections_close_status_inc(ConnectionCloseStatus::RxClosed);
                                 break 'outer;
                             }
                         }
@@ -972,13 +985,15 @@ impl GrpcService {
                         let (commitment, messages) = match message {
                             Ok((commitment, messages)) => (commitment, messages),
                             Err(broadcast::error::RecvError::Closed) => {
+                                metrics::connections_close_status_inc(ConnectionCloseStatus::Internal);
                                 break 'outer;
                             },
                             Err(broadcast::error::RecvError::Lagged(_)) => {
                                 info!("client #{id}: lagged to receive geyser messages");
                                 tokio::spawn(async move {
                                     let _ = stream_tx.send(Err(Status::internal("lagged to receive geyser messages"))).await;
                                 });
+                                metrics::connections_close_status_inc(ConnectionCloseStatus::Lagged);
                                 break 'outer;
                             }
                         };
@@ -993,10 +1008,12 @@ impl GrpcService {
                                             tokio::spawn(async move {
                                                 let _ = stream_tx.send(Err(Status::internal("lagged to send an update"))).await;
                                             });
+                                            metrics::connections_close_status_inc(ConnectionCloseStatus::Lagged);
                                             break 'outer;
                                         }
                                         Err(mpsc::error::TrySendError::Closed(_)) => {
                                             error!("client #{id}: stream closed");
+                                            metrics::connections_close_status_inc(ConnectionCloseStatus::TxClosed);
                                             break 'outer;
                                         }
                                     }
@@ -1089,7 +1106,7 @@ impl GrpcService {
 
 #[tonic::async_trait]
 impl Geyser for GrpcService {
-    type SubscribeStream = ReceiverStream<TonicResult<FilteredUpdate>>;
+    type SubscribeStream = ReceiverStream;
 
     async fn subscribe(
         &self,
@@ -1206,7 +1223,10 @@ impl Geyser for GrpcService {
             },
         ));
 
-        Ok(Response::new(ReceiverStream::new(stream_rx)))
+        Ok(Response::new(ReceiverStream::new(
+            stream_rx,
+            self.metric_connection_slot_lag,
+        )))
     }
 
     async fn ping(&self, request: Request<PingRequest>) -> Result<Response<PongResponse>, Status> {
@@ -1297,3 +1317,44 @@ impl Geyser for GrpcService {
         }))
     }
 }
+
+#[derive(Debug)]
+pub struct ReceiverStream {
+    rx: mpsc::Receiver<TonicResult<FilteredUpdate>>,
+    metric_connection_slot_lag: bool,
+    max_slot: Slot,
+}
+
+impl ReceiverStream {
+    const fn new(
+        rx: mpsc::Receiver<TonicResult<FilteredUpdate>>,
+        metric_connection_slot_lag: bool,
+    ) -> Self {
+        Self {
+            rx,
+            metric_connection_slot_lag,
+            max_slot: 0,
+        }
+    }
+}
+
+impl Stream for ReceiverStream {
+    type Item = TonicResult<FilteredUpdate>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let value = futures::ready!(self.rx.poll_recv(cx));
+        if self.metric_connection_slot_lag {
+            if let Some(slot) = value
+                .as_ref()
+                .and_then(|item| item.as_ref().map(|item| item.message.get_slot()).ok())
+                .flatten()
+            {
+                if slot > self.max_slot {
+                    self.max_slot = slot;
+                    metrics::connections_slot_lag_observe(slot);
+                }
+            }
+        }
+        Poll::Ready(value)
+    }
+}