diff --git a/crates/harness-gateway-client/src/catalog.rs b/crates/harness-gateway-client/src/catalog.rs index 0e65b8f3f..cd0c8c1d0 100644 --- a/crates/harness-gateway-client/src/catalog.rs +++ b/crates/harness-gateway-client/src/catalog.rs @@ -7,6 +7,7 @@ use runvil::model::{ModelCatalog, ModelId, ModelInfo, ThinkingMode}; use serde::Deserialize; use crate::failure::{malformed, transport_failure, with_detail}; +use crate::transport::shared_client; use crate::wire::classify::classify_http_failure; use crate::wire::stream::escape_controls; @@ -115,17 +116,6 @@ async fn read_error_body_bounded( Ok(escaped) } -/// Returns the process-wide catalog HTTP client, building it once on first use. -/// -/// A single reusable client (MODEL-018) lets catalog fetches share one -/// connection pool and transport configuration rather than each constructing a -/// throwaway client with its own pool. The returned handle is a cheap clone of -/// the shared client (its state is reference-counted internally). -fn catalog_client() -> reqwest::Client { - static CATALOG_CLIENT: std::sync::OnceLock = std::sync::OnceLock::new(); - CATALOG_CLIENT.get_or_init(reqwest::Client::new).clone() -} - /// Sends a bearer-authed GET through the shared client (MODEL-018) and /// returns the success response, classifying every failure the same way for /// each gateway endpoint: `Transport` (or `Timeout`) when the send fails, the @@ -135,7 +125,7 @@ fn catalog_client() -> reqwest::Client { /// the [`reqwest::Error`] to the message under the same timeout marking as a /// send failure. async fn get_authed(url: String, token: &str) -> std::result::Result { - let response = catalog_client() + let response = shared_client() .get(url) .bearer_auth(token) .send() diff --git a/crates/harness-gateway-client/src/search.rs b/crates/harness-gateway-client/src/search.rs index 4337b159c..a6f49b48a 100644 --- a/crates/harness-gateway-client/src/search.rs +++ b/crates/harness-gateway-client/src/search.rs @@ -14,6 +14,7 @@ use runvil::error::{Error, ErrorCode}; use runvil::tools::tool_error; use crate::config::{GatewayEndpoint, SecretString}; +use crate::transport::shared_client; /// The largest error body kept for diagnostics, in bytes. const MAX_ERROR_BODY: usize = 2000; @@ -142,7 +143,7 @@ impl GatewaySearch { timeout: Duration, ) -> GatewaySearch { GatewaySearch { - http: reqwest::Client::new(), + http: shared_client(), base_url: endpoint.url, key, timeout, diff --git a/crates/harness-gateway-client/src/transport.rs b/crates/harness-gateway-client/src/transport.rs index 7b68cfc10..37776293b 100644 --- a/crates/harness-gateway-client/src/transport.rs +++ b/crates/harness-gateway-client/src/transport.rs @@ -42,6 +42,18 @@ pub struct GatewayChat { max_response_bytes: u64, } +/// Returns the process-wide HTTP client, building it once on first use. +/// +/// Every request this crate sends goes through it (MODEL-018), so every +/// [`GatewayChat`], `GatewaySearch`, and catalog fetch shares one connection +/// pool however often the caller rebuilds them. The pool is keyed by scheme, +/// host, and port, so a gateway at a new address gets its own connections. +/// The handle is a reference-counted clone. +pub(crate) fn shared_client() -> reqwest::Client { + static CLIENT: std::sync::OnceLock = std::sync::OnceLock::new(); + CLIENT.get_or_init(reqwest::Client::new).clone() +} + /// Default longest wait for the next receive. const DEFAULT_RECEIVE_TIMEOUT: Duration = Duration::from_secs(120); /// Default response-body ceiling. @@ -93,7 +105,7 @@ impl GatewayChat { #[must_use] pub fn new(endpoint: GatewayEndpoint, key: SecretString) -> GatewayChat { GatewayChat { - transport: GatewayTransport::Http(reqwest::Client::new()), + transport: GatewayTransport::Http(shared_client()), base_url: endpoint.url, key: Some(key), receive_timeout: DEFAULT_RECEIVE_TIMEOUT, @@ -116,7 +128,7 @@ impl GatewayChat { #[must_use] pub fn keyless(endpoint: GatewayEndpoint) -> GatewayChat { GatewayChat { - transport: GatewayTransport::Http(reqwest::Client::new()), + transport: GatewayTransport::Http(shared_client()), base_url: endpoint.url, key: None, receive_timeout: DEFAULT_RECEIVE_TIMEOUT, diff --git a/crates/harness-gateway-client/src/transport/tests.rs b/crates/harness-gateway-client/src/transport/tests.rs index d13abc9ae..36547a295 100644 --- a/crates/harness-gateway-client/src/transport/tests.rs +++ b/crates/harness-gateway-client/src/transport/tests.rs @@ -8,6 +8,7 @@ use super::*; mod env; mod limits; +mod pool; mod streaming; /// Serves `app` on a loopback port and returns its `/v1` root. diff --git a/crates/harness-gateway-client/src/transport/tests/pool.rs b/crates/harness-gateway-client/src/transport/tests/pool.rs new file mode 100644 index 000000000..a0687f536 --- /dev/null +++ b/crates/harness-gateway-client/src/transport/tests/pool.rs @@ -0,0 +1,59 @@ +//! The shared connection pool: clients built separately reuse one +//! connection to the same gateway. + +use std::net::SocketAddr; +use std::sync::{Arc, Mutex, PoisonError}; + +use axum::Router; +use axum::extract::ConnectInfo; +use axum::routing::post; +use runvil::model::Message; + +use super::*; + +#[tokio::test] +async fn two_clients_built_separately_send_their_rounds_over_one_connection() { + let peers: Arc>> = Arc::default(); + let seen = Arc::clone(&peers); + let app = Router::new().route( + "/v1/chat/completions", + post(move |ConnectInfo(peer): ConnectInfo| { + let seen = Arc::clone(&seen); + async move { + seen.lock() + .unwrap_or_else(PoisonError::into_inner) + .push(peer); + ( + [(axum::http::header::CONTENT_TYPE, "text/event-stream")], + ok_stream(), + ) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); + }); + let endpoint = GatewayEndpoint::new(&format!("http://{addr}/v1")).expect("valid test endpoint"); + let key = SecretString::new("tok").expect("non-empty test key"); + + for _ in 0..2 { + GatewayChat::new(endpoint.clone(), key.clone()) + .complete(&[Message::user("hi")], None, &openai_options(), |_| {}) + .await + .expect("the mock gateway answers the round"); + } + + let peers = peers.lock().unwrap_or_else(PoisonError::into_inner).clone(); + assert_eq!(peers.len(), 2, "both rounds reach the gateway"); + assert_eq!( + peers[0], peers[1], + "the second client reuses the first client's connection" + ); +} diff --git a/crates/workshop/server/src/agents/broker.rs b/crates/workshop/server/src/agents/broker.rs index 7ab7dded9..b2abba747 100644 --- a/crates/workshop/server/src/agents/broker.rs +++ b/crates/workshop/server/src/agents/broker.rs @@ -5,18 +5,19 @@ //! //! The Workshop broker holds the server's [`Registry`] and reads the //! gateway through it on every call, as the search provider does, so a -//! replaced gateway serves the next round. It keeps the Gateway broker of -//! the last generation it served under. A model list first waits until -//! the menu's catalog holds a chat-capable model, so a conversation -//! launched before the Gateway's catalog arrives starts its run only once -//! it can bind one. +//! replaced gateway serves the next round. It builds a Gateway broker per +//! call; the Gateway client's shared connection pool makes that a few +//! clones and keeps connections alive across rounds. A model list first +//! waits until the menu's catalog holds a chat-capable model, so a +//! conversation launched before the Gateway's catalog arrives starts its +//! run only once it can bind one. //! //! Each round reads the dropdown's pick through the registry as it is //! sent, so a model picked between rounds serves the next one; a round //! already in flight finishes on its model. use std::num::NonZeroU32; -use std::sync::{Arc, Mutex, PoisonError}; +use std::sync::Arc; use harness::{BoxFuture, InferenceBroker}; use harness_gateway_client::{GatewayBroker, StreamDelta}; @@ -40,37 +41,22 @@ pub(super) struct WorkshopBroker { /// The subsystem registry the gateway and menu handles are read /// through. registry: Registry, - /// The broker built for the last generation served under, shared with - /// the model lists still waiting. - cached: Arc>>, } impl WorkshopBroker { /// Builds the broker over the server's subsystem registry. pub(super) fn new(registry: Registry) -> Self { - Self { - registry, - cached: Arc::new(Mutex::new(None)), - } + Self { registry } } - /// The Gateway broker for the current generation, built on the first - /// call under it. No usable gateway fails as `Unavailable` with the - /// code's fixed phrase, so no URL or key reaches the message. + /// A Gateway broker for the current generation. No usable gateway + /// fails as `Unavailable` with the code's fixed phrase, so no URL or + /// key reaches the message. fn current(&self) -> Result { let unavailable = ErrorCode::Unavailable; let gateway = usable_gateway(&self.registry) .ok_or_else(|| completion_error(unavailable, phrase(unavailable)))?; - // A poisoned lock holds a pair written whole by one store. - let mut cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); - if let Some((generation, broker)) = cached.as_ref() - && *generation == gateway.generation - { - return Ok(broker.clone()); - } - let broker = GatewayBroker::new(gateway.endpoint, gateway.key); - *cached = Some((gateway.generation, broker.clone())); - Ok(broker) + Ok(GatewayBroker::new(gateway.endpoint, gateway.key)) } /// Resolves once the menu's catalog holds a chat-capable model, at diff --git a/crates/workshop/server/src/agents/gateway.rs b/crates/workshop/server/src/agents/gateway.rs index 8543d092a..4030a1282 100644 --- a/crates/workshop/server/src/agents/gateway.rs +++ b/crates/workshop/server/src/agents/gateway.rs @@ -1,6 +1,6 @@ //! The one read of a usable gateway that the agent sessions' broker, -//! search provider, and launch refusal share: the generation, the API -//! root, and the bearer of the gateway the registry currently holds. +//! search provider, and launch refusal share: the API root and the bearer +//! of the gateway the registry currently holds. //! //! It reads no heartbeat health; a caller that needs reachability checks //! it itself. @@ -11,8 +11,6 @@ use workshop_registry::Registry; /// One gateway generation whose endpoint and key both build. pub(super) struct UsableGateway { - /// The generation the server assigned before publishing it. - pub(super) generation: u64, /// The gateway's OpenAI-compatible API root: its base URL with `/v1`. pub(super) endpoint: GatewayEndpoint, /// The bearer paired with the base URL. @@ -26,7 +24,6 @@ impl UsableGateway { let snapshot = handles.binding().snapshot(); let api_root = format!("{}/v1", snapshot.base_url().trim_end_matches('/')); Some(Self { - generation: snapshot.generation(), endpoint: GatewayEndpoint::new(&api_root).ok()?, key: SecretString::new(snapshot.api_key()).ok()?, }) diff --git a/crates/workshop/server/src/agents/search.rs b/crates/workshop/server/src/agents/search.rs index 5b40545a7..b0cd10df1 100644 --- a/crates/workshop/server/src/agents/search.rs +++ b/crates/workshop/server/src/agents/search.rs @@ -3,10 +3,9 @@ //! //! The provider holds the server's [`Registry`] and reads the gateway //! handles through it on every search, as the inference broker does, so a -//! replaced gateway serves the next search. It keeps the [`GatewaySearch`] -//! of the last generation it searched under and hands each search to it. - -use std::sync::{Mutex, PoisonError}; +//! replaced gateway serves the next search. It builds a [`GatewaySearch`] +//! per search; the Gateway client's shared connection pool makes that a +//! few clones and keeps connections alive across searches. use harness_gateway_client::GatewaySearch; use plugin_web::{SearchError, SearchErrorKind, SearchProvider, SearchQuery, SearchResults}; @@ -23,23 +22,17 @@ const NO_GATEWAY: &str = "request failed"; pub(super) struct GatewaySearchProvider { /// The subsystem registry the gateway handles are read through. registry: Registry, - /// The client built for the last generation searched under. - cached: Mutex>, } impl GatewaySearchProvider { /// Builds the provider over the server's subsystem registry. pub(super) fn new(registry: Registry) -> Self { - Self { - registry, - cached: Mutex::new(None), - } + Self { registry } } - /// The client for the current gateway generation, built on the first - /// search under it. A missing gateway registration, a gateway the - /// heartbeat reports down, or an endpoint or key that cannot be built - /// fails as transport with [`NO_GATEWAY`]. + /// A client for the current gateway generation. A missing gateway + /// registration, a gateway the heartbeat reports down, or an endpoint + /// or key that cannot be built fails as transport with [`NO_GATEWAY`]. fn client(&self) -> Result { let no_gateway = || SearchError::new(SearchErrorKind::Transport, NO_GATEWAY); let handles = self @@ -50,16 +43,7 @@ impl GatewaySearchProvider { return Err(no_gateway()); } let gateway = UsableGateway::read(&handles).ok_or_else(no_gateway)?; - // A poisoned lock holds a pair written whole by one store. - let mut cached = self.cached.lock().unwrap_or_else(PoisonError::into_inner); - if let Some((generation, client)) = cached.as_ref() - && *generation == gateway.generation - { - return Ok(client.clone()); - } - let client = GatewaySearch::new(gateway.endpoint, gateway.key); - *cached = Some((gateway.generation, client.clone())); - Ok(client) + Ok(GatewaySearch::new(gateway.endpoint, gateway.key)) } } diff --git a/vibe/2026-11/2026-11-03-2-drop-http-client-mutex-java-smell.md b/vibe/2026-11/2026-11-03-2-drop-http-client-mutex-java-smell.md new file mode 100644 index 000000000..ad745fb5c --- /dev/null +++ b/vibe/2026-11/2026-11-03-2-drop-http-client-mutex-java-smell.md @@ -0,0 +1,96 @@ +--- +name: Drop the HTTP client Mutex +overview: Workshop's inference broker and search provider each cache one Gateway client per gateway generation behind a Mutex, only to keep a reqwest::Client's connection pool alive across rounds. Share one process-wide reqwest::Client inside harness-gateway-client instead, so building a GatewayBroker or GatewaySearch is a few clones, and delete both caches. +todos: + - id: shared-client + content: "Turn catalog.rs's catalog_client into a crate-wide transport::shared_client and use it in GatewayChat::new, GatewayChat::keyless, GatewaySearch::with_timeout, and the catalog fetch; add a test that two separately built clients reuse one connection" + status: completed + - id: drop-broker-cache + content: "Remove WorkshopBroker's Arc>>; current() builds a GatewayBroker from the usable gateway on every call; update the module doc" + status: completed + - id: drop-search-cache + content: "Remove GatewaySearchProvider's Mutex>; client() builds a GatewaySearch on every search; update the module doc" + status: completed + - id: gates + content: "Run fmt, clippy (both halves), nextest for harness-gateway-client and the workshop crates, the docs gate, and the docs-claims scan" + status: completed +isProject: false +--- + +# Drop the HTTP client Mutex + +## Problem + +`crates/workshop/server/src/agents/broker.rs` holds +`cached: Arc>>`, and +`crates/workshop/server/src/agents/search.rs` holds +`cached: Mutex>`. Each stores the Gateway client +built for the last gateway generation and rebuilds it when the generation +changes. + +The caches exist for one reason: `GatewayChat::new`, `GatewayChat::keyless`, +and `GatewaySearch::with_timeout` each call `reqwest::Client::new()`, which +builds a fresh connection pool and TLS configuration. Without the cache, +every round would pay a new connection and lose keep-alive. + +## Rejected: a lock-free mutable cell + +`InferenceBroker` is `Send + Sync` +(`crates/harness-internal/runner/src/performers.rs`), and `WorkshopBroker` is +cloned into `models()` futures that run on any worker thread. `Cell` and +`RefCell` are `!Sync`, so they do not compile here. `ArcSwapOption` (already a +workspace dependency) would drop the `Mutex` but keep the `Arc`, the cache, +and the generation bookkeeping. + +## Rejected: build the broker where the gateway is published + +`workshop-gateway` would gain a dependency on `harness-gateway-client`, which +widens that crate's surface for a cache the next option makes unnecessary. + +## Chosen: one shared reqwest client + +`catalog.rs` already keeps a process-wide `OnceLock` +(`catalog_client`, MODEL-018). Generalize it to `transport::shared_client` +and use it everywhere the crate builds a client. A `reqwest::Client` clone is +an `Arc` clone, so `GatewayBroker::new` and `GatewaySearch::new` become cheap, +and the Workshop types build one per call from `UsableGateway`, which they +already read on every call. + +Facility order (AGENTS.md Principles): this reuses the existing shared-client +facility and extends it from the catalog fetch to every request the crate +sends, which serves every caller of `GatewayChat` and `GatewaySearch`, not +only Workshop. + +### Behavior + +- Every Gateway request the process sends shares one connection pool. reqwest + keys the pool by scheme, host, and port, so a replaced gateway at a new + address gets its own connections and the old ones close when idle. +- No per-client configuration is lost: the receive timeout, the byte cap, and + the search timeout live on the wrapper types and apply per request. +- A replaced gateway serves the next round and search as before, now without + any generation comparison. +- The shared client spans tokio runtimes when a process runs more than one. + Workshop runs one runtime; the catalog fetch already made this choice. + +### Tests + +- New, `harness-gateway-client`: two `GatewayChat` values built separately + against one mock gateway send their rounds over the same TCP connection + (the server records each request's peer address). +- Kept, `workshop-server`: the replacement tests in `broker-tests.rs` and + `search-tests.rs` still prove the next round and search reach the new + gateway. + +## Outcome + +- `GatewayChat::new`, `GatewayChat::keyless`, `GatewaySearch::with_timeout`, + and the catalog fetch all use `transport::shared_client`. +- `WorkshopBroker` and `GatewaySearchProvider` hold only the `Registry`; + `UsableGateway` lost its `generation` field, which had no other reader. +- The new pool test fails with `reqwest::Client::new()` restored in + `GatewayChat` (distinct peer ports) and passes with the shared client. +- Gates: fmt, both clippy runs, the docs gate, and docs-claims pass. The + non-Workshop suite passes (4126). The Workshop suite passes except + `realtime_relay::overload::stalled_browser_cleanup_is_bounded_after_gateway_disconnect`, + which fails identically on unmodified `master` and is unrelated.