STR-314: add Unix domain socket support for gRPC listener (#671)

* feat: UDS multi-listener support with SpyIncoming and MeteredLayer

* feat: add UDS client support with connect_uds and --uds CLI flag

Author: Ozodimgba

CHANGELOG.md

@@ -11,6 +11,9 @@ The minor version will be incremented upon a breaking change and the patch versi
 ## [Unreleased]
 
 ### Features
+- geyser: add Unix domain socket support for multiple gRPC listener ([#671](https://github.com/rpcpool/yellowstone-grpc/pull/671))
+- client: add `connect_uds` method for Unix domain socket connections ([#671](https://github.com/rpcpool/yellowstone-grpc/pull/671))
+- client-simple: add `--uds` flag for Unix domain socket connections ([#671](https://github.com/rpcpool/yellowstone-grpc/pull/671))
 
 - geyser: configurable per-subscriber subscription limit with enforce/observe mode ([#680](https://github.com/rpcpool/yellowstone-grpc/pull/680))
 

Cargo.lock

@@ -62,6 +62,7 @@ thiserror = "2.0.16"
 tokio = "1.47.1"
 tokio-stream = "0.1.17"
 tokio-util = "0.7.0"
+tower = "0.4"
 tower-layer = "0.3.3"
 tonic = "0.14.0"
 tonic-build = "0.14.0"

Cargo.toml

@@ -21,7 +21,9 @@ use {
     },
     tokio::{fs, sync::Mutex},
     tonic::transport::{channel::ClientTlsConfig, Certificate},
-    yellowstone_grpc_client::{GeyserGrpcClient, GeyserGrpcClientError, Interceptor},
+    yellowstone_grpc_client::{
+        GeyserGrpcBuilder, GeyserGrpcClient, GeyserGrpcClientError, Interceptor,
+    },
     yellowstone_grpc_geyser::plugin::{convert_from, filter::message::FilteredUpdate},
     yellowstone_grpc_proto::{
         geyser::SlotStatus,
@@ -89,6 +91,10 @@ struct Args {
     #[clap(long)]
     connect_timeout_ms: Option<u64>,
 
+    /// Connect via Unix Domain Socket at this path instead of TCP
+    #[clap(long)]
+    uds: Option<PathBuf>,
+
     /// Sets the tower service default internal buffer size, default is 1024
     #[clap(long)]
     buffer_size: Option<usize>,
@@ -150,7 +156,7 @@ impl Args {
         Some(self.commitment.unwrap_or_default().into())
     }
 
-    async fn connect(&self) -> anyhow::Result<GeyserGrpcClient<impl Interceptor + Clone>> {
+    async fn build(&self) -> anyhow::Result<GeyserGrpcBuilder> {
         let mut tls_config = ClientTlsConfig::new().with_native_roots();
         if let Some(path) = &self.ca_certificate {
             let bytes = fs::read(path).await?;
@@ -206,7 +212,7 @@ impl Args {
             builder = builder.timeout(Duration::from_millis(duration));
         }
 
-        builder.connect().await.map_err(Into::into)
+        Ok(builder)
     }
 }
 
@@ -652,6 +658,71 @@ impl Action {
     }
 }
 
+async fn run_action(
+    mut client: GeyserGrpcClient<impl Interceptor>,
+    action: &Action,
+    commitment: Option<CommitmentLevel>,
+) -> anyhow::Result<()> {
+    match action {
+        Action::HealthCheck => client
+            .health_check()
+            .await
+            .map_err(anyhow::Error::new)
+            .map(|response| info!("response: {response:?}")),
+        Action::HealthWatch => geyser_health_watch(client).await,
+        Action::Subscribe(_) => {
+            let (request, resub, stats, verify_encoding) = action
+                .get_subscribe_request(commitment)
+                .await?
+                .ok_or_else(|| anyhow::anyhow!("expect subscribe action"))?;
+
+            geyser_subscribe(client, request, resub, stats, verify_encoding).await
+        }
+        Action::SubscribeDeshred(_) => {
+            let (request, stats) = action
+                .get_subscribe_deshred_request()
+                .ok_or_else(|| anyhow::anyhow!("expect subscribe deshred action"))?;
+
+            geyser_subscribe_deshred(client, request, stats).await
+        }
+        Action::SubscribeReplayInfo => client
+            .subscribe_replay_info()
+            .await
+            .map_err(anyhow::Error::new)
+            .map(|response| info!("response: {response:?}")),
+        Action::Ping { count } => client
+            .ping(*count)
+            .await
+            .map_err(anyhow::Error::new)
+            .map(|response| info!("response: {response:?}")),
+        Action::GetLatestBlockhash => client
+            .get_latest_blockhash(commitment)
+            .await
+            .map_err(anyhow::Error::new)
+            .map(|response| info!("response: {response:?}")),
+        Action::GetBlockHeight => client
+            .get_block_height(commitment)
+            .await
+            .map_err(anyhow::Error::new)
+            .map(|response| info!("response: {response:?}")),
+        Action::GetSlot => client
+            .get_slot(commitment)
+            .await
+            .map_err(anyhow::Error::new)
+            .map(|response| info!("response: {response:?}")),
+        Action::IsBlockhashValid { blockhash } => client
+            .is_blockhash_valid(blockhash.clone(), commitment)
+            .await
+            .map_err(anyhow::Error::new)
+            .map(|response| info!("response: {response:?}")),
+        Action::GetVersion => client
+            .get_version()
+            .await
+            .map_err(anyhow::Error::new)
+            .map(|response| info!("response: {response:?}")),
+    }
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     env::set_var(
@@ -680,74 +751,29 @@ async fn main() -> anyhow::Result<()> {
             drop(zero_attempts);
 
             let commitment = args.get_commitment();
-            let mut client = args.connect().await.map_err(backoff::Error::transient)?;
-            info!("Connected");
+            let builder = args.build().await.map_err(backoff::Error::transient)?;
 
-            match &args.action {
-                Action::HealthCheck => client
-                    .health_check()
-                    .await
-                    .map_err(anyhow::Error::new)
-                    .map(|response| info!("response: {response:?}")),
-                Action::HealthWatch => geyser_health_watch(client).await,
-                Action::Subscribe(_) => {
-                    let (request, resub, stats, verify_encoding) = args
-                        .action
-                        .get_subscribe_request(commitment)
-                        .await
-                        .map_err(backoff::Error::Permanent)?
-                        .ok_or(backoff::Error::Permanent(anyhow::anyhow!(
-                            "expect subscribe action"
-                        )))?;
-
-                    geyser_subscribe(client, request, resub, stats, verify_encoding).await
-                }
-                Action::SubscribeDeshred(_) => {
-                    let (request, stats) = args.action.get_subscribe_deshred_request().ok_or(
-                        backoff::Error::Permanent(anyhow::anyhow!(
-                            "expect subscribe deshred action"
-                        )),
-                    )?;
-
-                    geyser_subscribe_deshred(client, request, stats).await
-                }
-                Action::SubscribeReplayInfo => client
-                    .subscribe_replay_info()
-                    .await
-                    .map_err(anyhow::Error::new)
-                    .map(|response| info!("response: {response:?}")),
-                Action::Ping { count } => client
-                    .ping(*count)
-                    .await
-                    .map_err(anyhow::Error::new)
-                    .map(|response| info!("response: {response:?}")),
-                Action::GetLatestBlockhash => client
-                    .get_latest_blockhash(commitment)
-                    .await
-                    .map_err(anyhow::Error::new)
-                    .map(|response| info!("response: {response:?}")),
-                Action::GetBlockHeight => client
-                    .get_block_height(commitment)
+            if let Some(uds_path) = &args.uds {
+                let client = builder
+                    .connect_uds(uds_path)
                     .await
                     .map_err(anyhow::Error::new)
-                    .map(|response| info!("response: {response:?}")),
-                Action::GetSlot => client
-                    .get_slot(commitment)
+                    .map_err(backoff::Error::transient)?;
+                info!("Connected");
+                run_action(client, &args.action, commitment)
                     .await
-                    .map_err(anyhow::Error::new)
-                    .map(|response| info!("response: {response:?}")),
-                Action::IsBlockhashValid { blockhash } => client
-                    .is_blockhash_valid(blockhash.clone(), commitment)
+                    .map_err(backoff::Error::transient)?;
+            } else {
+                let client = builder
+                    .connect()
                     .await
                     .map_err(anyhow::Error::new)
-                    .map(|response| info!("response: {response:?}")),
-                Action::GetVersion => client
-                    .get_version()
+                    .map_err(backoff::Error::transient)?;
+                info!("Connected");
+                run_action(client, &args.action, commitment)
                     .await
-                    .map_err(anyhow::Error::new)
-                    .map(|response| info!("response: {response:?}")),
+                    .map_err(backoff::Error::transient)?;
             }
-            .map_err(backoff::Error::transient)?;
 
             Ok::<(), backoff::Error<anyhow::Error>>(())
         }

examples/rust/src/bin/client.rs

@@ -18,9 +18,13 @@ account-data-as-bytes = [
 [dependencies]
 bytes = { workspace = true }
 futures = { workspace = true }
+hyper-util = { workspace = true }
 thiserror = { workspace = true }
+tokio = { workspace = true }
 tonic = { workspace = true, features = ["tls-native-roots"] }
 tonic-health = { workspace = true }
+tower = { workspace = true, features = ["util"] }
+
 
 # Yellowstone
 yellowstone-grpc-proto = { workspace = true, features = ["tonic", "tonic-compression"] }

yellowstone-grpc-client/Cargo.toml

@@ -6,12 +6,16 @@ use {
         sink::{Sink, SinkExt},
         stream::Stream,
     },
-    std::time::Duration,
+    std::{path::PathBuf, time::Duration},
+    tokio::net::UnixStream,
     tonic::{
         codec::{CompressionEncoding, Streaming},
         metadata::{errors::InvalidMetadataValue, AsciiMetadataValue, MetadataValue},
         service::interceptor::InterceptedService,
-        transport::channel::{Channel, Endpoint},
+        transport::{
+            channel::{Channel, Endpoint},
+            Uri,
+        },
         Request, Response, Status,
     },
     tonic_health::pb::{health_client::HealthClient, HealthCheckRequest, HealthCheckResponse},
@@ -333,6 +337,32 @@ impl GeyserGrpcBuilder {
         self.build(channel)
     }
 
+    /// Connect to a gRPC server over a Unix Domain Socket.
+    ///
+    /// The `path` is the filesystem path to the socket (e.g. "/tmp/yellowstone.sock").
+    /// tonic requires a dummy HTTP URI for the channel, but the actual transport
+    /// goes through the UDS connector.
+    pub async fn connect_uds(
+        self,
+        path: impl Into<PathBuf>,
+    ) -> GeyserGrpcBuilderResult<GeyserGrpcClient<impl Interceptor + Clone>> {
+        let path = path.into();
+
+        // tonic needs an Endpoint to hang config off of, but the URI is ignored
+        // by the connector — all traffic goes through the UnixStream.
+        let channel = Endpoint::from_static("http://[::]:0")
+            .connect_with_connector(tower::service_fn(move |_: Uri| {
+                let path = path.clone();
+                async move {
+                    let stream = UnixStream::connect(path).await?;
+                    Ok::<_, std::io::Error>(hyper_util::rt::TokioIo::new(stream))
+                }
+            }))
+            .await?;
+
+        self.build(channel)
+    }
+
     // Set x-token
     pub fn x_token<T>(self, x_token: Option<T>) -> GeyserGrpcBuilderResult<Self>
     where