From a2652df49510a0c8681037c0e353a7f996ffbeb3 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 3 Nov 2026 21:59:19 +0200 Subject: [PATCH] feat(client): report checkout, write, header-wait and body-read phase timings Four new ClientEvent variants (the enum is non_exhaustive, so observers compile unchanged): - CheckoutCompleted: how long the pool took to hand out a connection (reused, newly connected with DNS and connect, or after waiting) - RequestWritten: writing the request head and body - ResponseHeadersReceived: from the request being written to the response head being read (server time plus network, as the client sees it) - ResponseBodyRead: reading the buffered response body Together with DnsCompleted and ConnectionCompleted an observer can now split a slow call into pool wait, connect, write, server wait and body read without per-request state on its side. --- src/client/observer.rs | 33 +++++++ src/client/rich.rs | 205 +++++++++++++++++++++++++++++++++++------ 2 files changed, 212 insertions(+), 26 deletions(-) diff --git a/src/client/observer.rs b/src/client/observer.rs index 2e7eb5f..f1b1e64 100644 --- a/src/client/observer.rs +++ b/src/client/observer.rs @@ -80,6 +80,39 @@ pub enum ClientEvent<'a> { request_id: u64, origin: ObservedOrigin<'a>, }, + /// The pool handed out a connection: reused, newly connected (DNS and connect included), or + /// after waiting for one. One per attempt. + CheckoutCompleted { + request_id: u64, + origin: ObservedOrigin<'a>, + duration: Duration, + reused: bool, + error: Option, + }, + /// The request head and body were written to the connection. + RequestWritten { + request_id: u64, + origin: ObservedOrigin<'a>, + duration: Duration, + error: Option, + }, + /// From the request being written to the response head being read: the server's time plus + /// the network, as the client sees it. + ResponseHeadersReceived { + request_id: u64, + origin: ObservedOrigin<'a>, + duration: Duration, + error: Option, + }, + /// The response body was read in full by the buffered client (`send`). Streaming responses + /// are read by the caller, which times them itself. + ResponseBodyRead { + request_id: u64, + origin: ObservedOrigin<'a>, + duration: Duration, + bytes: usize, + error: Option, + }, ResponseHeaders { request_id: u64, origin: ObservedOrigin<'a>, diff --git a/src/client/rich.rs b/src/client/rich.rs index 4441eae..7fe628c 100644 --- a/src/client/rich.rs +++ b/src/client/rich.rs @@ -889,6 +889,7 @@ impl Client { body, deadline, &trace.span, + trace.request_id, )?; let status = response.status(); @@ -938,31 +939,44 @@ impl Client { "http_client_body", "http.response.body.size" = tracing::field::Empty ); - loop { - if response.body_complete() { - break; - } - let remaining = remaining(deadline)?; - response.set_timeout(Some(self.inner.config.io_timeout.min(remaining)))?; - if bytes.len() == bytes.capacity() { - bytes.reserve(8 * 1024); - } - let filled = bytes.len(); - let allowed = (bytes.capacity() - filled).min(limit - filled.min(limit) + 1); - bytes.resize(filled + allowed, 0); - let read = response.read(&mut bytes[filled..])?; - bytes.truncate(filled + read); - if read == 0 { - break; + let body_started = Instant::now(); + let body_read = (|| -> io::Result<()> { + loop { + if response.body_complete() { + break; + } + let remaining = remaining(deadline)?; + response.set_timeout(Some(self.inner.config.io_timeout.min(remaining)))?; + if bytes.len() == bytes.capacity() { + bytes.reserve(8 * 1024); + } + let filled = bytes.len(); + let allowed = + (bytes.capacity() - filled).min(limit - filled.min(limit) + 1); + bytes.resize(filled + allowed, 0); + let read = response.read(&mut bytes[filled..])?; + bytes.truncate(filled + read); + if read == 0 { + break; + } + if bytes.len() > limit { + response.abandon_body(); + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("HTTP response body exceeds configured {limit}-byte limit"), + )); + } } - if bytes.len() > limit { - response.abandon_body(); - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("HTTP response body exceeds configured {limit}-byte limit"), - )); - } - } + Ok(()) + })(); + self.inner.observe(ClientEvent::ResponseBodyRead { + request_id: trace.request_id, + origin: observed_url(url), + duration: body_started.elapsed(), + bytes: bytes.len(), + error: body_read.as_ref().err().map(ClientErrorKind::classify), + }); + body_read?; span_record!(body_span, "http.response.body.size", bytes.len() as u64); drop(body_span); drop(response); @@ -1090,6 +1104,7 @@ impl Client { body, deadline, &trace.span, + trace.request_id, ); if I::ENABLED { if let Some(timing) = &timing { @@ -1398,8 +1413,10 @@ impl Client { body: &mut RequestBody, deadline: Instant, span: &trace::Span, + request_id: u64, ) -> io::Result { let _ = span; + let write_started = Instant::now(); if headers.contains_key(HOST) { return Err(io::Error::new( io::ErrorKind::InvalidInput, @@ -1454,15 +1471,30 @@ impl Client { } } let client = &mut lease.connection_mut().client; - client.write_request(request)?; + let written = client.write_request(request); drop(write_span); + self.inner.observe(ClientEvent::RequestWritten { + request_id, + origin: observed_url(url), + duration: write_started.elapsed(), + error: written.as_ref().err().map(ClientErrorKind::classify), + }); + written?; + let wait_started = Instant::now(); let wait_span = phase_span!( DEBUG, parent: span, "http_client_wait", "http.response.status_code" = tracing::field::Empty ); - let response = client.read_response()?; + let response = client.read_response(); + self.inner.observe(ClientEvent::ResponseHeadersReceived { + request_id, + origin: observed_url(url), + duration: wait_started.elapsed(), + error: response.as_ref().err().map(ClientErrorKind::classify), + }); + let response = response?; span_record!( wait_span, "http.response.status_code", @@ -1530,12 +1562,34 @@ impl Client { } } + /// Take a connection for `key` (reused, new, or after waiting), reporting how long that took. fn checkout( &self, key: &OriginKey, tls_config: Arc, deadline: Instant, trace: &RequestTrace, + ) -> io::Result { + let started = Instant::now(); + let lease = self.checkout_connection(key, tls_config, deadline, trace); + self.inner.observe(ClientEvent::CheckoutCompleted { + request_id: trace.request_id, + origin: observed_key(key), + duration: started.elapsed(), + reused: lease + .as_ref() + .is_ok_and(|lease| lease.reused_idle_connection), + error: lease.as_ref().err().map(ClientErrorKind::classify), + }); + lease + } + + fn checkout_connection( + &self, + key: &OriginKey, + tls_config: Arc, + deadline: Instant, + trace: &RequestTrace, ) -> io::Result { loop { let now = Instant::now(); @@ -2780,6 +2834,11 @@ mod tests { ClientEvent::RequestAbandoned { request_id, status, .. } => format!("abandoned:{request_id}:{}", status.as_u16()), + // Phase timings: asserted by `observer_reports_request_phase_timings`. + ClientEvent::CheckoutCompleted { .. } + | ClientEvent::RequestWritten { .. } + | ClientEvent::ResponseHeadersReceived { .. } + | ClientEvent::ResponseBodyRead { .. } => return, }; self.0.lock().unwrap().push(value); } @@ -3689,6 +3748,100 @@ mod tests { assert!(!joined.contains("secret")); } + #[derive(Default)] + struct PhaseObserver(StdMutex>); + + impl ClientObserver for PhaseObserver { + fn observe(&self, event: ClientEvent<'_>) { + let entry = match event { + ClientEvent::CheckoutCompleted { + request_id, + duration, + reused, + error, + .. + } => ( + format!("checkout:{request_id}:{reused}:{error:?}"), + duration, + ), + ClientEvent::RequestWritten { + request_id, + duration, + error, + .. + } => (format!("written:{request_id}:{error:?}"), duration), + ClientEvent::ResponseHeadersReceived { + request_id, + duration, + error, + .. + } => (format!("headers:{request_id}:{error:?}"), duration), + ClientEvent::ResponseBodyRead { + request_id, + duration, + bytes, + error, + .. + } => (format!("body:{request_id}:{bytes}:{error:?}"), duration), + _ => return, + }; + self.0.lock().unwrap().push(entry); + } + } + + #[test] + fn observer_reports_request_phase_timings() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + for _ in 0..2 { + let _ = read_head(&mut stream); + // the server "thinks" before answering: that is header wait, not write or body + thread::sleep(Duration::from_millis(60)); + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") + .unwrap(); + stream.flush().unwrap(); + } + }); + let observer = Arc::new(PhaseObserver::default()); + let client = Client::builder() + .observer(observer.clone()) + .build() + .unwrap(); + for _ in 0..2 { + let response = client + .get(&format!("http://127.0.0.1:{port}/phases")) + .unwrap() + .send() + .unwrap(); + assert_eq!(response.body(), b"ok"); + } + server.join().unwrap(); + + let events = observer.0.lock().unwrap().clone(); + let names: Vec<&str> = events.iter().map(|(name, _)| name.as_str()).collect(); + assert_eq!( + names, + [ + "checkout:1:false:None", + "written:1:None", + "headers:1:None", + "body:1:2:None", + "checkout:2:true:None", + "written:2:None", + "headers:2:None", + "body:2:2:None", + ] + ); + let duration = |name: &str| events.iter().find(|(n, _)| n == name).unwrap().1; + assert!(duration("headers:1:None") >= Duration::from_millis(60)); + assert!(duration("headers:2:None") >= Duration::from_millis(60)); + assert!(duration("written:2:None") < Duration::from_millis(60)); + assert!(duration("body:2:2:None") < Duration::from_millis(60)); + } + #[test] fn fully_consumed_streaming_response_reuses_the_connection() { let listener = TcpListener::bind("127.0.0.1:0").unwrap();