From 5ea4aeda62ff24adee2e6319027aebc29a6e537f Mon Sep 17 00:00:00 2001 From: Sama Setty Date: Thu, 4 Jun 2026 14:49:02 -0700 Subject: [PATCH] Share reqwest HTTP client --- .../src/client/reqwest_http_client.rs | 36 +++++--- codex-rs/exec-server/tests/http_request.rs | 83 +++++++++++++++++++ 2 files changed, 109 insertions(+), 10 deletions(-) diff --git a/codex-rs/exec-server/src/client/reqwest_http_client.rs b/codex-rs/exec-server/src/client/reqwest_http_client.rs index 9e22fc01d3..725aee6f6b 100644 --- a/codex-rs/exec-server/src/client/reqwest_http_client.rs +++ b/codex-rs/exec-server/src/client/reqwest_http_client.rs @@ -6,6 +6,7 @@ //! orchestrator has forwarded `http/request` over JSON-RPC use std::error::Error as StdError; +use std::sync::OnceLock; use std::time::Duration; use codex_app_server_protocol::JSONRPCErrorError; @@ -47,19 +48,28 @@ pub(crate) struct PendingReqwestHttpBodyStream { /// by the exec-server route and the local [`HttpClient`] backend. pub(crate) struct ReqwestHttpRequestRunner { client: reqwest::Client, + timeout: Option, } impl ReqwestHttpClient { - fn build_client(timeout_ms: Option) -> Result { - let builder = match timeout_ms { - None => reqwest::Client::builder(), - Some(timeout_ms) => { - reqwest::Client::builder().timeout(Duration::from_millis(timeout_ms)) - } - }; - build_reqwest_client_with_custom_ca(builder) + fn build_client() -> Result { + build_reqwest_client_with_custom_ca(reqwest::Client::builder()) .map_err(|error| ExecServerError::HttpRequest(error.to_string())) } + + fn shared_client() -> Result { + static CLIENT: OnceLock = OnceLock::new(); + + if let Some(client) = CLIENT.get() { + return Ok(client.clone()); + } + + let client = Self::build_client()?; + match CLIENT.set(client.clone()) { + Ok(()) => Ok(client), + Err(client) => Ok(CLIENT.get().cloned().unwrap_or(client)), + } + } } impl HttpClient for ReqwestHttpClient { @@ -112,9 +122,12 @@ impl HttpClient for ReqwestHttpClient { impl ReqwestHttpRequestRunner { pub(crate) fn new(timeout_ms: Option) -> Result { - let client = ReqwestHttpClient::build_client(timeout_ms) + let client = ReqwestHttpClient::shared_client() .map_err(|error| internal_error(error.to_string()))?; - Ok(Self { client }) + Ok(Self { + client, + timeout: timeout_ms.map(Duration::from_millis), + }) } pub(crate) async fn run( @@ -137,6 +150,9 @@ impl ReqwestHttpRequestRunner { let headers = Self::build_headers(params.headers)?; let mut request = self.client.request(method.clone(), url).headers(headers); + if let Some(timeout) = self.timeout { + request = request.timeout(timeout); + } if let Some(body) = params.body { request = request.body(body.into_inner()); } diff --git a/codex-rs/exec-server/tests/http_request.rs b/codex-rs/exec-server/tests/http_request.rs index f45261dad7..e9e1e2812c 100644 --- a/codex-rs/exec-server/tests/http_request.rs +++ b/codex-rs/exec-server/tests/http_request.rs @@ -6,16 +6,19 @@ use std::collections::BTreeMap; use std::io::ErrorKind; use std::time::Duration; +use anyhow::Context; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::JSONRPCMessage; use codex_app_server_protocol::JSONRPCNotification; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; +use codex_exec_server::HttpClient; use codex_exec_server::HttpHeader; use codex_exec_server::HttpRequestBodyDeltaNotification; use codex_exec_server::HttpRequestParams; use codex_exec_server::HttpRequestResponse; use codex_exec_server::InitializeParams; +use codex_exec_server::ReqwestHttpClient; use common::exec_server::ExecServerHarness; use common::exec_server::exec_server; use pretty_assertions::assert_eq; @@ -30,6 +33,8 @@ use tokio::net::TcpStream; use tokio::sync::oneshot; use tokio::time::timeout; +const TEST_TIMEOUT: Duration = Duration::from_secs(5); + /// HTTP request captured by the ad-hoc TCP server in these integration tests. #[derive(Debug)] struct CapturedHttpRequest { @@ -39,6 +44,64 @@ struct CapturedHttpRequest { body: Vec, } +/// What this tests: the local reqwest-backed HTTP client is shared across +/// calls, so repeated requests to the same origin can reuse one TCP connection. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn reqwest_http_client_reuses_keep_alive_connection() -> anyhow::Result<()> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let url = format!("http://{}/mcp?case=reuse", listener.local_addr()?); + + let first_task = tokio::spawn({ + let url = url.clone(); + async move { + ReqwestHttpClient + .http_request(HttpRequestParams { + method: "GET".to_string(), + url, + headers: Vec::new(), + body: None, + timeout_ms: Some(5_000), + request_id: "reuse-1".to_string(), + stream_response: false, + }) + .await + } + }); + + let first_request = accept_http_request(&listener).await?; + assert_eq!(first_request.request_line, "GET /mcp?case=reuse HTTP/1.1"); + let stream = respond_with_keep_alive(first_request.stream, "200 OK", b"first").await?; + let first_response = first_task.await??; + assert_eq!(first_response.body.into_inner(), b"first".to_vec()); + + let second_task = tokio::spawn({ + let url = url.clone(); + async move { + ReqwestHttpClient + .http_request(HttpRequestParams { + method: "GET".to_string(), + url, + headers: Vec::new(), + body: None, + timeout_ms: Some(5_000), + request_id: "reuse-2".to_string(), + stream_response: false, + }) + .await + } + }); + + let second_request = timeout(TEST_TIMEOUT, read_http_request(stream)) + .await + .context("second request should reuse the first TCP connection")??; + assert_eq!(second_request.request_line, "GET /mcp?case=reuse HTTP/1.1"); + respond_with_status_and_headers(second_request.stream, "200 OK", &[], b"second").await?; + let second_response = second_task.await??; + assert_eq!(second_response.body.into_inner(), b"second".to_vec()); + + Ok(()) +} + /// What this tests: a real exec-server websocket `http/request` performs one /// HTTP request through the runner and returns the complete response body in /// the JSON-RPC response. @@ -410,6 +473,10 @@ async fn wait_for_error_response( /// Accepts one HTTP/1.1 request and captures its wire-visible fields. async fn accept_http_request(listener: &TcpListener) -> anyhow::Result { let (stream, _) = timeout(Duration::from_secs(5), listener.accept()).await??; + read_http_request(stream).await +} + +async fn read_http_request(stream: TcpStream) -> anyhow::Result { let mut reader = BufReader::new(stream); let mut request_line = String::new(); @@ -445,6 +512,22 @@ async fn accept_http_request(listener: &TcpListener) -> anyhow::Result anyhow::Result { + let response = format!( + "HTTP/1.1 {status}\r\ncontent-type: text/plain\r\ncontent-length: {}\r\nconnection: keep-alive\r\n\r\n", + body.len(), + ); + stream.write_all(response.as_bytes()).await?; + stream.write_all(body).await?; + stream.flush().await?; + Ok(stream) +} + /// Writes a fixed-length HTTP response to the captured request stream. async fn respond_with_status_and_headers( mut stream: TcpStream,