fix(mcp): apply the upstream.* connection settings to the rmcp transport

crates/aisix-gateway/src/upstream_http.rs

@@ -259,7 +259,17 @@ mod tests {
                 Some(i) => &src[..i],
                 None => &src[..],
             };
+            // `shared_http_client()` in aisix-mcp is the sanctioned
+            // counterpart of `client_builder()` for rmcp's own reqwest
+            // line (rmcp pins 0.13; it gets the same
+            // `upstream_http::config()` values applied). The exemption is
+            // scoped to that one file so a bare rmcp client anywhere else
+            // still gets flagged.
+            let sanctioned_rmcp_site = file.ends_with("aisix-mcp/src/bridge.rs");
             for (n, line) in production.lines().enumerate() {
+                if sanctioned_rmcp_site && line.contains("rmcp_reqwest::Client::") {
+                    continue;
+                }
                 if line.contains("reqwest::Client::builder()")
                     || line.contains("reqwest::Client::new()")
                 {

crates/aisix-mcp/Cargo.toml

@@ -26,12 +26,15 @@ tracing.workspace = true
 http.workspace = true
 reqwest.workspace = true
 # rmcp 1.8.0 builds its transport on the reqwest 0.13 line (the workspace is
-# on 0.12) and does not re-export it. This renamed dep exists ONLY to name
+# on 0.12) and does not re-export it. This renamed dep resolves to the same
+# crate version rmcp links, and is used for two things: naming
 # `StreamableHttpError<reqwest::Error>` when downcasting a connect failure to
-# detect an upstream 401 — it resolves to the same crate version rmcp links,
-# so the downcast sees the identical type. Kept `default-features = false`:
-# only the `Error` type is used, never a client. The 401-invalidation
-# integration test fails loudly if these ever stop unifying.
+# detect an upstream 401 (the downcast sees the identical type), and building
+# `shared_http_client()` — the client handed to every rmcp transport with the
+# deployment's `upstream.*` settings applied. Kept `default-features = false`:
+# TLS comes in through rmcp's own `reqwest?/rustls` feature unification. The
+# 401-invalidation integration test fails loudly if the versions ever stop
+# unifying.
 rmcp-reqwest = { package = "reqwest", version = "0.13", default-features = false }
 # Token-cache keying: the client secret is folded into the cache key as a
 # SHA-256 digest so a rotated secret never reuses the previous secret's token.

crates/aisix-mcp/src/bridge.rs

@@ -33,6 +33,56 @@ use crate::error::McpError;
 /// indefinitely. Overridable per upstream via [`McpUpstream::with_timeout`].
 pub const DEFAULT_UPSTREAM_TIMEOUT: Duration = Duration::from_secs(30);
 
+/// Shared HTTP client for every rmcp streamable-http transport, with the
+/// deployment's `upstream.*` connection settings applied.
+///
+/// rmcp pins its own reqwest line (0.13 — see Cargo.toml), so the shared
+/// `aisix_gateway::client_builder()` (workspace reqwest) cannot be handed
+/// to it directly; this is the sanctioned second construction site that
+/// applies the SAME `upstream_http::config()` values to rmcp's reqwest.
+///
+/// What this changes vs rmcp's `default_http_client()`:
+/// - a connect timeout exists at all (rmcp sets none, so a black-holed
+///   upstream was bounded only by the coarse per-operation deadline —
+///   the actual bug this fixes);
+/// - connection POOLING turns ON. rmcp deliberately disables idle
+///   pooling (`pool_max_idle_per_host(0)`) to dodge ~40 ms delayed-ACK
+///   stalls on reused connections; here reuse wins — every other
+///   outbound path pools under `upstream.*` management, and per-call
+///   TCP+TLS handshakes to a remote MCP server cost far more than the
+///   stall rmcp avoids. An operator can restore rmcp's behaviour with
+///   `upstream.pool_max_idle_per_host: 0`;
+/// - the TCP keepalive triple moves from reqwest's default 15 s/15 s/3
+///   to the deployment's 60 s/30 s/5.
+///
+/// One client for all MCP upstreams = one shared pool, matching how the
+/// provider bridges share theirs (auth is injected per-request by the
+/// transport, never client-wide).
+fn shared_http_client() -> rmcp_reqwest::Client {
+    static CLIENT: std::sync::OnceLock<rmcp_reqwest::Client> = std::sync::OnceLock::new();
+    CLIENT
+        .get_or_init(|| {
+            let cfg = aisix_gateway::upstream_http::config();
+            let mut b = rmcp_reqwest::Client::builder()
+                .pool_idle_timeout(cfg.pool_idle_timeout)
+                .tcp_keepalive(cfg.tcp_keepalive);
+            if let Some(d) = cfg.connect_timeout {
+                b = b.connect_timeout(d);
+            }
+            if let Some(d) = cfg.tcp_keepalive_interval {
+                b = b.tcp_keepalive_interval(d);
+            }
+            if let Some(n) = cfg.tcp_keepalive_retries {
+                b = b.tcp_keepalive_retries(n);
+            }
+            if let Some(n) = cfg.pool_max_idle_per_host {
+                b = b.pool_max_idle_per_host(n);
+            }
+            b.build().unwrap_or_else(|_| rmcp_reqwest::Client::new())
+        })
+        .clone()
+}
+
 /// Header carrying the gateway-held key for `api_key` upstream auth.
 const API_KEY_HEADER: &str = "x-api-key";
 
@@ -223,9 +273,17 @@ impl RmcpBridge {
     /// timeout.
     pub async fn connect(upstream: &McpUpstream) -> Result<Self, McpError> {
         let establish = async {
+            // Every arm goes through `with_client(shared_http_client(), ..)`
+            // so the transport inherits the deployment's `upstream.*`
+            // connection settings — `from_uri`/`from_config` would build
+            // rmcp's own default client with none of them.
             let transport = match &upstream.auth {
-                McpAuth::None => StreamableHttpClientTransport::from_uri(upstream.url.clone()),
-                McpAuth::Bearer(token) => StreamableHttpClientTransport::from_config(
+                McpAuth::None => StreamableHttpClientTransport::with_client(
+                    shared_http_client(),
+                    StreamableHttpClientTransportConfig::with_uri(upstream.url.clone()),
+                ),
+                McpAuth::Bearer(token) => StreamableHttpClientTransport::with_client(
+                    shared_http_client(),
                     StreamableHttpClientTransportConfig::with_uri(upstream.url.clone())
                         .auth_header(token.clone()),
                 ),
@@ -241,14 +299,16 @@ impl RmcpBridge {
                     // header map, mirroring this module's redaction posture.
                     value.set_sensitive(true);
                     let headers = HashMap::from([(HeaderName::from_static(API_KEY_HEADER), value)]);
-                    StreamableHttpClientTransport::from_config(
+                    StreamableHttpClientTransport::with_client(
+                        shared_http_client(),
                         StreamableHttpClientTransportConfig::with_uri(upstream.url.clone())
                             .custom_headers(headers),
                     )
                 }
                 McpAuth::OAuth2(cfg) => {
                     let token = crate::oauth::get_or_fetch(cfg).await?;
-                    StreamableHttpClientTransport::from_config(
+                    StreamableHttpClientTransport::with_client(
+                        shared_http_client(),
                         StreamableHttpClientTransportConfig::with_uri(upstream.url.clone())
                             .auth_header(token),
                     )
@@ -459,6 +519,31 @@ impl McpBridge for EphemeralBridge {
 mod tests {
     use super::*;
 
+    /// The dial to an upstream that swallows SYNs must be cut by the
+    /// shared client's `upstream.connect_timeout_ms` (default 5 s) —
+    /// before the transport ran on rmcp's default reqwest, which has no
+    /// connect timeout, so the dial hung until the OUTER
+    /// `upstream.timeout` (12 s here; kernel SYN retries run ≈127 s).
+    /// 203.0.113.1 (TEST-NET-3) is reserved and unrouted, the standard
+    /// black-hole address; on networks that answer with an immediate
+    /// "unreachable" the call fails fast either way — the assertion is
+    /// the upper bound, which only the connect timeout guarantees.
+    #[tokio::test]
+    async fn connect_timeout_bounds_an_unreachable_upstream() {
+        let mut upstream = McpUpstream::new("http://203.0.113.1:81/mcp");
+        upstream.timeout = Duration::from_secs(12);
+        let started = std::time::Instant::now();
+        let err = RmcpBridge::connect(&upstream)
+            .await
+            .err()
+            .expect("dial to a black-holed upstream must fail");
+        assert!(
+            started.elapsed() < Duration::from_secs(8),
+            "dial was not bounded by connect_timeout: took {:?} ({err})",
+            started.elapsed()
+        );
+    }
+
     /// The hand-written Debug impls are the only guard between a credential
     /// and the logs; pin them so a `#[derive(Debug)]` regression fails loudly
     /// for every secret-bearing variant.