From d1a35f8b8ef46d821e03346f858390773e995d1d Mon Sep 17 00:00:00 2001 From: bradleyshep Date: Wed, 30 Sep 2026 12:15:10 -0400 Subject: [PATCH] Reuse one HTTP client per replica for procedure requests `InstanceEnv::http_request` built a new reqwest client for every `ctx.http.fetch`, so no connection or TLS session was ever reused, and a module sending to an HTTP/2 service like APNs paid a full TCP and TLS handshake per request (the TODO(perf) this resolves). The client now lives on the ReplicaContext, built on first use and shared by the replica's module instances. It keeps the same DNS filter and redirect policy; per-request timeouts stay on the request. A failed build still surfaces as an HTTP error to the procedure and is retried on the next request. --- crates/core/src/host/host_controller.rs | 1 + crates/core/src/host/instance_env.rs | 116 +++++++++++++++++++----- crates/core/src/replica_context.rs | 4 + tools/ci/commands/test/src/main.rs | 15 +++ 4 files changed, 112 insertions(+), 24 deletions(-) diff --git a/crates/core/src/host/host_controller.rs b/crates/core/src/host/host_controller.rs index 55e52350a41..9721e539095 100644 --- a/crates/core/src/host/host_controller.rs +++ b/crates/core/src/host/host_controller.rs @@ -1004,6 +1004,7 @@ async fn make_replica_ctx( subscriptions, module_instance_memory_tracker, module_http, + module_http_client: Default::default(), }) } diff --git a/crates/core/src/host/instance_env.rs b/crates/core/src/host/instance_env.rs index 99e6806c236..e1a3e8fad2d 100644 --- a/crates/core/src/host/instance_env.rs +++ b/crates/core/src/host/instance_env.rs @@ -1039,22 +1039,14 @@ impl InstanceEnv { return Err(NodesError::HttpError(BLOCKED_HTTP_ADDRESS_ERROR.to_string())); } - let redirect_policy = reqwest::redirect::Policy::custom(|attempt| { - if is_blocked_ip_literal(attempt.url()) { - attempt.error(BLOCKED_HTTP_ADDRESS_ERROR) - } else { - reqwest::redirect::Policy::default().redirect(attempt) - } - }); - // TODO(procedure-metrics): record size in bytes of response, time spent awaiting response. // Actually execute the HTTP request! - // TODO(perf): Stash a long-lived `Client` in the env somewhere, rather than building a new one for each call. - let execute_fut = reqwest::Client::builder() - .dns_resolver(Arc::new(FilteredDnsResolver)) - .redirect(redirect_policy) - .build() + // All of the replica's requests share one client, and so one connection pool. + let execute_fut = self + .replica_ctx + .module_http_client + .get_or_try_init(build_module_http_client) .map_err(http_error)? .execute(reqwest); @@ -1131,6 +1123,23 @@ const HTTP_MAX_TIMEOUT: Duration = Duration::from_secs(180); const MODULE_HTTP_DISABLED_ERROR: &str = "module outbound HTTP requests are disabled"; const BLOCKED_HTTP_ADDRESS_ERROR: &str = "refusing to connect to private or special-purpose addresses"; +/// Build the client [`InstanceEnv::http_request`] sends requests with, +/// which refuses to connect to private or special-purpose addresses, +/// whether they come from DNS or from an IP literal in a redirect. +fn build_module_http_client() -> reqwest::Result { + let redirect_policy = reqwest::redirect::Policy::custom(|attempt| { + if is_blocked_ip_literal(attempt.url()) { + attempt.error(BLOCKED_HTTP_ADDRESS_ERROR) + } else { + reqwest::redirect::Policy::default().redirect(attempt) + } + }); + reqwest::Client::builder() + .dns_resolver(Arc::new(FilteredDnsResolver)) + .redirect(redirect_policy) + .build() +} + struct FilteredDnsResolver; impl reqwest::dns::Resolve for FilteredDnsResolver { @@ -1497,6 +1506,7 @@ mod test { subscriptions: subs, module_instance_memory_tracker: ModuleInstanceMemoryTracker::new(Identity::ZERO, Arc::new(())), module_http, + module_http_client: Default::default(), }, runtime, )) @@ -2586,7 +2596,7 @@ mod test { /// otherwise blocked by [`is_blocked_ip`]. #[cfg(feature = "allow_loopback_http_for_tests")] fn spawn_encoded_body_server(encoding: &str, body: Vec) -> (u16, std::thread::JoinHandle) { - use std::io::{Read, Write}; + use std::io::Write; use std::net::TcpListener; let listener = TcpListener::bind(("127.0.0.1", 0)).expect("failed to bind test server"); @@ -2604,16 +2614,7 @@ mod test { ); let handle = std::thread::spawn(move || { let (mut stream, _) = listener.accept().expect("test server failed to accept"); - - // Read the request head. We never send a request body, so `\r\n\r\n` terminates it. - let mut request = Vec::new(); - let mut byte = [0u8; 1]; - while !request.ends_with(b"\r\n\r\n") { - match stream.read(&mut byte).expect("test server failed to read request") { - 0 => break, - _ => request.push(byte[0]), - } - } + let request = read_request_head(&mut stream); stream .write_all(head.as_bytes()) @@ -2628,6 +2629,53 @@ mod test { (port, handle) } + /// Read a request head, which `\r\n\r\n` terminates as our test requests have no body. + /// Returns an empty head if the client closed the connection first. + #[cfg(feature = "allow_loopback_http_for_tests")] + fn read_request_head(stream: &mut impl std::io::Read) -> Vec { + let mut request = Vec::new(); + let mut byte = [0u8; 1]; + while !request.ends_with(b"\r\n\r\n") { + match stream.read(&mut byte).expect("test server failed to read request") { + 0 => break, + _ => request.push(byte[0]), + } + } + request + } + + /// Spin up a keep-alive HTTP/1.1 server on loopback which answers `requests` requests, + /// on as many connections as the client opens. + /// + /// Returns the bound port, and a handle yielding how many connections it accepted. + #[cfg(feature = "allow_loopback_http_for_tests")] + fn spawn_keep_alive_server(requests: usize) -> (u16, std::thread::JoinHandle) { + use std::io::Write; + use std::net::TcpListener; + + let listener = TcpListener::bind(("127.0.0.1", 0)).expect("failed to bind test server"); + let port = listener + .local_addr() + .expect("failed to read test server address") + .port(); + let handle = std::thread::spawn(move || { + let (mut served, mut accepted) = (0, 0); + while served < requests { + let (mut stream, _) = listener.accept().expect("test server failed to accept"); + accepted += 1; + // Serve this connection until the client closes it or we're done. + while served < requests && !read_request_head(&mut stream).is_empty() { + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") + .expect("test server failed to write response"); + served += 1; + } + } + accepted + }); + (port, handle) + } + /// `GET http://127.0.0.1:{port}/` through [`InstanceEnv::http_request`], /// i.e. the same path a procedure's `ctx.http` call takes. #[cfg(feature = "allow_loopback_http_for_tests")] @@ -2776,4 +2824,24 @@ mod test { Ok(()) } + + /// Consecutive requests from a replica reuse one pooled connection, + /// rather than paying for a new connection (and TLS handshake) each time. + #[test] + #[cfg(feature = "allow_loopback_http_for_tests")] + fn http_requests_reuse_connection() -> Result<()> { + let db = relational_db()?; + let (mut env, runtime) = instance_env(db)?; + + let (port, server) = spawn_keep_alive_server(2); + for _ in 0..2 { + let (response, body) = get_via_http_request(&mut env, &runtime, port); + assert_eq!(response.code, 200); + assert_eq!(&body[..], b"ok"); + } + let accepted = server.join().expect("test server thread panicked"); + + assert_eq!(accepted, 1, "both requests should share one connection"); + Ok(()) + } } diff --git a/crates/core/src/replica_context.rs b/crates/core/src/replica_context.rs index 0a4dd1e0b6f..c9845c2c665 100644 --- a/crates/core/src/replica_context.rs +++ b/crates/core/src/replica_context.rs @@ -7,6 +7,7 @@ use crate::error::DBError; use crate::messages::control_db::Database; use crate::resource::ModuleInstanceMemoryTracker; use crate::subscription::module_subscription_actor::ModuleSubscriptions; +use once_cell::sync::OnceCell; use std::io; use std::ops::Deref; use std::sync::Arc; @@ -22,6 +23,9 @@ pub struct ReplicaContext { pub subscriptions: ModuleSubscriptions, pub module_instance_memory_tracker: ModuleInstanceMemoryTracker, pub module_http: ModuleHttpConfig, + /// The client for procedure HTTP requests, built on first use. + /// Shared by all of this replica's module instances, so they reuse connections. + pub module_http_client: OnceCell, } impl ReplicaContext { diff --git a/tools/ci/commands/test/src/main.rs b/tools/ci/commands/test/src/main.rs index 5a17850f7b0..62488b7baf5 100644 --- a/tools/ci/commands/test/src/main.rs +++ b/tools/ci/commands/test/src/main.rs @@ -57,6 +57,21 @@ fn main() -> Result<()> { "--test-threads=2", ) .run()?; + // Procedure HTTP tests that talk to a loopback server only build with this feature, + // which `cargo test --all` above doesn't enable. + cmd!( + "cargo", + "test", + "-p", + "spacetimedb-core", + "--lib", + "--features", + "allow_loopback_http_for_tests", + "--", + "--test-threads=2", + "host::instance_env::test::http_request", + ) + .run()?; // The SDK test harness uses the same child-process server guard as smoketests, // which expects release CLI/standalone binaries to already exist. if !use_prebuilt_runtime {