rust volume: build outgoing gRPC clients in one place and give every dial a connect timeout (#11354)

server/grpc_client.rs stopped at build_grpc_endpoint() -> Endpoint, so all 13 production call sites hand-wrote the same .connect() + X::with_interceptor() + two max_*_message_size() lines. Four of them -- VolumeCopy, VolumeTailReceiver, VolumeEcShardsCopy and the HTTP chunk batch-delete fan-out -- dialed with no timeout at all, so an unreachable peer whose TCP handshake never completes (SYN dropped, blackholed route, host behind a silent firewall) left the operation waiting on the kernel's own retry budget, minutes long. Add GrpcDialOptions (unary / long / stream presets), connect_channel(), and volume_server_client() / master_client() / filer_client() constructors that attach the request-id interceptor and lift both message-size limits, then route all 13 sites through them. build_grpc_endpoint is private again, so connect_channel is the only way out of the module and no call site can dial without picking up a bound. Each site's existing timeouts are preserved exactly; the four bare dials gain a 5 s connect timeout and nothing else. No per-request deadline was added to any streaming call: Endpoint::timeout is a per-request bound on time-to-first-response-headers for every request the channel carries, so a value picked for one short call would also be the header deadline for the whole-volume transfer sharing the dial. The new bound covers the TCP handshake only -- tonic hands connect_timeout to HttpConnector::set_connect_timeout. A peer that completes the handshake and then stalls in the TLS or HTTP/2 exchange is still unbounded at those four sites, as are the RPCs themselves. That is why the three ping_* helpers keep their outer tokio::time::timeout: replacing it with connect_timeout would have narrowed a whole-connect bound they already had. main.rs no longer re-declares GRPC_MAX_MESSAGE_SIZE and the three keepalive/window constants; it imports them from grpc_client.rs so the inbound server and the outgoing clients cannot drift apart. Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-24 00:44:18 +00:00 · 2026-09-20 19:41:37 -07:00
parent ddf009ffe7
commit ac876eef21
6 changed files with 396 additions and 180 deletions
@@ -12,7 +12,10 @@ use seaweed_volume::security::tls::{
 use seaweed_volume::security::{Guard, SigningKey};
 #[cfg(unix)]
 use seaweed_volume::server::debug::build_debug_router;
-use seaweed_volume::server::grpc_client::load_outgoing_grpc_tls;
+use seaweed_volume::server::grpc_client::{
+    GRPC_INITIAL_WINDOW_SIZE, GRPC_KEEPALIVE_INTERVAL, GRPC_KEEPALIVE_TIMEOUT,
+    GRPC_MAX_MESSAGE_SIZE, load_outgoing_grpc_tls,
+};
 use seaweed_volume::server::grpc_server::VolumeGrpcService;
 #[cfg(unix)]
 use seaweed_volume::server::profiling::CpuProfileSession;
@@ -31,10 +34,10 @@ type CpuProfileParam = Option<CpuProfileSession>;
 #[cfg(not(unix))]
 type CpuProfileParam = Option<()>;

-const GRPC_MAX_MESSAGE_SIZE: usize = 1 << 30;
-const GRPC_KEEPALIVE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);
-const GRPC_KEEPALIVE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
-const GRPC_INITIAL_WINDOW_SIZE: u32 = 16 * 1024 * 1024;
+// The two settings that only make sense for the inbound server. The rest of
+// this server's HTTP/2 tuning — keepalive, window sizes, message size — is
+// imported from `server::grpc_client` above, which is also what the outgoing
+// clients dial with, so the two directions cannot drift apart.
 const GRPC_MAX_HEADER_LIST_SIZE: u32 = 8 * 1024 * 1024;
 const GRPC_MAX_CONCURRENT_STREAMS: u32 = 1000;

@@ -1,16 +1,42 @@
+//! Construction of the volume server's *outgoing* gRPC clients: TLS material,
+//! endpoint tuning, dial bounds, and the three client constructors every call
+//! site goes through.
+//!
+//! The keepalive, window-size and message-size constants below are shared with
+//! the *inbound* server built in `main.rs`, which imports them from here rather
+//! than declaring its own. Changing one therefore changes both directions at
+//! once, which is deliberate: a volume server talks to its peers with the same
+//! HTTP/2 settings it offers them.
+
 use std::error::Error;
 use std::fmt;
 use std::time::Duration;

 use hyper::http::Uri;
+use tonic::service::interceptor::InterceptedService;
 use tonic::transport::{Certificate, Channel, ClientTlsConfig, Endpoint, Identity};
+use tonic::{Request, Status};

 use crate::config::VolumeServerConfig;
+use crate::pb::filer_pb::seaweed_filer_client::SeaweedFilerClient;
+use crate::pb::master_pb::seaweed_client::SeaweedClient;
+use crate::pb::volume_server_pb::volume_server_client::VolumeServerClient;
+use crate::server::request_id::outgoing_request_id_interceptor;

 pub const GRPC_MAX_MESSAGE_SIZE: usize = 1 << 30;
-const GRPC_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(60);
-const GRPC_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);
-const GRPC_INITIAL_WINDOW_SIZE: u32 = 16 * 1024 * 1024;
+pub const GRPC_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(60);
+pub const GRPC_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);
+pub const GRPC_INITIAL_WINDOW_SIZE: u32 = 16 * 1024 * 1024;
+
+/// Bound on the TCP connect of every outgoing dial. `build_grpc_endpoint` is
+/// private and `connect_channel` is the only way out of this module, so every
+/// call site picks this up whether it thinks about timeouts or not.
+///
+/// It bounds the TCP handshake only — tonic hands it to
+/// `HttpConnector::set_connect_timeout`. A peer that completes the handshake
+/// and then stalls in the TLS or HTTP/2 exchange is not covered; callers that
+/// need that bound wrap the whole dial (see `connect_ping_target`).
+const GRPC_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);

 #[derive(Clone, Debug)]
 pub struct OutgoingGrpcTlsConfig {
@@ -81,7 +107,7 @@ pub fn grpc_endpoint_uri(grpc_host_port: &str, tls: Option<&OutgoingGrpcTlsConfi
    format!("{}://{}", scheme, grpc_host_port)
 }

-pub fn build_grpc_endpoint(
+fn build_grpc_endpoint(
    grpc_host_port: &str,
    tls: Option<&OutgoingGrpcTlsConfig>,
 ) -> Result<Endpoint, GrpcClientError> {
@@ -149,6 +175,152 @@ pub async fn connect_guarded(
        .map_err(|e| GrpcClientError(format!("connect {} failed: {}", target, e)))
 }

+/// How a dial is bounded.
+///
+/// `connect_timeout` is handed to the TCP connector. `request_timeout` becomes
+/// [`Endpoint::timeout`], which tonic installs as a `GrpcTimeout` layer in
+/// front of *every* request the resulting channel carries — it is not a
+/// property of one call.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct GrpcDialOptions {
+    /// Bound on establishing the connection to the peer.
+    pub connect_timeout: Duration,
+    /// Deadline applied to each RPC on the channel, or `None` to leave them
+    /// unbounded.
+    pub request_timeout: Option<Duration>,
+}
+
+impl GrpcDialOptions {
+    /// A short request/response call: connect within 5 s, answer within 10 s.
+    pub fn unary() -> Self {
+        Self {
+            connect_timeout: GRPC_CONNECT_TIMEOUT,
+            request_timeout: Some(Duration::from_secs(10)),
+        }
+    }
+
+    /// A call the peer may take a while to answer: connect within 5 s, answer
+    /// within 30 s.
+    pub fn long() -> Self {
+        Self {
+            connect_timeout: GRPC_CONNECT_TIMEOUT,
+            request_timeout: Some(Duration::from_secs(30)),
+        }
+    }
+
+    /// A bounded connect with no deadline on the RPCs themselves.
+    ///
+    /// `request_timeout` must stay `None` here. [`Endpoint::timeout`] is not a
+    /// transfer budget: tonic layers it as a `GrpcTimeout` around the
+    /// response future, which resolves when the server's *first response
+    /// headers* arrive, so it bounds how long the peer may take to start
+    /// answering — per request, for every request the channel carries. A 10 s
+    /// value picked to suit one short call would therefore also be the header
+    /// deadline for the `VolumeCopy` that shares the dial, and a busy source
+    /// that takes longer than that to open its file would lose the whole copy.
+    /// `VolumeCopy`, `VolumeTailSender` and `VolumeEcShardsCopy` have never
+    /// carried one.
+    pub fn stream() -> Self {
+        Self {
+            connect_timeout: GRPC_CONNECT_TIMEOUT,
+            request_timeout: None,
+        }
+    }
+}
+
+/// Dial a peer and return a connected channel.
+///
+/// The error carries only the transport failure: every caller already wraps it
+/// with the address and the operation it was attempting.
+pub async fn connect_channel(
+    grpc_host_port: &str,
+    tls: Option<&OutgoingGrpcTlsConfig>,
+    opts: GrpcDialOptions,
+) -> Result<Channel, GrpcClientError> {
+    let mut endpoint =
+        build_grpc_endpoint(grpc_host_port, tls)?.connect_timeout(opts.connect_timeout);
+    if let Some(request_timeout) = opts.request_timeout {
+        endpoint = endpoint.timeout(request_timeout);
+    }
+    endpoint
+        .connect()
+        .await
+        .map_err(|e| GrpcClientError(e.to_string()))
+}
+
+/// Dial a copy/tail source and return a connected channel, re-validating every
+/// resolved address at connect time.
+///
+/// The guarded equivalent of [`connect_channel`]: same `opts` bounds, but the
+/// dial goes through [`connect_guarded`] so a source address that passed
+/// validation cannot be re-pointed by DNS between the check and the connect.
+/// The bounds are applied to the endpoint *before* delegating, so the
+/// `allow_untrusted` opt-out is timed too.
+///
+/// `target` is the caller-facing source address (the unparsed
+/// `"ip:port.grpcPort"` form), which is what the guard pins against; the error
+/// carries only the transport failure, as every caller already wraps it with
+/// the address and the operation it was attempting.
+pub async fn connect_channel_guarded(
+    grpc_host_port: &str,
+    target: &str,
+    tls: Option<&OutgoingGrpcTlsConfig>,
+    opts: GrpcDialOptions,
+    allow_untrusted: bool,
+) -> Result<Channel, GrpcClientError> {
+    let mut endpoint =
+        build_grpc_endpoint(grpc_host_port, tls)?.connect_timeout(opts.connect_timeout);
+    if let Some(request_timeout) = opts.request_timeout {
+        endpoint = endpoint.timeout(request_timeout);
+    }
+    connect_guarded(endpoint, target, allow_untrusted).await
+}
+
+/// The outgoing request-id interceptor as a concrete type, so the client
+/// aliases below can name it.
+pub type RequestIdInterceptor = fn(Request<()>) -> Result<Request<()>, Status>;
+
+/// A volume-server client with the request-id interceptor attached.
+pub type VolumeServerGrpcClient =
+    VolumeServerClient<InterceptedService<Channel, RequestIdInterceptor>>;
+/// A master client with the request-id interceptor attached.
+pub type MasterGrpcClient = SeaweedClient<InterceptedService<Channel, RequestIdInterceptor>>;
+/// A filer client with the request-id interceptor attached.
+pub type FilerGrpcClient = SeaweedFilerClient<InterceptedService<Channel, RequestIdInterceptor>>;
+
+/// Wrap a connected channel in a volume-server client that forwards the
+/// current request id and lifts both message-size limits.
+pub fn volume_server_client(channel: Channel) -> VolumeServerGrpcClient {
+    VolumeServerClient::with_interceptor(
+        channel,
+        outgoing_request_id_interceptor as RequestIdInterceptor,
+    )
+    .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
+    .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE)
+}
+
+/// Wrap a connected channel in a master client that forwards the current
+/// request id and lifts both message-size limits.
+pub fn master_client(channel: Channel) -> MasterGrpcClient {
+    SeaweedClient::with_interceptor(
+        channel,
+        outgoing_request_id_interceptor as RequestIdInterceptor,
+    )
+    .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
+    .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE)
+}
+
+/// Wrap a connected channel in a filer client that forwards the current
+/// request id and lifts both message-size limits.
+pub fn filer_client(channel: Channel) -> FilerGrpcClient {
+    SeaweedFilerClient::with_interceptor(
+        channel,
+        outgoing_request_id_interceptor as RequestIdInterceptor,
+    )
+    .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
+    .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE)
+}
+
 /// Parse a SeaweedFS server address (`"ip:port.grpcPort"` or
 /// `"ip:port"`) into the `host:grpcPort` form `build_grpc_endpoint`
 /// expects. With the trailing `.grpcPort` segment, that segment IS
@@ -169,9 +341,16 @@ pub fn parse_grpc_address(source: &str) -> Result<String, String> {

 #[cfg(test)]
 mod tests {
-    use super::{build_grpc_endpoint, grpc_endpoint_uri, load_outgoing_grpc_tls};
+    use super::{
+        GrpcDialOptions, build_grpc_endpoint, connect_channel, grpc_endpoint_uri,
+        load_outgoing_grpc_tls, volume_server_client,
+    };
    use crate::config::{NeedleMapKind, ReadMode, VolumeServerConfig};
+    use crate::pb::volume_server_pb;
    use crate::security::tls::TlsPolicy;
+    use crate::server::request_id::scope_request_id;
+    use std::sync::{Arc, Mutex};
+    use std::time::Duration;

    const TEST_CERT_PEM: &str = "-----BEGIN CERTIFICATE-----\nMIIBPDCB76ADAgECAhRuRPQgeAu43BT/M7EfAWSdapVdYDAFBgMrZXAwFDESMBAG\nA1UEAwwJbG9jYWxob3N0MB4XDTI2MDcwNTE2MTUyOVoXDTM2MDcwMjE2MTUyOVow\nFDESMBAGA1UEAwwJbG9jYWxob3N0MCowBQYDK2VwAyEAr/3bNIFI+8V32oCiY6y+\nXRFmZpdNQ2g//VtRkT+nQg+jUzBRMB0GA1UdDgQWBBTsy9tLf1zPiXCQfgci6zNi\ndEzRSjAfBgNVHSMEGDAWgBTsy9tLf1zPiXCQfgci6zNidEzRSjAPBgNVHRMBAf8E\nBTADAQH/MAUGAytlcANBAIvsdw0IbvOBBkb9cd7BfMJfIP9pQQrAL03pCRWJFnFh\nSysaLVgFXI4T078IiaM874oO+iB+5vNbWEpc7CkGow4=\n-----END CERTIFICATE-----\n";
    const TEST_KEY_PEM: &str = "-----BEGIN PRIVATE KEY-----\nMC4CAQAwBQYDK2VwBCIEIHbyn71Kk+Y7KT3sBctit7uZpErpoH6qDbFj6P8qGaZH\n-----END PRIVATE KEY-----\n";
@@ -385,4 +564,112 @@ mod tests {
        let endpoint = build_grpc_endpoint(&parse_grpc_address("::1:9333").unwrap(), None).unwrap();
        assert_eq!(endpoint.uri().port_u16(), Some(19333));
    }
+    /// A minimal HTTP/2 server that records the gRPC request headers it is
+    /// sent and answers every call with a trailers-only `unimplemented`. It is
+    /// enough to prove what a helper-built client puts on the wire, without
+    /// standing up the whole `VolumeServer` service behind a tonic server.
+    async fn serve_header_capture() -> (u16, Arc<Mutex<Option<String>>>) {
+        use hyper::service::service_fn;
+        use hyper_util::rt::{TokioExecutor, TokioIo};
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let seen: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
+        let captured = Arc::clone(&seen);
+
+        tokio::spawn(async move {
+            while let Ok((stream, _)) = listener.accept().await {
+                let captured = Arc::clone(&captured);
+                tokio::spawn(async move {
+                    let _ = hyper::server::conn::http2::Builder::new(TokioExecutor::new())
+                        .serve_connection(
+                            TokioIo::new(stream),
+                            service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
+                                let captured = Arc::clone(&captured);
+                                async move {
+                                    let value = req
+                                        .headers()
+                                        .get("x-amz-request-id")
+                                        .and_then(|v| v.to_str().ok())
+                                        .map(str::to_string);
+                                    *captured.lock().unwrap() = value;
+                                    Ok::<_, std::convert::Infallible>(
+                                        hyper::http::Response::builder()
+                                            .status(200)
+                                            .header("content-type", "application/grpc")
+                                            .header("grpc-status", "12")
+                                            .body(tonic::body::Body::empty())
+                                            .unwrap(),
+                                    )
+                                }
+                            }),
+                        )
+                        .await;
+                });
+            }
+        });
+
+        (port, seen)
+    }
+
+    #[tokio::test]
+    async fn test_helper_built_client_sends_the_scoped_request_id() {
+        let (port, seen) = serve_header_capture().await;
+
+        let channel = connect_channel(
+            &format!("127.0.0.1:{}", port),
+            None,
+            GrpcDialOptions::unary(),
+        )
+        .await
+        .expect("dial the header-capturing server");
+
+        let mut client = volume_server_client(channel);
+        // The interceptor has a request id to forward only inside a scope, so
+        // the call has to run inside one for this to test anything.
+        let _ = scope_request_id("REQUEST-ID-ON-THE-WIRE".to_string(), async move {
+            client
+                .ping(volume_server_pb::PingRequest {
+                    target: String::new(),
+                    target_type: String::new(),
+                })
+                .await
+        })
+        .await;
+
+        assert_eq!(
+            seen.lock().unwrap().as_deref(),
+            Some("REQUEST-ID-ON-THE-WIRE"),
+            "a client built by volume_server_client must carry the outgoing request id"
+        );
+    }
+
+    #[test]
+    fn test_dial_presets_match_the_call_sites_they_replace() {
+        assert_eq!(
+            GrpcDialOptions::unary().connect_timeout,
+            Duration::from_secs(5)
+        );
+        assert_eq!(
+            GrpcDialOptions::unary().request_timeout,
+            Some(Duration::from_secs(10))
+        );
+        assert_eq!(
+            GrpcDialOptions::long().connect_timeout,
+            Duration::from_secs(5)
+        );
+        assert_eq!(
+            GrpcDialOptions::long().request_timeout,
+            Some(Duration::from_secs(30))
+        );
+        assert_eq!(
+            GrpcDialOptions::stream().connect_timeout,
+            Duration::from_secs(5)
+        );
+        assert_eq!(
+            GrpcDialOptions::stream().request_timeout,
+            None,
+            "a streaming dial must not put a per-request deadline on the channel"
+        );
+    }
 }
@@ -14,7 +14,6 @@ use tonic::{Request, Response, Status, Streaming};

 use crate::pb::filer_pb;
 use crate::pb::master_pb;
-use crate::pb::master_pb::seaweed_client::SeaweedClient;
 use crate::pb::volume_server_pb;
 use crate::pb::volume_server_pb::volume_server_server::VolumeServer;
 use crate::storage::erasure_coding::ec_shard::{DATA_SHARDS_COUNT, ShardId, shard_id_try_from};
@@ -22,7 +21,10 @@ use crate::storage::needle::needle::{self, Needle};
 use crate::storage::types::*;
 use crate::storage::volume::VolumeSpec;

-use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint};
+use super::grpc_client::{
+    GrpcDialOptions, connect_channel, connect_channel_guarded, filer_client, master_client,
+    volume_server_client,
+};
 use super::volume_server::VolumeServerState;

 type BoxStream<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send + 'static>>;
@@ -337,20 +339,14 @@ impl VolumeGrpcService {
        let grpc_addr = parse_grpc_address(&master_url).map_err(|e| {
            Status::internal(format!("invalid master address {}: {}", master_url, e))
        })?;
-        let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref())
-            .map_err(|e| Status::internal(format!("master address {}: {}", master_url, e)))?
-            .connect_timeout(std::time::Duration::from_secs(5))
-            .timeout(std::time::Duration::from_secs(30));
-        let channel = endpoint
-            .connect()
-            .await
-            .map_err(|e| Status::internal(format!("connect to master {}: {}", master_url, e)))?;
-        let mut client = SeaweedClient::with_interceptor(
-            channel,
-            super::request_id::outgoing_request_id_interceptor,
+        let channel = connect_channel(
+            &grpc_addr,
+            self.state.outgoing_grpc_tls.as_ref(),
+            GrpcDialOptions::long(),
        )
-        .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-        .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+        .await
+        .map_err(|e| Status::internal(format!("connect to master {}: {}", master_url, e)))?;
+        let mut client = master_client(channel);
        client
            .volume_mark_readonly(master_pb::VolumeMarkReadonlyRequest {
                ip: info.ip.clone(),
@@ -1844,13 +1840,11 @@ impl VolumeServer for VolumeGrpcService {
            ))
        })?;

-        let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref())
-            .map_err(|e| {
-                Status::internal(format!("VolumeCopy volume {} parse source: {}", vid, e))
-            })?;
-        let channel = super::grpc_client::connect_guarded(
-            endpoint,
+        let channel = connect_channel_guarded(
+            &grpc_addr,
            source,
+            self.state.outgoing_grpc_tls.as_ref(),
+            GrpcDialOptions::stream(),
            self.state.allow_untrusted_remote_endpoints,
        )
        .await
@@ -1861,13 +1855,7 @@ impl VolumeServer for VolumeGrpcService {
            ))
        })?;

-        let mut client =
-            volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor(
-                channel,
-                super::request_id::outgoing_request_id_interceptor,
-            )
-            .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-            .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+        let mut client = volume_server_client(channel);

        // Get file status from source
        let vol_info = client
@@ -2932,23 +2920,17 @@ impl VolumeServer for VolumeGrpcService {
        let grpc_addr = parse_grpc_address(source)
            .map_err(|e| Status::internal(format!("invalid source address {}: {}", source, e)))?;

-        let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref())
-            .map_err(|e| Status::internal(format!("parse source: {}", e)))?;
-        let channel = super::grpc_client::connect_guarded(
-            endpoint,
+        let channel = connect_channel_guarded(
+            &grpc_addr,
            source,
+            self.state.outgoing_grpc_tls.as_ref(),
+            GrpcDialOptions::stream(),
            self.state.allow_untrusted_remote_endpoints,
        )
        .await
        .map_err(|e| Status::internal(format!("connect to {}: {}", grpc_addr, e)))?;

-        let mut client =
-            volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor(
-                channel,
-                super::request_id::outgoing_request_id_interceptor,
-            )
-            .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-            .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+        let mut client = volume_server_client(channel);

        // Call VolumeTailSender on source
        let mut stream = client
@@ -3449,16 +3431,11 @@ impl VolumeServer for VolumeGrpcService {
            ))
        })?;

-        let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref())
-            .map_err(|e| {
-                Status::internal(format!(
-                    "VolumeEcShardsCopy volume {} parse source: {}",
-                    vid, e
-                ))
-            })?;
-        let channel = super::grpc_client::connect_guarded(
-            endpoint,
+        let channel = connect_channel_guarded(
+            &grpc_addr,
            source,
+            self.state.outgoing_grpc_tls.as_ref(),
+            GrpcDialOptions::stream(),
            self.state.allow_untrusted_remote_endpoints,
        )
        .await
@@ -3469,13 +3446,7 @@ impl VolumeServer for VolumeGrpcService {
            ))
        })?;

-        let mut client =
-            volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor(
-                channel,
-                super::request_id::outgoing_request_id_interceptor,
-            )
-            .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-            .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+        let mut client = volume_server_client(channel);

        // Copy each shard
        for &shard_id in &shard_ids {
@@ -5414,13 +5385,26 @@ impl VolumeServer for VolumeGrpcService {
    }
 }

-/// Build a gRPC endpoint from a SeaweedFS server address.
-fn to_grpc_endpoint(
+/// Dial a ping target, bounding the whole connect at 5s.
+///
+/// The outer timeout is not redundant with `GrpcDialOptions`' connect timeout:
+/// tonic hands that one to the HTTP connector, so it bounds the TCP dial only.
+/// A ping to a TLS peer that accepts the connection and then stalls in the
+/// handshake needs this wrapper to come back at all.
+async fn connect_ping_target(
    target: &str,
    tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>,
-) -> Result<tonic::transport::Endpoint, String> {
+) -> Result<tonic::transport::Channel, String> {
    let grpc_host_port = parse_grpc_address(target)?;
-    build_grpc_endpoint(&grpc_host_port, tls).map_err(|e| e.to_string())
+    // Ping is unary, but `stream()` is still right: these three have no
+    // per-request deadline today, and `unary()` would add a 10 s one.
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        connect_channel(&grpc_host_port, tls, GrpcDialOptions::stream()),
+    )
+    .await
+    .map_err(|_| "connection timeout".to_string())?
+    .map_err(|e| e.to_string())
 }

 /// Ping a remote volume server target by actually calling its Ping RPC (matches Go behavior).
@@ -5428,18 +5412,7 @@ async fn ping_volume_server_target(
    target: &str,
    tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>,
 ) -> Result<i64, String> {
-    let endpoint = to_grpc_endpoint(target, tls)?;
-    let channel = tokio::time::timeout(std::time::Duration::from_secs(5), endpoint.connect())
-        .await
-        .map_err(|_| "connection timeout".to_string())?
-        .map_err(|e| e.to_string())?;
-
-    let mut client = volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor(
-        channel,
-        super::request_id::outgoing_request_id_interceptor,
-    )
-    .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-    .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+    let mut client = volume_server_client(connect_ping_target(target, tls).await?);
    let resp = client
        .ping(volume_server_pb::PingRequest {
            target: String::new(),
@@ -5455,18 +5428,7 @@ async fn ping_master_target(
    target: &str,
    tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>,
 ) -> Result<i64, String> {
-    let endpoint = to_grpc_endpoint(target, tls)?;
-    let channel = tokio::time::timeout(std::time::Duration::from_secs(5), endpoint.connect())
-        .await
-        .map_err(|_| "connection timeout".to_string())?
-        .map_err(|e| e.to_string())?;
-
-    let mut client = master_pb::seaweed_client::SeaweedClient::with_interceptor(
-        channel,
-        super::request_id::outgoing_request_id_interceptor,
-    )
-    .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-    .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+    let mut client = master_client(connect_ping_target(target, tls).await?);
    let resp = client
        .ping(master_pb::PingRequest {
            target: String::new(),
@@ -5482,18 +5444,7 @@ async fn ping_filer_target(
    target: &str,
    tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>,
 ) -> Result<i64, String> {
-    let endpoint = to_grpc_endpoint(target, tls)?;
-    let channel = tokio::time::timeout(std::time::Duration::from_secs(5), endpoint.connect())
-        .await
-        .map_err(|_| "connection timeout".to_string())?
-        .map_err(|e| e.to_string())?;
-
-    let mut client = filer_pb::seaweed_filer_client::SeaweedFilerClient::with_interceptor(
-        channel,
-        super::request_id::outgoing_request_id_interceptor,
-    )
-    .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-    .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+    let mut client = filer_client(connect_ping_target(target, tls).await?);
    let resp = client
        .ping(filer_pb::PingRequest::default())
        .await
@@ -5997,6 +5948,7 @@ mod tests {
    use crate::config::MinFreeSpace;
    use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig, global_s3_tier_registry};
    use crate::security::{Guard, SigningKey};
+    use crate::server::grpc_client::GRPC_MAX_MESSAGE_SIZE;
    use crate::storage::needle_map::NeedleMapKind;
    use crate::storage::store::Store;
    use std::sync::RwLock;
@@ -16,7 +16,7 @@ use axum::response::{IntoResponse, Response};
 use serde::{Deserialize, Serialize};

 use super::absolute_display_path;
-use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint};
+use super::grpc_client::{GrpcDialOptions, connect_channel, volume_server_client};
 use super::volume_server::{VolumeServerState, normalize_outgoing_http_url, to_http_address};
 use crate::config::ReadMode;
 use crate::metrics;
@@ -541,19 +541,16 @@ async fn batch_delete_file_ids(
    }

    for (grpc_addr, batch) in server_to_file_ids {
-        let endpoint = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref())
-            .map_err(|e| format!("batch delete {}: {}", grpc_addr, e))?;
-        let channel = endpoint
-            .connect()
-            .await
-            .map_err(|e| format!("batch delete {}: {}", grpc_addr, e))?;
-        let mut client =
-            volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor(
-                channel,
-                super::request_id::outgoing_request_id_interceptor,
-            )
-            .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-            .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+        // BatchDelete is unary, but `stream()` is still right: this fan-out has
+        // no per-request deadline today, and `unary()` would add a 10 s one.
+        let channel = connect_channel(
+            &grpc_addr,
+            state.outgoing_grpc_tls.as_ref(),
+            GrpcDialOptions::stream(),
+        )
+        .await
+        .map_err(|e| format!("batch delete {}: {}", grpc_addr, e))?;
+        let mut client = volume_server_client(channel);

        let response = client
            .batch_delete(volume_server_pb::BatchDeleteRequest {
@@ -12,10 +12,9 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::broadcast;
 use tracing::{error, info, warn};

-use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint};
+use super::grpc_client::{GrpcDialOptions, connect_channel, master_client};
 use super::volume_server::VolumeServerState;
 use crate::pb::master_pb;
-use crate::pb::master_pb::seaweed_client::SeaweedClient;
 use crate::pb::volume_server_pb;
 use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig};
 use crate::storage::store::Store;
@@ -245,17 +244,8 @@ pub async fn try_get_master_configuration(
    grpc_addr: &str,
    tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>,
 ) -> Result<master_pb::GetMasterConfigurationResponse, Box<dyn std::error::Error>> {
-    let channel = build_grpc_endpoint(grpc_addr, tls)?
-        .connect_timeout(Duration::from_secs(5))
-        .timeout(Duration::from_secs(10))
-        .connect()
-        .await?;
-    let mut client = SeaweedClient::with_interceptor(
-        channel,
-        super::request_id::outgoing_request_id_interceptor,
-    )
-    .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-    .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+    let channel = connect_channel(grpc_addr, tls, GrpcDialOptions::unary()).await?;
+    let mut client = master_client(channel);
    let resp = client
        .get_master_configuration(master_pb::GetMasterConfigurationRequest {})
        .await?;
@@ -390,18 +380,14 @@ async fn do_heartbeat(
    pulse: Duration,
    shutdown_rx: &mut broadcast::Receiver<()>,
 ) -> Result<Option<String>, Box<dyn std::error::Error>> {
-    let channel = build_grpc_endpoint(grpc_addr, state.outgoing_grpc_tls.as_ref())?
-        .connect_timeout(Duration::from_secs(5))
-        .timeout(Duration::from_secs(30))
-        .connect()
-        .await?;
-
-    let mut client = SeaweedClient::with_interceptor(
-        channel,
-        super::request_id::outgoing_request_id_interceptor,
+    let channel = connect_channel(
+        grpc_addr,
+        state.outgoing_grpc_tls.as_ref(),
+        GrpcDialOptions::long(),
    )
-    .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-    .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+    .await?;
+
+    let mut client = master_client(channel);

    let (tx, rx) = tokio::sync::mpsc::channel::<master_pb::Heartbeat>(32);

@@ -38,12 +38,11 @@ use reed_solomon_erasure::galois_8::ReedSolomon;
 use tokio::sync::Semaphore;
 use tonic::Request;

-use crate::pb::master_pb::{self, LookupEcVolumeRequest, seaweed_client::SeaweedClient};
-use crate::pb::volume_server_pb::{
-    CopyFileRequest, VolumeEcShardReadRequest, volume_server_client::VolumeServerClient,
+use crate::pb::master_pb::{self, LookupEcVolumeRequest};
+use crate::pb::volume_server_pb::{CopyFileRequest, VolumeEcShardReadRequest};
+use crate::server::grpc_client::{
+    GrpcDialOptions, connect_channel, master_client, parse_grpc_address, volume_server_client,
 };
-use crate::server::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint, parse_grpc_address};
-use crate::server::request_id::outgoing_request_id_interceptor;
 use crate::server::volume_server::{VolumeServerState, to_http_address};
 use crate::storage::erasure_coding::ec_shard::{ShardId, shard_id_try_from};
 use crate::storage::needle::needle::{Needle, NeedleError, get_actual_size};
@@ -868,18 +867,15 @@ async fn cached_lookup_ec_shard_locations(

    let grpc_addr =
        parse_grpc_address(&master).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
-    let endpoint = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref())
-        .map_err(|e| io::Error::other(e.to_string()))?;
-    let channel = endpoint
-        .connect_timeout(Duration::from_secs(5))
-        .timeout(Duration::from_secs(10))
-        .connect()
-        .await
-        .map_err(|e| io::Error::other(format!("master connect: {}", e)))?;
+    let channel = connect_channel(
+        &grpc_addr,
+        state.outgoing_grpc_tls.as_ref(),
+        GrpcDialOptions::unary(),
+    )
+    .await
+    .map_err(|e| io::Error::other(format!("master connect: {}", e)))?;

-    let mut client = SeaweedClient::with_interceptor(channel, outgoing_request_id_interceptor)
-        .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-        .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+    let mut client = master_client(channel);

    let resp = client
        .lookup_ec_volume(Request::new(LookupEcVolumeRequest { volume_id: vid.0 }))
@@ -1060,14 +1056,13 @@ async fn do_read_remote_ec_shard_interval(
    } = iv;
    let grpc_addr =
        parse_grpc_address(source).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
-    let endpoint = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref())
-        .map_err(|e| io::Error::other(e.to_string()))?;
-    let channel = endpoint
-        .connect_timeout(Duration::from_secs(5))
-        .timeout(Duration::from_secs(30))
-        .connect()
-        .await
-        .map_err(|e| io::Error::other(format!("connect to {}: {}", source, e)))?;
+    let channel = connect_channel(
+        &grpc_addr,
+        state.outgoing_grpc_tls.as_ref(),
+        GrpcDialOptions::long(),
+    )
+    .await
+    .map_err(|e| io::Error::other(format!("connect to {}: {}", source, e)))?;

    // TODO(grpc-jwt): clusters with `jwt.signing.key` configured will
    // reject peer-to-peer VolumeEcShardRead calls until the Rust
@@ -1077,9 +1072,7 @@ async fn do_read_remote_ec_shard_interval(
    // here in isolation would split the credential plumbing across
    // call sites. Re-visit when outgoing JWT signing lands as a
    // server-wide helper.
-    let mut client = VolumeServerClient::with_interceptor(channel, outgoing_request_id_interceptor)
-        .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-        .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+    let mut client = volume_server_client(channel);

    let req = VolumeEcShardReadRequest {
        volume_id: vid.0,
@@ -1467,16 +1460,14 @@ async fn fetch_ec_index_from_one_peer(
 ) -> io::Result<()> {
    let grpc_addr =
        parse_grpc_address(peer).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
-    let channel = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref())
-        .map_err(|e| io::Error::other(e.to_string()))?
-        .connect_timeout(Duration::from_secs(5))
-        .timeout(Duration::from_secs(30))
-        .connect()
-        .await
-        .map_err(|e| io::Error::other(format!("connect {}: {}", peer, e)))?;
-    let mut client = VolumeServerClient::with_interceptor(channel, outgoing_request_id_interceptor)
-        .max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
-        .max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
+    let channel = connect_channel(
+        &grpc_addr,
+        state.outgoing_grpc_tls.as_ref(),
+        GrpcDialOptions::long(),
+    )
+    .await
+    .map_err(|e| io::Error::other(format!("connect {}: {}", peer, e)))?;
+    let mut client = volume_server_client(channel);

    let copy_req = |ext: &str, ignore_not_found: bool| CopyFileRequest {
        volume_id: m.vid.0,