From 5e23bd3865f9df5594e5dcd459cccefbd5ed9a54 Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 24 Sep 2026 01:18:59 +0200 Subject: [PATCH 01/11] Fix client lint findings and pooled teardown Replace unsafe pooled-client extraction with owned, exactly-once teardown and a fallible live-client borrow. Preserve active-runtime and no-runtime close behaviour with focused regression tests. Make eligible client methods const, propagate ignored results, and simplify conditional code. Replace six independent tracing booleans with a compact operation selector and remove field-name suppression without changing the builder API. --- src/client/builder/codec.rs | 4 +- src/client/builder/connect.rs | 6 +- src/client/builder/lifecycle.rs | 16 ++-- src/client/builder/tracing.rs | 2 +- src/client/codec_config.rs | 2 +- src/client/config.rs | 21 ++--- src/client/connect_parts.rs | 6 +- src/client/error.rs | 2 +- src/client/hooks.rs | 22 +++--- src/client/messaging.rs | 16 +++- src/client/mod.rs | 1 + src/client/pool/config.rs | 2 +- src/client/pool/handle.rs | 2 +- src/client/pool/lease.rs | 16 ++-- src/client/pool/managed.rs | 54 ++++++------- src/client/pool/manager.rs | 2 +- src/client/pool/mod.rs | 2 +- src/client/pool/scheduler.rs | 16 ++-- src/client/preamble_exchange.rs | 2 +- src/client/response_stream.rs | 6 +- src/client/runtime.rs | 18 ++++- src/client/send_streaming.rs | 13 ++-- src/client/socket_option_methods.rs | 14 ++-- src/client/streaming.rs | 13 +++- src/client/tests/pool.rs | 61 ++++++++++++++- src/client/tracing_config.rs | 116 ++++++++++++++-------------- src/client/tracing_helpers.rs | 4 +- src/client/tracing_timing.rs | 57 ++++++++++++++ 28 files changed, 316 insertions(+), 180 deletions(-) create mode 100644 src/client/tracing_timing.rs diff --git a/src/client/builder/codec.rs b/src/client/builder/codec.rs index d28523a8a..83324ad6b 100644 --- a/src/client/builder/codec.rs +++ b/src/client/builder/codec.rs @@ -19,7 +19,7 @@ where /// let _ = builder; /// ``` #[must_use] - pub fn codec_config(mut self, codec_config: ClientCodecConfig) -> Self { + pub const fn codec_config(mut self, codec_config: ClientCodecConfig) -> Self { self.codec_config = codec_config; self } @@ -51,7 +51,7 @@ where /// let _ = builder; /// ``` #[must_use] - pub fn length_format(mut self, length_format: LengthFormat) -> Self { + pub const fn length_format(mut self, length_format: LengthFormat) -> Self { self.codec_config = self.codec_config.length_format(length_format); self } diff --git a/src/client/builder/connect.rs b/src/client/builder/connect.rs index 1dec4d8df..cdefae204 100644 --- a/src/client/builder/connect.rs +++ b/src/client/builder/connect.rs @@ -11,6 +11,7 @@ use crate::{ ClientError, WireframeClient, tracing_helpers::{connect_span, emit_timing_event}, + tracing_timing::ClientOperation, }, rewind_stream::RewindStream, serializer::Serializer, @@ -54,7 +55,10 @@ where addr: SocketAddr, ) -> Result, C>, ClientError> { let span = connect_span(&self.tracing_config, &addr.to_string()); - let timing_start = self.tracing_config.connect_timing.then(Instant::now); + let timing_start = self + .tracing_config + .timing_enabled(ClientOperation::Connect) + .then(Instant::now); async { let result = self.connect_inner(addr).await; diff --git a/src/client/builder/lifecycle.rs b/src/client/builder/lifecycle.rs index 1ffcbe86a..dafa56db1 100644 --- a/src/client/builder/lifecycle.rs +++ b/src/client/builder/lifecycle.rs @@ -47,15 +47,15 @@ where Fut: Future + Send + 'static, C2: Send + 'static, { - // Preserve on_error since it is not parameterized by C. - // on_disconnect must be cleared because its signature depends on C. - let on_error = self.lifecycle_hooks.on_error; + // Preserve the error hook since it is not parameterized by C. + // The disconnect hook must be cleared because its signature depends on C. + let error = self.lifecycle_hooks.error; builder_field_update!( self, lifecycle_hooks = LifecycleHooks { - on_connect: Some(Arc::new(move || Box::pin(f()))), - on_disconnect: None, - on_error, + connect: Some(Arc::new(move || Box::pin(f()))), + disconnect: None, + error, } ) } @@ -95,7 +95,7 @@ where F: Fn(C) -> Fut + Send + Sync + 'static, Fut: Future + Send + 'static, { - self.lifecycle_hooks.on_disconnect = Some(Arc::new(move |c| Box::pin(f(c)))); + self.lifecycle_hooks.disconnect = Some(Arc::new(move |c| Box::pin(f(c)))); self } @@ -124,7 +124,7 @@ where F: for<'a> Fn(&'a ClientError) -> Fut + Send + Sync + 'static, Fut: Future + Send + 'static, { - self.lifecycle_hooks.on_error = Some(Arc::new(move |e| Box::pin(f(e)))); + self.lifecycle_hooks.error = Some(Arc::new(move |e| Box::pin(f(e)))); self } } diff --git a/src/client/builder/tracing.rs b/src/client/builder/tracing.rs index 7bb94e97e..e6465f289 100644 --- a/src/client/builder/tracing.rs +++ b/src/client/builder/tracing.rs @@ -28,7 +28,7 @@ where /// let _ = builder; /// ``` #[must_use] - pub fn tracing_config(mut self, config: TracingConfig) -> Self { + pub const fn tracing_config(mut self, config: TracingConfig) -> Self { self.tracing_config = config; self } diff --git a/src/client/codec_config.rs b/src/client/codec_config.rs index 574426015..440bea5d6 100644 --- a/src/client/codec_config.rs +++ b/src/client/codec_config.rs @@ -72,7 +72,7 @@ impl ClientCodecConfig { /// assert_eq!(codec.length_format_value().endianness(), Endianness::Little); /// ``` #[must_use] - pub fn length_format(mut self, length_format: LengthFormat) -> Self { + pub const fn length_format(mut self, length_format: LengthFormat) -> Self { self.length_format = length_format; self } diff --git a/src/client/config.rs b/src/client/config.rs index c87bf92e3..d5a63f426 100644 --- a/src/client/config.rs +++ b/src/client/config.rs @@ -76,7 +76,7 @@ impl SocketOptions { /// assert_eq!(options, expected); /// ``` #[must_use] - pub fn nodelay(mut self, enabled: bool) -> Self { + pub const fn nodelay(mut self, enabled: bool) -> Self { self.nodelay = Some(enabled); self } @@ -95,11 +95,12 @@ impl SocketOptions { /// assert_eq!(options, expected); /// ``` #[must_use] - pub fn keepalive(mut self, duration: Option) -> Self { - self.keepalive = Some(match duration { - Some(value) => KeepAliveSetting::Duration(value), - None => KeepAliveSetting::Disabled, - }); + pub const fn keepalive(mut self, duration: Option) -> Self { + let Some(value) = duration else { + self.keepalive = Some(KeepAliveSetting::Disabled); + return self; + }; + self.keepalive = Some(KeepAliveSetting::Duration(value)); self } @@ -115,7 +116,7 @@ impl SocketOptions { /// assert_eq!(options, expected); /// ``` #[must_use] - pub fn send_buffer_size(mut self, size: u32) -> Self { + pub const fn send_buffer_size(mut self, size: u32) -> Self { self.send_buffer_size = Some(size); self } @@ -132,7 +133,7 @@ impl SocketOptions { /// assert_eq!(options, expected); /// ``` #[must_use] - pub fn recv_buffer_size(mut self, size: u32) -> Self { + pub const fn recv_buffer_size(mut self, size: u32) -> Self { self.recv_buffer_size = Some(size); self } @@ -149,7 +150,7 @@ impl SocketOptions { /// assert_eq!(options, expected); /// ``` #[must_use] - pub fn reuseaddr(mut self, enabled: bool) -> Self { + pub const fn reuseaddr(mut self, enabled: bool) -> Self { self.reuseaddr = Some(enabled); self } @@ -172,7 +173,7 @@ impl SocketOptions { not(target_os = "cygwin"), ))] #[must_use] - pub fn reuseport(mut self, enabled: bool) -> Self { + pub const fn reuseport(mut self, enabled: bool) -> Self { self.reuseport = Some(enabled); self } diff --git a/src/client/connect_parts.rs b/src/client/connect_parts.rs index 545c7e964..3b69b7a1d 100644 --- a/src/client/connect_parts.rs +++ b/src/client/connect_parts.rs @@ -98,7 +98,7 @@ where .read_buffer_mut() .reserve(initial_read_buffer_capacity); - let connection_state = if let Some(ref setup) = self.lifecycle_hooks.on_connect { + let connection_state = if let Some(ref setup) = self.lifecycle_hooks.connect { Some(setup().await) } else { None @@ -109,8 +109,8 @@ where serializer: self.serializer, codec_config, connection_state, - on_disconnect: self.lifecycle_hooks.on_disconnect, - on_error: self.lifecycle_hooks.on_error, + on_disconnect: self.lifecycle_hooks.disconnect, + on_error: self.lifecycle_hooks.error, request_hooks: self.request_hooks, tracing_config: self.tracing_config, correlation_counter: AtomicU64::new(1), diff --git a/src/client/error.rs b/src/client/error.rs index fa0c297d6..b9a1816bd 100644 --- a/src/client/error.rs +++ b/src/client/error.rs @@ -90,7 +90,7 @@ impl ClientError { /// Return `true` when this error leaves the underlying connection unsafe /// to reuse in the pool. #[must_use] - pub fn should_recycle_connection(&self) -> bool { + pub const fn should_recycle_connection(&self) -> bool { match self { Self::Wireframe(_) | Self::PreambleWrite(_) diff --git a/src/client/hooks.rs b/src/client/hooks.rs index bb9ef59c5..0b9119794 100644 --- a/src/client/hooks.rs +++ b/src/client/hooks.rs @@ -93,25 +93,21 @@ pub type ClientErrorHandler = /// This struct holds the optional lifecycle callbacks configured via the /// builder. It is used internally to pass hook configuration from the builder /// to the client runtime. -#[expect( - clippy::struct_field_names, - reason = "on_ prefix is idiomatic for callback fields" -)] pub(crate) struct LifecycleHooks { /// Callback invoked when the connection is established. - pub(crate) on_connect: Option>, + pub(crate) connect: Option>, /// Callback invoked when the connection is closed. - pub(crate) on_disconnect: Option>, + pub(crate) disconnect: Option>, /// Callback invoked when an error occurs. - pub(crate) on_error: Option, + pub(crate) error: Option, } impl Clone for LifecycleHooks { fn clone(&self) -> Self { Self { - on_connect: self.on_connect.clone(), - on_disconnect: self.on_disconnect.clone(), - on_error: self.on_error.clone(), + connect: self.connect.clone(), + disconnect: self.disconnect.clone(), + error: self.error.clone(), } } } @@ -119,9 +115,9 @@ impl Clone for LifecycleHooks { impl Default for LifecycleHooks { fn default() -> Self { Self { - on_connect: None, - on_disconnect: None, - on_error: None, + connect: None, + disconnect: None, + error: None, } } } diff --git a/src/client/messaging.rs b/src/client/messaging.rs index 4ed7268e5..5026de87e 100644 --- a/src/client/messaging.rs +++ b/src/client/messaging.rs @@ -12,6 +12,7 @@ use super::{ ClientError, runtime::ClientStream, tracing_helpers::{call_correlated_span, emit_timing_event, receive_span, send_envelope_span}, + tracing_timing::ClientOperation, }; use crate::{ app::Packet, @@ -126,7 +127,10 @@ where envelope.set_correlation_id(Some(correlation_id)); } - let timing_start = self.tracing_config.send_timing.then(Instant::now); + let timing_start = self + .tracing_config + .timing_enabled(ClientOperation::Send) + .then(Instant::now); self.serialize_and_send(&envelope, timing_start, |config, frame_bytes| { send_envelope_span(config, correlation_id, frame_bytes) }) @@ -213,7 +217,10 @@ where P: Packet + EncodeWith + DecodeWith, { let span = call_correlated_span(&self.tracing_config); - let timing_start = self.tracing_config.call_timing.then(Instant::now); + let timing_start = self + .tracing_config + .timing_enabled(ClientOperation::Call) + .then(Instant::now); self.call_correlated_inner(request, &span, timing_start) .instrument(span.clone()) @@ -267,7 +274,10 @@ where /// Internal helper for receiving and deserializing a frame. pub(crate) async fn receive_internal>(&mut self) -> Result { let span = receive_span(&self.tracing_config); - let timing_start = self.tracing_config.receive_timing.then(Instant::now); + let timing_start = self + .tracing_config + .timing_enabled(ClientOperation::Receive) + .then(Instant::now); self.receive_frame(&span, timing_start) .instrument(span.clone()) diff --git a/src/client/mod.rs b/src/client/mod.rs index cdeae7b08..1d12469eb 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -30,6 +30,7 @@ mod streaming; mod streaming_helpers; mod tracing_config; mod tracing_helpers; +mod tracing_timing; pub use builder::WireframeClientBuilder; pub use codec_config::ClientCodecConfig; diff --git a/src/client/pool/config.rs b/src/client/pool/config.rs index 00d5deaef..bf493c077 100644 --- a/src/client/pool/config.rs +++ b/src/client/pool/config.rs @@ -86,7 +86,7 @@ impl ClientPoolConfig { /// Set how blocked `PoolHandle`s are ordered when capacity returns. #[must_use] - pub fn fairness_policy(mut self, fairness_policy: PoolFairnessPolicy) -> Self { + pub const fn fairness_policy(mut self, fairness_policy: PoolFairnessPolicy) -> Self { self.fairness_policy = fairness_policy; self } diff --git a/src/client/pool/handle.rs b/src/client/pool/handle.rs index ad3ea3fb0..b34f69a15 100644 --- a/src/client/pool/handle.rs +++ b/src/client/pool/handle.rs @@ -51,7 +51,7 @@ where C: Send + 'static, { /// Bind a scheduler-assigned identity to the shared pool state. - pub(crate) fn new(inner: Arc>, handle_id: u64) -> Self { + pub(crate) const fn new(inner: Arc>, handle_id: u64) -> Self { Self { inner, handle_id } } diff --git a/src/client/pool/lease.rs b/src/client/pool/lease.rs index 48a602015..29a31b230 100644 --- a/src/client/pool/lease.rs +++ b/src/client/pool/lease.rs @@ -38,7 +38,7 @@ where C: Send + 'static, { /// Construct a lease while retaining the permit until drop. - pub(crate) fn new( + pub(crate) const fn new( slot: Arc>, permit: OwnedSemaphorePermit, release_inner: Option>>, @@ -51,7 +51,7 @@ where } /// Classify transport errors that make a pooled connection unsafe to reuse. - fn should_recycle(err: &ClientError) -> bool { err.should_recycle_connection() } + const fn should_recycle(err: &ClientError) -> bool { err.should_recycle_connection() } /// Run one operation against a checked-out connection and mark bad sockets. async fn dispatch_on_connection( @@ -75,7 +75,7 @@ where /// Returns [`ClientError`] when checkout, serialization, or transport I/O /// fails. pub async fn send>(&self, message: &M) -> Result<(), ClientError> { - self.dispatch_on_connection(async |conn| conn.send(message).await) + self.dispatch_on_connection(async |conn| conn.client_mut()?.send(message).await) .await } @@ -85,7 +85,7 @@ where /// /// Returns [`ClientError`] when checkout, decode, or transport I/O fails. pub async fn receive>(&self) -> Result { - self.dispatch_on_connection(async |conn| conn.receive().await) + self.dispatch_on_connection(async |conn| conn.client_mut()?.receive().await) .await } @@ -100,7 +100,7 @@ where Req: EncodeWith, Resp: DecodeWith, { - self.dispatch_on_connection(async |conn| conn.call(request).await) + self.dispatch_on_connection(async |conn| conn.client_mut()?.call(request).await) .await } @@ -114,7 +114,7 @@ where where M: Packet + EncodeWith, { - self.dispatch_on_connection(async |conn| conn.send_envelope(envelope).await) + self.dispatch_on_connection(async |conn| conn.client_mut()?.send_envelope(envelope).await) .await } @@ -127,7 +127,7 @@ where where M: Packet + DecodeWith, { - self.dispatch_on_connection(async |conn| conn.receive_envelope().await) + self.dispatch_on_connection(async |conn| conn.client_mut()?.receive_envelope().await) .await } @@ -141,7 +141,7 @@ where where M: Packet + EncodeWith + DecodeWith, { - self.dispatch_on_connection(async |conn| conn.call_correlated(request).await) + self.dispatch_on_connection(async |conn| conn.client_mut()?.call_correlated(request).await) .await } } diff --git a/src/client/pool/managed.rs b/src/client/pool/managed.rs index e0a041951..f56c912de 100644 --- a/src/client/pool/managed.rs +++ b/src/client/pool/managed.rs @@ -4,14 +4,13 @@ //! `bb8` and guarantees connection teardown hooks still run when the pooled //! connection is dropped or reaped. -use std::{ - mem::ManuallyDrop, - ops::{Deref, DerefMut}, -}; - use tokio::{net::TcpStream, runtime::Handle}; -use crate::{client::WireframeClient, rewind_stream::RewindStream, serializer::Serializer}; +use crate::{ + client::{ClientError, WireframeClient}, + rewind_stream::RewindStream, + serializer::Serializer, +}; /// One physical client connection managed by `bb8`. pub(crate) struct ManagedClientConnection @@ -19,8 +18,8 @@ where S: Serializer + Send + Sync + 'static, C: Send + 'static, { - /// Manually dropped client so teardown can run on an available runtime. - client: ManuallyDrop, C>>, + /// Owned client, taken once when teardown starts. + client: Option, C>>, /// Set after protocol or I/O failure to force pool replacement. is_broken: bool, } @@ -31,36 +30,29 @@ where C: Send + 'static, { /// Wrap a connected client as a healthy pooled resource. - pub(crate) fn new(client: WireframeClient, C>) -> Self { + pub(crate) const fn new(client: WireframeClient, C>) -> Self { Self { - client: ManuallyDrop::new(client), + client: Some(client), is_broken: false, } } /// Mark the resource so `bb8` discards it instead of reusing it. - pub(crate) fn mark_broken(&mut self) { self.is_broken = true; } + pub(crate) const fn mark_broken(&mut self) { self.is_broken = true; } /// Report whether a prior operation invalidated this physical connection. pub(crate) const fn is_broken(&self) -> bool { self.is_broken } -} - -impl Deref for ManagedClientConnection -where - S: Serializer + Send + Sync + 'static, - C: Send + 'static, -{ - type Target = WireframeClient, C>; - fn deref(&self) -> &Self::Target { &self.client } -} - -impl DerefMut for ManagedClientConnection -where - S: Serializer + Send + Sync + 'static, - C: Send + 'static, -{ - fn deref_mut(&mut self) -> &mut Self::Target { &mut self.client } + /// Borrow the live client while the resource is checked out of the pool. + /// + /// # Errors + /// + /// Returns a disconnection error if teardown has already taken the client. + pub(crate) fn client_mut( + &mut self, + ) -> Result<&mut WireframeClient, C>, ClientError> { + self.client.as_mut().ok_or_else(ClientError::disconnected) + } } impl Drop for ManagedClientConnection @@ -69,9 +61,9 @@ where C: Send + 'static, { fn drop(&mut self) { - // SAFETY: `client` is valid here; `ManuallyDrop::take` is called - // exactly once, in `Drop`, before the memory is freed. - let client = unsafe { ManuallyDrop::take(&mut self.client) }; + let Some(client) = self.client.take() else { + return; + }; if let Ok(handle) = Handle::try_current() { handle.spawn(async move { diff --git a/src/client/pool/manager.rs b/src/client/pool/manager.rs index cdb0239fe..94b40b7e6 100644 --- a/src/client/pool/manager.rs +++ b/src/client/pool/manager.rs @@ -22,7 +22,7 @@ pub(crate) struct WireframeConnectionManager { impl WireframeConnectionManager { /// Capture the endpoint and recipe for one pool slot. - pub(crate) fn new(addr: SocketAddr, parts: ClientBuildParts) -> Self { + pub(crate) const fn new(addr: SocketAddr, parts: ClientBuildParts) -> Self { Self { addr, parts } } } diff --git a/src/client/pool/mod.rs b/src/client/pool/mod.rs index 4ebd8cdaa..6900cbf56 100644 --- a/src/client/pool/mod.rs +++ b/src/client/pool/mod.rs @@ -8,7 +8,7 @@ mod client_pool; mod config; mod handle; mod lease; -mod managed; +pub(super) mod managed; mod manager; mod policy; mod scheduler; diff --git a/src/client/pool/scheduler.rs b/src/client/pool/scheduler.rs index 9e5b0144f..23f016d0b 100644 --- a/src/client/pool/scheduler.rs +++ b/src/client/pool/scheduler.rs @@ -79,12 +79,12 @@ where true } PoolFairnessPolicy::RoundRobin => { - if let Some(queue) = self.handle_waiters.get_mut(&handle_id) { - queue.push_back(sender); - true - } else { - false - } + self.handle_waiters + .get_mut(&handle_id) + .is_some_and(|queue| { + queue.push_back(sender); + true + }) } } } @@ -204,7 +204,7 @@ where /// Complete every queued waiter with a disconnected error. pub(crate) fn notify_shutdown(&self) { while let Some(waiter) = self.next_waiter() { - let _ = waiter.send(Err(ClientError::disconnected())); + drop(waiter.send(Err(ClientError::disconnected()))); } } @@ -307,7 +307,7 @@ where }) } () = inner.shutdown_notified() => { - let _ = sender.send(Err(ClientError::disconnected())); + drop(sender.send(Err(ClientError::disconnected()))); return; } }; diff --git a/src/client/preamble_exchange.rs b/src/client/preamble_exchange.rs index 52cbce0b2..dda109396 100644 --- a/src/client/preamble_exchange.rs +++ b/src/client/preamble_exchange.rs @@ -54,7 +54,7 @@ where } /// Set the timeout for the preamble exchange. - pub(crate) fn set_timeout(&mut self, timeout: Duration) { self.timeout = Some(timeout); } + pub(crate) const fn set_timeout(&mut self, timeout: Duration) { self.timeout = Some(timeout); } /// Set the success handler. pub(crate) fn set_success_handler(&mut self, handler: H) diff --git a/src/client/response_stream.rs b/src/client/response_stream.rs index ce4e938cb..d751f68a8 100644 --- a/src/client/response_stream.rs +++ b/src/client/response_stream.rs @@ -102,15 +102,15 @@ where /// Returns the correlation identifier that frames are validated against. #[must_use] - pub fn correlation_id(&self) -> u64 { self.correlation_id } + pub const fn correlation_id(&self) -> u64 { self.correlation_id } /// Returns `true` if the stream has received the end-of-stream terminator. #[must_use] - pub fn is_terminated(&self) -> bool { self.terminated } + pub const fn is_terminated(&self) -> bool { self.terminated } /// Returns the number of data frames received so far. #[must_use] - pub fn frame_count(&self) -> usize { self.frame_count } + pub const fn frame_count(&self) -> usize { self.frame_count } /// Increment the frame counter and emit a per-frame tracing event for /// successfully decoded data frames. diff --git a/src/client/runtime.rs b/src/client/runtime.rs index 832ec02f6..8c452cb30 100644 --- a/src/client/runtime.rs +++ b/src/client/runtime.rs @@ -17,6 +17,7 @@ use super::{ hooks::{ClientConnectionTeardownHandler, ClientErrorHandler, RequestHooks}, tracing_config::TracingConfig, tracing_helpers::{call_span, close_span, emit_timing_event, send_span}, + tracing_timing::ClientOperation, }; use crate::{ message::{DecodeWith, EncodeWith}, @@ -152,7 +153,10 @@ where /// # } /// ``` pub async fn send>(&mut self, message: &M) -> Result<(), ClientError> { - let timing_start = self.tracing_config.send_timing.then(Instant::now); + let timing_start = self + .tracing_config + .timing_enabled(ClientOperation::Send) + .then(Instant::now); self.serialize_and_send(message, timing_start, send_span) .await } @@ -234,7 +238,10 @@ where request: &Req, ) -> Result { let span = call_span(&self.tracing_config); - let timing_start = self.tracing_config.call_timing.then(Instant::now); + let timing_start = self + .tracing_config + .timing_enabled(ClientOperation::Call) + .then(Instant::now); async { if let Err(err) = self.send(request).await { @@ -365,12 +372,15 @@ where /// ``` pub async fn close(mut self) { let span = close_span(&self.tracing_config); - let timing_start = self.tracing_config.close_timing.then(Instant::now); + let timing_start = self + .tracing_config + .timing_enabled(ClientOperation::Close) + .then(Instant::now); async { // Flush pending frames and send EOF before teardown. // Ignore errors since we're closing anyway. - let _ = self.framed.close().await; + drop(self.framed.close().await); if let (Some(state), Some(handler)) = (self.connection_state.take(), &self.on_disconnect) diff --git a/src/client/send_streaming.rs b/src/client/send_streaming.rs index 8bc890cf2..ec2cc1665 100644 --- a/src/client/send_streaming.rs +++ b/src/client/send_streaming.rs @@ -63,7 +63,7 @@ impl SendStreamingConfig { /// let config = SendStreamingConfig::default().with_chunk_size(8192); /// ``` #[must_use] - pub fn with_chunk_size(mut self, size: usize) -> Self { + pub const fn with_chunk_size(mut self, size: usize) -> Self { self.chunk_size = Some(size); self } @@ -84,7 +84,7 @@ impl SendStreamingConfig { /// let config = SendStreamingConfig::default().with_timeout(Duration::from_secs(10)); /// ``` #[must_use] - pub fn with_timeout(mut self, duration: Duration) -> Self { + pub const fn with_timeout(mut self, duration: Duration) -> Self { self.timeout = Some(duration); self } @@ -211,7 +211,7 @@ where Err(_elapsed) => { // Shut down the write side so buffered codec state cannot // leak into subsequent operations on this connection. - let _ = self.framed.get_mut().shutdown().await; + drop(self.framed.get_mut().shutdown().await); let err = ClientError::from(io::Error::new( io::ErrorKind::TimedOut, "streaming send timed out", @@ -311,10 +311,9 @@ fn effective_chunk_size( let available = max_frame_length - header_len; - let size = match config.chunk_size { - Some(requested) => requested.min(available), - None => available, - }; + let size = config + .chunk_size + .map_or(available, |requested| requested.min(available)); if size == 0 { return Err(ClientError::from(io::Error::new( diff --git a/src/client/socket_option_methods.rs b/src/client/socket_option_methods.rs index 65e80f97f..1fb0b8c21 100644 --- a/src/client/socket_option_methods.rs +++ b/src/client/socket_option_methods.rs @@ -25,7 +25,7 @@ where /// let _ = builder; /// ``` #[must_use] - pub fn socket_options(mut self, socket_options: SocketOptions) -> Self { + pub const fn socket_options(mut self, socket_options: SocketOptions) -> Self { self.socket_options = socket_options; self } @@ -41,7 +41,7 @@ where /// let _ = builder; /// ``` #[must_use] - pub fn nodelay(mut self, enabled: bool) -> Self { + pub const fn nodelay(mut self, enabled: bool) -> Self { self.socket_options = self.socket_options.nodelay(enabled); self } @@ -59,7 +59,7 @@ where /// let _ = builder; /// ``` #[must_use] - pub fn keepalive(mut self, duration: Option) -> Self { + pub const fn keepalive(mut self, duration: Option) -> Self { self.socket_options = self.socket_options.keepalive(duration); self } @@ -75,7 +75,7 @@ where /// let _ = builder; /// ``` #[must_use] - pub fn send_buffer_size(mut self, size: u32) -> Self { + pub const fn send_buffer_size(mut self, size: u32) -> Self { self.socket_options = self.socket_options.send_buffer_size(size); self } @@ -91,7 +91,7 @@ where /// let _ = builder; /// ``` #[must_use] - pub fn recv_buffer_size(mut self, size: u32) -> Self { + pub const fn recv_buffer_size(mut self, size: u32) -> Self { self.socket_options = self.socket_options.recv_buffer_size(size); self } @@ -107,7 +107,7 @@ where /// let _ = builder; /// ``` #[must_use] - pub fn reuseaddr(mut self, enabled: bool) -> Self { + pub const fn reuseaddr(mut self, enabled: bool) -> Self { self.socket_options = self.socket_options.reuseaddr(enabled); self } @@ -129,7 +129,7 @@ where not(target_os = "cygwin"), ))] #[must_use] - pub fn reuseport(mut self, enabled: bool) -> Self { + pub const fn reuseport(mut self, enabled: bool) -> Self { self.socket_options = self.socket_options.reuseport(enabled); self } diff --git a/src/client/streaming.rs b/src/client/streaming.rs index c0e57fa22..2dfcd6705 100644 --- a/src/client/streaming.rs +++ b/src/client/streaming.rs @@ -7,7 +7,13 @@ use std::{sync::atomic::Ordering, time::Instant}; -use super::{ClientError, ResponseStream, runtime::ClientStream, tracing_helpers::streaming_span}; +use super::{ + ClientError, + ResponseStream, + runtime::ClientStream, + tracing_helpers::streaming_span, + tracing_timing::ClientOperation, +}; use crate::{ app::Packet, message::{DecodeWith, EncodeWith}, @@ -82,7 +88,10 @@ where request.set_correlation_id(Some(correlation_id)); } - let timing_start = self.tracing_config.streaming_timing.then(Instant::now); + let timing_start = self + .tracing_config + .timing_enabled(ClientOperation::Streaming) + .then(Instant::now); self.serialize_and_send(&request, timing_start, |config, frame_bytes| { let span = streaming_span(config, correlation_id); diff --git a/src/client/tests/pool.rs b/src/client/tests/pool.rs index 816bc1690..c5f84f8c0 100644 --- a/src/client/tests/pool.rs +++ b/src/client/tests/pool.rs @@ -11,8 +11,10 @@ use std::{ use rstest::{fixture, rstest}; use tokio::time::{advance, timeout}; +use super::helpers::test_with_client; use crate::{ - client::{ClientError, ClientPoolConfig}, + client::{ClientError, ClientPoolConfig, pool::managed::ManagedClientConnection}, + serializer::BincodeSerializer, test_helpers::{ Ping, Pong, @@ -25,6 +27,63 @@ use crate::{ type TestResult = Result<(), Box>; +/// Keep teardown failures attributed to the test that dropped the connection. +#[track_caller] +fn assert_teardown_once(teardown_count: &AtomicUsize) { + assert_eq!( + teardown_count.load(Ordering::SeqCst), + 1, + "pooled connection teardown callback must run exactly once" + ); +} + +/// Build a real connected resource so its teardown path exercises socket close +/// and the callback together. +async fn managed_connection_with_teardown( + teardown_count: Arc, +) -> super::helpers::TestResult> { + let client = test_with_client(|builder| { + builder + .on_connection_setup(|| async {}) + .on_connection_teardown(move |()| { + let callback_count = Arc::clone(&teardown_count); + async move { + callback_count.fetch_add(1, Ordering::SeqCst); + } + }) + }) + .await?; + Ok(ManagedClientConnection::new(client)) +} + +#[tokio::test] +async fn managed_connection_drop_runs_teardown_on_active_runtime() -> TestResult { + let teardown_count = Arc::new(AtomicUsize::new(0)); + let managed = managed_connection_with_teardown(Arc::clone(&teardown_count)).await?; + + drop(managed); + timeout(Duration::from_secs(1), async { + while teardown_count.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }) + .await?; + assert_teardown_once(&teardown_count); + Ok(()) +} + +#[tokio::test] +async fn managed_connection_drop_runs_teardown_without_runtime() -> TestResult { + let teardown_count = Arc::new(AtomicUsize::new(0)); + let managed = managed_connection_with_teardown(Arc::clone(&teardown_count)).await?; + + std::thread::spawn(move || drop(managed)) + .join() + .map_err(|_| std::io::Error::other("pooled connection drop thread panicked"))?; + assert_teardown_once(&teardown_count); + Ok(()) +} + #[rustfmt::skip] #[fixture] fn client_pool_config() -> ClientPoolConfig { diff --git a/src/client/tracing_config.rs b/src/client/tracing_config.rs index 2fabefda2..0f7e33d38 100644 --- a/src/client/tracing_config.rs +++ b/src/client/tracing_config.rs @@ -5,6 +5,8 @@ use tracing::Level; +use super::tracing_timing::{ClientOperation, ClientTimingFlags}; + /// Controls tracing span levels and per-command timing for client operations. /// /// By default, lifecycle operations (`connect`, `close`) emit spans at @@ -35,10 +37,6 @@ use tracing::Level; /// .with_all_timing(true); /// let _ = verbose; /// ``` -#[expect( - clippy::struct_excessive_bools, - reason = "six independent on/off timing flags — one per operation category" -)] #[derive(Clone, Debug)] pub struct TracingConfig { /// Span level for connection establishment. @@ -53,18 +51,8 @@ pub struct TracingConfig { pub(crate) streaming_level: Level, /// Span level for orderly connection close. pub(crate) close_level: Level, - /// Emit elapsed microseconds for connection establishment. - pub(crate) connect_timing: bool, - /// Emit elapsed microseconds for one-way sends. - pub(crate) send_timing: bool, - /// Emit elapsed microseconds for one-way receives. - pub(crate) receive_timing: bool, - /// Emit elapsed microseconds for request-response calls. - pub(crate) call_timing: bool, - /// Emit elapsed microseconds for streaming calls. - pub(crate) streaming_timing: bool, - /// Emit elapsed microseconds for orderly connection close. - pub(crate) close_timing: bool, + /// Per-operation elapsed-time selection. + timing: ClientTimingFlags, } impl Default for TracingConfig { @@ -76,12 +64,7 @@ impl Default for TracingConfig { call_level: Level::DEBUG, streaming_level: Level::DEBUG, close_level: Level::INFO, - connect_timing: false, - send_timing: false, - receive_timing: false, - call_timing: false, - streaming_timing: false, - close_timing: false, + timing: ClientTimingFlags::default(), } } } @@ -99,7 +82,7 @@ impl TracingConfig { /// let _ = config; /// ``` #[must_use] - pub fn with_connect_level(mut self, level: Level) -> Self { + pub const fn with_connect_level(mut self, level: Level) -> Self { self.connect_level = level; self } @@ -118,28 +101,28 @@ impl TracingConfig { /// let _ = config; /// ``` #[must_use] - pub fn with_connect_timing(mut self, enabled: bool) -> Self { - self.connect_timing = enabled; + pub const fn with_connect_timing(mut self, enabled: bool) -> Self { + self.timing = self.timing.with_enabled(ClientOperation::Connect, enabled); self } /// Set the tracing level for `send` and `send_envelope` operations. #[must_use] - pub fn with_send_level(mut self, level: Level) -> Self { + pub const fn with_send_level(mut self, level: Level) -> Self { self.send_level = level; self } /// Enable or disable per-command timing for `send` and `send_envelope`. #[must_use] - pub fn with_send_timing(mut self, enabled: bool) -> Self { - self.send_timing = enabled; + pub const fn with_send_timing(mut self, enabled: bool) -> Self { + self.timing = self.timing.with_enabled(ClientOperation::Send, enabled); self } /// Set the tracing level for `receive` and `receive_envelope` operations. #[must_use] - pub fn with_receive_level(mut self, level: Level) -> Self { + pub const fn with_receive_level(mut self, level: Level) -> Self { self.receive_level = level; self } @@ -147,50 +130,52 @@ impl TracingConfig { /// Enable or disable per-command timing for `receive` and /// `receive_envelope`. #[must_use] - pub fn with_receive_timing(mut self, enabled: bool) -> Self { - self.receive_timing = enabled; + pub const fn with_receive_timing(mut self, enabled: bool) -> Self { + self.timing = self.timing.with_enabled(ClientOperation::Receive, enabled); self } /// Set the tracing level for `call` and `call_correlated` operations. #[must_use] - pub fn with_call_level(mut self, level: Level) -> Self { + pub const fn with_call_level(mut self, level: Level) -> Self { self.call_level = level; self } /// Enable or disable per-command timing for `call` and `call_correlated`. #[must_use] - pub fn with_call_timing(mut self, enabled: bool) -> Self { - self.call_timing = enabled; + pub const fn with_call_timing(mut self, enabled: bool) -> Self { + self.timing = self.timing.with_enabled(ClientOperation::Call, enabled); self } /// Set the tracing level for `call_streaming` operations. #[must_use] - pub fn with_streaming_level(mut self, level: Level) -> Self { + pub const fn with_streaming_level(mut self, level: Level) -> Self { self.streaming_level = level; self } /// Enable or disable per-command timing for `call_streaming`. #[must_use] - pub fn with_streaming_timing(mut self, enabled: bool) -> Self { - self.streaming_timing = enabled; + pub const fn with_streaming_timing(mut self, enabled: bool) -> Self { + self.timing = self + .timing + .with_enabled(ClientOperation::Streaming, enabled); self } /// Set the tracing level for the `close` operation. #[must_use] - pub fn with_close_level(mut self, level: Level) -> Self { + pub const fn with_close_level(mut self, level: Level) -> Self { self.close_level = level; self } /// Enable or disable per-command timing for the `close` operation. #[must_use] - pub fn with_close_timing(mut self, enabled: bool) -> Self { - self.close_timing = enabled; + pub const fn with_close_timing(mut self, enabled: bool) -> Self { + self.timing = self.timing.with_enabled(ClientOperation::Close, enabled); self } @@ -206,7 +191,7 @@ impl TracingConfig { /// let _ = config; /// ``` #[must_use] - pub fn with_all_levels(mut self, level: Level) -> Self { + pub const fn with_all_levels(mut self, level: Level) -> Self { self.connect_level = level; self.send_level = level; self.receive_level = level; @@ -227,15 +212,16 @@ impl TracingConfig { /// let _ = config; /// ``` #[must_use] - pub fn with_all_timing(mut self, enabled: bool) -> Self { - self.connect_timing = enabled; - self.send_timing = enabled; - self.receive_timing = enabled; - self.call_timing = enabled; - self.streaming_timing = enabled; - self.close_timing = enabled; + pub const fn with_all_timing(mut self, enabled: bool) -> Self { + self.timing = ClientTimingFlags::all(enabled); self } + + /// Report whether one client operation emits an elapsed-time event. + #[must_use] + pub(crate) const fn timing_enabled(&self, operation: ClientOperation) -> bool { + self.timing.is_enabled(operation) + } } #[cfg(test)] @@ -247,7 +233,7 @@ mod tests { use rstest::rstest; use tracing::Level; - use super::TracingConfig; + use super::{ClientOperation, TracingConfig}; /// Assert all level fields match the expected pattern. /// @@ -267,16 +253,20 @@ mod tests { /// Assert all timing flags match the expected pattern. /// /// Order: connect, send, receive, call, streaming, close. - fn assert_timing_flags( - cfg: &TracingConfig, - [connect, send, receive, call, streaming, close]: [bool; 6], - ) { - assert_eq!(cfg.connect_timing, connect, "connect_timing"); - assert_eq!(cfg.send_timing, send, "send_timing"); - assert_eq!(cfg.receive_timing, receive, "receive_timing"); - assert_eq!(cfg.call_timing, call, "call_timing"); - assert_eq!(cfg.streaming_timing, streaming, "streaming_timing"); - assert_eq!(cfg.close_timing, close, "close_timing"); + #[track_caller] + fn assert_timing_flags(cfg: &TracingConfig, expected: [bool; 6]) { + let actual = [ + cfg.timing_enabled(ClientOperation::Connect), + cfg.timing_enabled(ClientOperation::Send), + cfg.timing_enabled(ClientOperation::Receive), + cfg.timing_enabled(ClientOperation::Call), + cfg.timing_enabled(ClientOperation::Streaming), + cfg.timing_enabled(ClientOperation::Close), + ]; + assert_eq!( + actual, expected, + "timing flags are ordered connect, send, receive, call, streaming, close" + ); } #[test] @@ -321,6 +311,14 @@ mod tests { assert_timing_flags(&cfg, [false; 6]); } + #[test] + fn disabling_one_timing_flag_preserves_other_enabled_operations() { + let cfg = TracingConfig::default() + .with_all_timing(true) + .with_send_timing(false); + assert_timing_flags(&cfg, [true, false, true, true, true, true]); + } + #[rstest] #[case::connect( TracingConfig::default().with_connect_timing(true), diff --git a/src/client/tracing_helpers.rs b/src/client/tracing_helpers.rs index 2e13664f4..723da4288 100644 --- a/src/client/tracing_helpers.rs +++ b/src/client/tracing_helpers.rs @@ -147,8 +147,8 @@ pub(crate) fn close_span(config: &TracingConfig) -> Span { /// `Some(instant)` when enabled. When `Some`, an event is emitted with /// the `elapsed_us` field at `DEBUG` level. pub(crate) fn emit_timing_event(start: Option) { - if let Some(start) = start { - let elapsed_us = start.elapsed().as_micros(); + if let Some(started_at) = start { + let elapsed_us = started_at.elapsed().as_micros(); tracing::debug!(elapsed_us = elapsed_us, "operation.timing"); } } diff --git a/src/client/tracing_timing.rs b/src/client/tracing_timing.rs new file mode 100644 index 000000000..902ae1cd0 --- /dev/null +++ b/src/client/tracing_timing.rs @@ -0,0 +1,57 @@ +//! Compact timing switches for the six client operation categories. + +/// Client operation whose elapsed time can be recorded. +#[derive(Clone, Copy, Debug)] +pub(crate) enum ClientOperation { + /// Connection establishment. + Connect, + /// One-way send. + Send, + /// One-way receive. + Receive, + /// Request-response call. + Call, + /// Streaming call. + Streaming, + /// Orderly connection close. + Close, +} + +impl ClientOperation { + /// Return the independent bit reserved for this operation. + const fn mask(self) -> u8 { + match self { + Self::Connect => 0b00_0001, + Self::Send => 0b00_0010, + Self::Receive => 0b00_0100, + Self::Call => 0b00_1000, + Self::Streaming => 0b01_0000, + Self::Close => 0b10_0000, + } + } +} + +/// Timing selection stored as one value rather than six independent booleans. +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct ClientTimingFlags(u8); + +impl ClientTimingFlags { + /// Set one operation without changing the other five. + pub(crate) const fn with_enabled(mut self, operation: ClientOperation, enabled: bool) -> Self { + let mask = operation.mask(); + if enabled { + self.0 |= mask; + } else { + self.0 &= !mask; + } + self + } + + /// Set every operation to the same timing policy. + pub(crate) const fn all(enabled: bool) -> Self { Self(if enabled { 0b11_1111 } else { 0 }) } + + /// Report whether timing is enabled for one operation. + pub(crate) const fn is_enabled(self, operation: ClientOperation) -> bool { + self.0 & operation.mask() != 0 + } +} From 2eaae1012366c9b8825ac0ed68dd11442cedc533 Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 24 Sep 2026 02:26:26 +0200 Subject: [PATCH 02/11] Prove client timing and teardown contracts Strengthen the client lint wave with operation-specific timing assertions, state-transition property coverage, and const API compilation checks. Document the internal timing and pooled-client ownership boundaries. Use a test-only completion signal to prove both pooled teardown paths finish before asserting exactly-once cleanup, and make the shared pool setup an async rstest fixture. --- docs/developers-guide.md | 13 +++ docs/wireframe-client-design.md | 13 ++- src/client/pool/managed.rs | 27 ++++++ src/client/tests/pool.rs | 43 +++++---- src/client/tests/tracing.rs | 4 +- src/client/tests/tracing_streaming.rs | 5 +- src/client/tracing_timing.rs | 119 +++++++++++++++++++++++++ tests/compile_error.rs | 1 + tests/ui/client_const_api.rs | 120 ++++++++++++++++++++++++++ 9 files changed, 324 insertions(+), 21 deletions(-) create mode 100644 tests/ui/client_const_api.rs diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 50ac68dbf..b6f30622e 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -401,6 +401,19 @@ hooks, request hooks, and tracing configuration on the same path. Pooled lease methods should go through `PooledClientLease::dispatch_on_connection` so checkout and recycle-on-error policy stay in one place. +`ManagedClientConnection` owns its client in an `Option` and takes it once when +teardown starts. Calls through `client_mut` return a disconnection error after +that transfer. Dropping a live connection schedules `close()` on the current +Tokio runtime; without a current runtime, it builds a current-thread runtime +and runs `close()` synchronously so teardown hooks can complete. + +Client timing selection has six independent categories: connect, send, receive, +call, streaming, and close. Each `TracingConfig::with_*_timing` setter changes +only its category; `with_all_timing` changes all six. Keep these selectors +independent when adding or changing client operations. Internally, +`ClientOperation` identifies a category, and `ClientTimingFlags` stores one bit +per operation. + ## Error surface conventions Library-facing errors should stay typed and inspectable by default. Use diff --git a/docs/wireframe-client-design.md b/docs/wireframe-client-design.md index c438a43a5..5904906cf 100644 --- a/docs/wireframe-client-design.md +++ b/docs/wireframe-client-design.md @@ -302,7 +302,13 @@ high-frequency data operations (send, receive, call, streaming). When no **Per-command timing**: When enabled via `TracingConfig::with_*_timing(true)`, an additional `tracing::debug!` event recording `elapsed_us` is emitted when the operation completes. Timing events fire on both success and error paths. -Timing is disabled by default for all operations. +Timing is disabled by default for all operations. The six independent timing +categories are connect, send (`send`, `send_envelope`), receive (`receive`, +`receive_envelope`), call (`call`, `call_correlated`), streaming +(`call_streaming`), and close. Each per-operation setter changes only its own +category, leaving the other five unchanged. `with_all_timing` sets all six +categories together. Internally, `ClientOperation` identifies a category and +`ClientTimingFlags` stores one bit per operation. **Design rationale — async-safe span instrumentation**: `Span::enter()` guards must not be held across `.await` points because a multi-threaded runtime may @@ -463,6 +469,11 @@ cargo run --example client_echo_login --features examples - `PooledClientLease` forwards request methods (`send`, `receive`, `call`, `send_envelope`, `receive_envelope`, `call_correlated`) through its slot rather than dereferencing to a long-lived mutable `WireframeClient`. + - Each `ManagedClientConnection` owns its client in an `Option`. Teardown + takes that client exactly once, and later `client_mut` calls return a + disconnection error. Drop schedules `close()` on the current Tokio runtime; + without one, it creates a current-thread runtime and runs close + synchronously, allowing teardown hooks to run in either context. - Rationale: this keeps socket lifecycle, reconnect, and idle recycle on battle-tested `bb8` machinery while preserving explicit Wireframe control over how many operations may target one warm socket. diff --git a/src/client/pool/managed.rs b/src/client/pool/managed.rs index f56c912de..73b063a88 100644 --- a/src/client/pool/managed.rs +++ b/src/client/pool/managed.rs @@ -4,6 +4,8 @@ //! `bb8` and guarantees connection teardown hooks still run when the pooled //! connection is dropped or reaped. +#[cfg(test)] +use tokio::sync::oneshot; use tokio::{net::TcpStream, runtime::Handle}; use crate::{ @@ -12,6 +14,14 @@ use crate::{ serializer::Serializer, }; +/// Complete a test wait only after the close future has returned. +#[cfg(test)] +fn signal_close_completion(sender: Option>) { + if let Some(sender) = sender { + sender.send(()).unwrap_or_default(); + } +} + /// One physical client connection managed by `bb8`. pub(crate) struct ManagedClientConnection where @@ -22,6 +32,9 @@ where client: Option, C>>, /// Set after protocol or I/O failure to force pool replacement. is_broken: bool, + /// Signals tests only after the detached close future completes. + #[cfg(test)] + close_completion: Option>, } impl ManagedClientConnection @@ -34,9 +47,17 @@ where Self { client: Some(client), is_broken: false, + #[cfg(test)] + close_completion: None, } } + /// Install a test-only signal for completion of the detached close task. + #[cfg(test)] + pub(crate) fn notify_after_close(&mut self, sender: oneshot::Sender<()>) { + self.close_completion = Some(sender); + } + /// Mark the resource so `bb8` discards it instead of reusing it. pub(crate) const fn mark_broken(&mut self) { self.is_broken = true; } @@ -66,8 +87,12 @@ where }; if let Ok(handle) = Handle::try_current() { + #[cfg(test)] + let close_completion = self.close_completion.take(); handle.spawn(async move { client.close().await; + #[cfg(test)] + signal_close_completion(close_completion); }); return; } @@ -77,6 +102,8 @@ where .build() { runtime.block_on(client.close()); + #[cfg(test)] + signal_close_completion(self.close_completion.take()); } } } diff --git a/src/client/tests/pool.rs b/src/client/tests/pool.rs index c5f84f8c0..84db2e628 100644 --- a/src/client/tests/pool.rs +++ b/src/client/tests/pool.rs @@ -26,6 +26,10 @@ use crate::{ }; type TestResult = Result<(), Box>; +type ManagedFixture = ( + ManagedClientConnection, + Arc, +); /// Keep teardown failures attributed to the test that dropped the connection. #[track_caller] @@ -39,47 +43,52 @@ fn assert_teardown_once(teardown_count: &AtomicUsize) { /// Build a real connected resource so its teardown path exercises socket close /// and the callback together. -async fn managed_connection_with_teardown( - teardown_count: Arc, -) -> super::helpers::TestResult> { +#[fixture] +async fn managed_connection_with_teardown() -> super::helpers::TestResult { + let teardown_count = Arc::new(AtomicUsize::new(0)); + let teardown_callback_count = Arc::clone(&teardown_count); let client = test_with_client(|builder| { builder .on_connection_setup(|| async {}) .on_connection_teardown(move |()| { - let callback_count = Arc::clone(&teardown_count); + let callback_count = Arc::clone(&teardown_callback_count); async move { callback_count.fetch_add(1, Ordering::SeqCst); } }) }) .await?; - Ok(ManagedClientConnection::new(client)) + Ok((ManagedClientConnection::new(client), teardown_count)) } +#[rstest] #[tokio::test] -async fn managed_connection_drop_runs_teardown_on_active_runtime() -> TestResult { - let teardown_count = Arc::new(AtomicUsize::new(0)); - let managed = managed_connection_with_teardown(Arc::clone(&teardown_count)).await?; +async fn managed_connection_drop_runs_teardown_on_active_runtime( + #[future] managed_connection_with_teardown: super::helpers::TestResult, +) -> TestResult { + let (mut managed, teardown_count) = managed_connection_with_teardown.await?; + let (sender, receiver) = tokio::sync::oneshot::channel(); + managed.notify_after_close(sender); drop(managed); - timeout(Duration::from_secs(1), async { - while teardown_count.load(Ordering::SeqCst) == 0 { - tokio::task::yield_now().await; - } - }) - .await?; + timeout(Duration::from_secs(1), receiver).await??; assert_teardown_once(&teardown_count); Ok(()) } +#[rstest] #[tokio::test] -async fn managed_connection_drop_runs_teardown_without_runtime() -> TestResult { - let teardown_count = Arc::new(AtomicUsize::new(0)); - let managed = managed_connection_with_teardown(Arc::clone(&teardown_count)).await?; +async fn managed_connection_drop_runs_teardown_without_runtime( + #[future] managed_connection_with_teardown: super::helpers::TestResult, +) -> TestResult { + let (mut managed, teardown_count) = managed_connection_with_teardown.await?; + let (sender, receiver) = tokio::sync::oneshot::channel(); + managed.notify_after_close(sender); std::thread::spawn(move || drop(managed)) .join() .map_err(|_| std::io::Error::other("pooled connection drop thread panicked"))?; + receiver.await?; assert_teardown_once(&teardown_count); Ok(()) } diff --git a/src/client/tests/tracing.rs b/src/client/tests/tracing.rs index eeb5cb524..8e8574688 100644 --- a/src/client/tests/tracing.rs +++ b/src/client/tests/tracing.rs @@ -117,7 +117,7 @@ async fn receive_emits_span_with_result() { test_span_emission!( TracingConfig::default().with_receive_timing(true), "client.receive", - &["result=\"ok\""], + &["result=\"ok\"", "elapsed_us"], |mut client, _addr| async move { let envelope = Envelope::new(1, None, vec![1, 2, 3]); client.send(&envelope).await.expect("send"); @@ -187,7 +187,7 @@ async fn close_emits_span() { .await .expect("run echo client"); - logs_assert(span_assertion("client.close", &[])); + logs_assert(span_assertion("client.close", &["elapsed_us"])); } #[rstest] diff --git a/src/client/tests/tracing_streaming.rs b/src/client/tests/tracing_streaming.rs index a081e0dc5..708b678f8 100644 --- a/src/client/tests/tracing_streaming.rs +++ b/src/client/tests/tracing_streaming.rs @@ -51,7 +51,10 @@ async fn call_streaming_emits_span_with_correlation_id() { // With streaming timing enabled, the elapsed_us event fires within // the client.call_streaming span, making the span name visible. - logs_assert(span_assertion("client.call_streaming", &["correlation_id"])); + logs_assert(span_assertion( + "client.call_streaming", + &["correlation_id", "elapsed_us"], + )); } #[rstest] diff --git a/src/client/tracing_timing.rs b/src/client/tracing_timing.rs index 902ae1cd0..87588a30a 100644 --- a/src/client/tracing_timing.rs +++ b/src/client/tracing_timing.rs @@ -55,3 +55,122 @@ impl ClientTimingFlags { self.0 & operation.mask() != 0 } } + +#[cfg(test)] +mod tests { + //! Checks timing updates against a six-operation reference model. + + use proptest::{collection::vec, prelude::*}; + + use super::{ClientOperation, ClientTimingFlags}; + use crate::client::TracingConfig; + + const OPERATIONS: [ClientOperation; 6] = [ + ClientOperation::Connect, + ClientOperation::Send, + ClientOperation::Receive, + ClientOperation::Call, + ClientOperation::Streaming, + ClientOperation::Close, + ]; + + #[derive(Clone, Copy, Debug)] + enum TimingUpdate { + Set(ClientOperation, bool), + SetAll(bool), + } + + fn timing_update_strategy() -> impl Strategy { + let operation = prop_oneof![ + Just(ClientOperation::Connect), + Just(ClientOperation::Send), + Just(ClientOperation::Receive), + Just(ClientOperation::Call), + Just(ClientOperation::Streaming), + Just(ClientOperation::Close), + ]; + + prop_oneof![ + (operation, any::()) + .prop_map(|(operation, enabled)| TimingUpdate::Set(operation, enabled)), + any::().prop_map(TimingUpdate::SetAll), + ] + } + + fn set_expected_timing(expected: &mut [bool; 6], operation: ClientOperation, enabled: bool) { + let [connect, send, receive, call, streaming, close] = expected; + let selected = match operation { + ClientOperation::Connect => connect, + ClientOperation::Send => send, + ClientOperation::Receive => receive, + ClientOperation::Call => call, + ClientOperation::Streaming => streaming, + ClientOperation::Close => close, + }; + *selected = enabled; + } + + fn set_config_timing( + config: TracingConfig, + operation: ClientOperation, + enabled: bool, + ) -> TracingConfig { + match operation { + ClientOperation::Connect => config.with_connect_timing(enabled), + ClientOperation::Send => config.with_send_timing(enabled), + ClientOperation::Receive => config.with_receive_timing(enabled), + ClientOperation::Call => config.with_call_timing(enabled), + ClientOperation::Streaming => config.with_streaming_timing(enabled), + ClientOperation::Close => config.with_close_timing(enabled), + } + } + + fn flags_snapshot(flags: ClientTimingFlags) -> [bool; 6] { + OPERATIONS.map(|operation| flags.is_enabled(operation)) + } + + fn config_snapshot(config: &TracingConfig) -> [bool; 6] { + OPERATIONS.map(|operation| config.timing_enabled(operation)) + } + + proptest! { + #[test] + fn timing_updates_match_reference_after_every_action( + updates in vec(timing_update_strategy(), 1..64), + ) { + let mut flags = ClientTimingFlags::default(); + let mut config = TracingConfig::default(); + let mut expected = [false; 6]; + + for (step, update) in updates.into_iter().enumerate() { + match update { + TimingUpdate::Set(operation, enabled) => { + flags = flags.with_enabled(operation, enabled); + config = set_config_timing(config, operation, enabled); + set_expected_timing(&mut expected, operation, enabled); + } + TimingUpdate::SetAll(enabled) => { + flags = ClientTimingFlags::all(enabled); + config = config.with_all_timing(enabled); + expected = [enabled; 6]; + } + } + + prop_assert_eq!( + flags_snapshot(flags), + expected, + "ClientTimingFlags diverged after update {}: {:?}", + step, + update, + ); + prop_assert_eq!( + config_snapshot(&config), + expected, + "TracingConfig diverged after update {}: {:?}", + step, + update, + ); + } + } + } +} diff --git a/tests/compile_error.rs b/tests/compile_error.rs index b0cdd264f..c1cf11e5a 100644 --- a/tests/compile_error.rs +++ b/tests/compile_error.rs @@ -7,6 +7,7 @@ fn compile_tests() { t.pass("tests/ui/server_from_app.rs"); t.pass("tests/ui/server_factory_forms.rs"); t.pass("tests/ui/wireframe_result_default_no_protocol.rs"); + t.pass("tests/ui/client_const_api.rs"); t.compile_fail("tests/ui/server_rejects_non_send_factory.rs"); t.compile_fail("tests/ui/wireframe_result_default_rejects_unit_protocol.rs"); t.compile_fail("tests/ui/prepared_app_rejects_route.rs"); diff --git a/tests/ui/client_const_api.rs b/tests/ui/client_const_api.rs new file mode 100644 index 000000000..ca76758c2 --- /dev/null +++ b/tests/ui/client_const_api.rs @@ -0,0 +1,120 @@ +//! Verifies public client configuration methods remain callable in const contexts. + +use std::time::Duration; + +use tracing::Level; +use wireframe::{ + client::{ + ClientCodecConfig, + SendStreamingConfig, + SendStreamingOutcome, + SocketOptions, + TracingConfig, + WireframeClientBuilder, + }, + frame::LengthFormat, +}; + +const STREAMING_OUTCOME: SendStreamingOutcome = SendStreamingOutcome::new(3); +const STREAMING_FRAMES_SENT: u64 = STREAMING_OUTCOME.frames_sent(); + +const fn configure_socket_options( + options: SocketOptions, + keepalive: Duration, +) -> SocketOptions { + let configured = options + .nodelay(true) + .keepalive(Some(keepalive)) + .send_buffer_size(2048) + .recv_buffer_size(4096) + .reuseaddr(true); + + #[cfg(all( + unix, + not(target_os = "solaris"), + not(target_os = "illumos"), + not(target_os = "cygwin"), + ))] + let configured = configured.reuseport(true); + + configured +} + +const fn configure_codec( + config: ClientCodecConfig, + length_format: LengthFormat, +) -> (ClientCodecConfig, usize, LengthFormat) { + let configured = config.length_format(length_format); + ( + configured, + configured.max_frame_length_value(), + configured.length_format_value(), + ) +} + +const fn configure_tracing(config: TracingConfig, level: Level) -> TracingConfig { + config + .with_connect_level(level) + .with_connect_timing(true) + .with_send_level(level) + .with_send_timing(true) + .with_receive_level(level) + .with_receive_timing(true) + .with_call_level(level) + .with_call_timing(true) + .with_streaming_level(level) + .with_streaming_timing(true) + .with_close_level(level) + .with_close_timing(true) + .with_all_levels(level) + .with_all_timing(true) +} + +const fn configure_streaming( + config: SendStreamingConfig, + timeout: Duration, +) -> (SendStreamingConfig, Option, Option) { + let configured = config.with_chunk_size(1024).with_timeout(timeout); + (configured, configured.chunk_size(), configured.timeout()) +} + +const fn configure_builder( + builder: WireframeClientBuilder, + socket_options: SocketOptions, + codec_config: ClientCodecConfig, + tracing_config: TracingConfig, + length_format: LengthFormat, + keepalive: Duration, +) -> WireframeClientBuilder { + let configured = builder + .socket_options(socket_options) + .nodelay(true) + .keepalive(Some(keepalive)) + .send_buffer_size(2048) + .recv_buffer_size(4096) + .reuseaddr(true) + .codec_config(codec_config) + .length_format(length_format) + .tracing_config(tracing_config); + + #[cfg(all( + unix, + not(target_os = "solaris"), + not(target_os = "illumos"), + not(target_os = "cygwin"), + ))] + let configured = configured.reuseport(true); + + configured +} + +fn main() { + let _ = ( + STREAMING_FRAMES_SENT, + configure_socket_options, + configure_codec, + configure_tracing, + configure_streaming, + configure_builder, + ); +} From a8cbfbc0186cfc5184b7cf61fb199333a79c17c8 Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 24 Sep 2026 02:48:56 +0200 Subject: [PATCH 03/11] Split const client builder checks by concern Keep every builder method covered in a const context while reducing the UI test helper argument count. Separate socket configuration from codec and tracing configuration so the fixture stays reviewable. --- tests/ui/client_const_api.rs | 25 ++++++++++++++++--------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/tests/ui/client_const_api.rs b/tests/ui/client_const_api.rs index ca76758c2..924217858 100644 --- a/tests/ui/client_const_api.rs +++ b/tests/ui/client_const_api.rs @@ -78,12 +78,9 @@ const fn configure_streaming( (configured, configured.chunk_size(), configured.timeout()) } -const fn configure_builder( +const fn configure_builder_socket( builder: WireframeClientBuilder, socket_options: SocketOptions, - codec_config: ClientCodecConfig, - tracing_config: TracingConfig, - length_format: LengthFormat, keepalive: Duration, ) -> WireframeClientBuilder { let configured = builder @@ -92,10 +89,7 @@ const fn configure_builder( .keepalive(Some(keepalive)) .send_buffer_size(2048) .recv_buffer_size(4096) - .reuseaddr(true) - .codec_config(codec_config) - .length_format(length_format) - .tracing_config(tracing_config); + .reuseaddr(true); #[cfg(all( unix, @@ -108,6 +102,18 @@ const fn configure_builder( configured } +const fn configure_builder_codec_and_tracing( + builder: WireframeClientBuilder, + codec_config: ClientCodecConfig, + tracing_config: TracingConfig, + length_format: LengthFormat, +) -> WireframeClientBuilder { + builder + .codec_config(codec_config) + .length_format(length_format) + .tracing_config(tracing_config) +} + fn main() { let _ = ( STREAMING_FRAMES_SENT, @@ -115,6 +121,7 @@ fn main() { configure_codec, configure_tracing, configure_streaming, - configure_builder, + configure_builder_socket, + configure_builder_codec_and_tracing, ); } From a09249a04be768b1942ae6f4bf8fe5e505ab2df6 Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 24 Sep 2026 02:54:15 +0200 Subject: [PATCH 04/11] Share const socket configuration checks Use one test macro to exercise the same socket setter sequence on both SocketOptions and WireframeClientBuilder in const contexts. This keeps all builder calls covered while removing duplicated test logic. --- tests/ui/client_const_api.rs | 58 ++++++++++++++---------------------- 1 file changed, 23 insertions(+), 35 deletions(-) diff --git a/tests/ui/client_const_api.rs b/tests/ui/client_const_api.rs index 924217858..060aca086 100644 --- a/tests/ui/client_const_api.rs +++ b/tests/ui/client_const_api.rs @@ -18,26 +18,30 @@ use wireframe::{ const STREAMING_OUTCOME: SendStreamingOutcome = SendStreamingOutcome::new(3); const STREAMING_FRAMES_SENT: u64 = STREAMING_OUTCOME.frames_sent(); -const fn configure_socket_options( - options: SocketOptions, - keepalive: Duration, -) -> SocketOptions { - let configured = options - .nodelay(true) - .keepalive(Some(keepalive)) - .send_buffer_size(2048) - .recv_buffer_size(4096) - .reuseaddr(true); +// Both public receiver types must compile the same const socket setters. +macro_rules! configure_socket_settings { + ($target:expr, $keepalive:expr) => {{ + let configured = $target + .nodelay(true) + .keepalive(Some($keepalive)) + .send_buffer_size(2048) + .recv_buffer_size(4096) + .reuseaddr(true); + + #[cfg(all( + unix, + not(target_os = "solaris"), + not(target_os = "illumos"), + not(target_os = "cygwin"), + ))] + let configured = configured.reuseport(true); - #[cfg(all( - unix, - not(target_os = "solaris"), - not(target_os = "illumos"), - not(target_os = "cygwin"), - ))] - let configured = configured.reuseport(true); + configured + }}; +} - configured +const fn configure_socket_options(options: SocketOptions, keepalive: Duration) -> SocketOptions { + configure_socket_settings!(options, keepalive) } const fn configure_codec( @@ -83,23 +87,7 @@ const fn configure_builder_socket( socket_options: SocketOptions, keepalive: Duration, ) -> WireframeClientBuilder { - let configured = builder - .socket_options(socket_options) - .nodelay(true) - .keepalive(Some(keepalive)) - .send_buffer_size(2048) - .recv_buffer_size(4096) - .reuseaddr(true); - - #[cfg(all( - unix, - not(target_os = "solaris"), - not(target_os = "illumos"), - not(target_os = "cygwin"), - ))] - let configured = configured.reuseport(true); - - configured + configure_socket_settings!(builder.socket_options(socket_options), keepalive) } const fn configure_builder_codec_and_tracing( From 824952572d42b34fbb9d49d8c91fa3da2e2984ed Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 24 Sep 2026 03:23:51 +0200 Subject: [PATCH 05/11] Name client lint probe bindings distinctly Keep the test-only close completion sender and timing strategy distinct from their extracted values. This removes two shadow-reuse findings introduced by the review tests without changing their behavior. --- src/client/pool/managed.rs | 4 ++-- src/client/tracing_timing.rs | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/client/pool/managed.rs b/src/client/pool/managed.rs index 73b063a88..db49600e0 100644 --- a/src/client/pool/managed.rs +++ b/src/client/pool/managed.rs @@ -16,8 +16,8 @@ use crate::{ /// Complete a test wait only after the close future has returned. #[cfg(test)] -fn signal_close_completion(sender: Option>) { - if let Some(sender) = sender { +fn signal_close_completion(completion_sender: Option>) { + if let Some(sender) = completion_sender { sender.send(()).unwrap_or_default(); } } diff --git a/src/client/tracing_timing.rs b/src/client/tracing_timing.rs index 87588a30a..613a7b4b6 100644 --- a/src/client/tracing_timing.rs +++ b/src/client/tracing_timing.rs @@ -81,7 +81,7 @@ mod tests { } fn timing_update_strategy() -> impl Strategy { - let operation = prop_oneof![ + let operation_strategy = prop_oneof![ Just(ClientOperation::Connect), Just(ClientOperation::Send), Just(ClientOperation::Receive), @@ -91,7 +91,7 @@ mod tests { ]; prop_oneof![ - (operation, any::()) + (operation_strategy, any::()) .prop_map(|(operation, enabled)| TimingUpdate::Set(operation, enabled)), any::().prop_map(TimingUpdate::SetAll), ] From 51a2188c66d97f3943ff67f4b400a38a0559656b Mon Sep 17 00:00:00 2001 From: leynos Date: Sun, 11 Oct 2026 03:59:58 +0200 Subject: [PATCH 06/11] Share the lazy hook counter across client tests `record_count` and `record_error_count` were byte-identical lazy incrementing counters defined privately in the lifecycle and error handling test modules. Both modules already import from `super::helpers`, so the duplicate moves there once as `pub(super) fn record_count` and both modules import it. This mirrors the existing `pub(super)` convention used by `request_hooks_support`, which sibling modules already consume. The now-unused `future::Future` and `futures::future::lazy` imports are dropped from `error_handling.rs`; `lifecycle.rs` keeps both because it still defines its own lazy helpers. Behaviour is unchanged: the counter still increments only on first poll, so a hook that is constructed but never driven leaves the count at zero. Co-Authored-By: Claude Code --- src/client/tests/error_handling.rs | 33 +++++++++++------------------- src/client/tests/helpers.rs | 13 +++++++++++- src/client/tests/lifecycle.rs | 14 ++++++------- 3 files changed, 30 insertions(+), 30 deletions(-) diff --git a/src/client/tests/error_handling.rs b/src/client/tests/error_handling.rs index fb7fff79c..0e2d80fd0 100644 --- a/src/client/tests/error_handling.rs +++ b/src/client/tests/error_handling.rs @@ -1,7 +1,6 @@ //! Error handling tests for the wireframe client. use std::{ - future::Future, sync::{ Arc, atomic::{AtomicUsize, Ordering}, @@ -10,13 +9,14 @@ use std::{ }; use bytes::Bytes; -use futures::{SinkExt, future::lazy}; +use futures::SinkExt; use tokio::net::TcpStream; use tokio_util::codec::{Framed, LengthDelimitedCodec}; use super::helpers::{ FailingSerializer, TestResult, + record_count, spawn_listener, test_error_hook_on_disconnect, test_with_client, @@ -28,13 +28,6 @@ use crate::{ serializer::Serializer, }; -/// Increment the error-hook count when its future is first polled. -fn record_error_count(count: Arc) -> impl Future + Send { - lazy(move |_| { - count.fetch_add(1, Ordering::SeqCst); - }) -} - /// Connects a client and returns both the client and server stream for custom server behaviour. /// /// This helper is used by tests that need to manipulate the server stream after connection, @@ -58,7 +51,7 @@ where #[tokio::test] async fn error_callback_invoked_on_receive_error() { let error_count = test_error_hook_on_disconnect(|builder, count| { - builder.on_error(move |_err| record_error_count(count.clone())) + builder.on_error(move |_err| record_count(count.clone())) }) .await .expect("run error hook scenario"); @@ -90,11 +83,10 @@ async fn error_callback_invoked_on_deserialize_error() { let error_count = Arc::new(AtomicUsize::new(0)); let count = error_count.clone(); - let (mut client, server_stream) = connect_with_server(|builder| { - builder.on_error(move |_err| record_error_count(count.clone())) - }) - .await - .expect("connect client and server"); + let (mut client, server_stream) = + connect_with_server(|builder| builder.on_error(move |_err| record_count(count.clone()))) + .await + .expect("connect client and server"); // Send invalid bincode data via the server stream let mut framed = Framed::new(server_stream, LengthDelimitedCodec::new()); @@ -133,11 +125,10 @@ async fn error_callback_invoked_on_send_io_error() { let error_count = Arc::new(AtomicUsize::new(0)); let count = error_count.clone(); - let (mut client, server_stream) = connect_with_server(|builder| { - builder.on_error(move |_err| record_error_count(count.clone())) - }) - .await - .expect("connect client and server"); + let (mut client, server_stream) = + connect_with_server(|builder| builder.on_error(move |_err| record_count(count.clone()))) + .await + .expect("connect client and server"); // Drop the server side to cause a broken pipe on send drop(server_stream); @@ -178,7 +169,7 @@ async fn error_callback_invoked_on_serialize_error() { let (mut client, _server) = connect_with_server(|builder| { builder .serializer(FailingSerializer) - .on_error(move |_err| record_error_count(count.clone())) + .on_error(move |_err| record_count(count.clone())) }) .await .expect("connect client and server"); diff --git a/src/client/tests/helpers.rs b/src/client/tests/helpers.rs index 09f6cba83..bd99cb85f 100644 --- a/src/client/tests/helpers.rs +++ b/src/client/tests/helpers.rs @@ -11,7 +11,7 @@ use std::{ }; use bytes::Bytes; -use futures::{SinkExt, StreamExt}; +use futures::{SinkExt, StreamExt, future::lazy}; use tokio::net::{TcpListener, TcpStream}; use tokio_util::codec::{Framed, LengthDelimitedCodec}; use wireframe_testing::{ServerMode, process_frame}; @@ -210,6 +210,17 @@ where Ok(client) } +/// Increment a hook counter when the returned future is first polled. +/// +/// Deferring the increment to the first poll keeps lifecycle and error-hook +/// tests honest: a hook that is merely constructed, but never driven, leaves +/// the counter at zero. +pub(super) fn record_count(count: Arc) -> impl Future + Send { + lazy(move |_| { + count.fetch_add(1, Ordering::SeqCst); + }) +} + /// Creates a counter and an incrementing hook for use in lifecycle tests. /// /// Returns a tuple of the counter and a closure that increments the counter diff --git a/src/client/tests/lifecycle.rs b/src/client/tests/lifecycle.rs index 48ae6747c..e9c5d6550 100644 --- a/src/client/tests/lifecycle.rs +++ b/src/client/tests/lifecycle.rs @@ -10,14 +10,12 @@ use std::{ use futures::future::lazy; -use super::helpers::{counting_hook, test_error_hook_on_disconnect, test_with_client}; - -/// Increment a lifecycle-hook count when its future is first polled. -fn record_count(count: Arc) -> impl Future + Send { - lazy(move |_| { - count.fetch_add(1, Ordering::SeqCst); - }) -} +use super::helpers::{ + counting_hook, + record_count, + test_error_hook_on_disconnect, + test_with_client, +}; /// Count setup when the callback future is first polled, then return the /// state that the teardown callback receives. From ce0cfae91ebfdba5ed6eec03961ee6cb520b15b0 Mon Sep 17 00:00:00 2001 From: leynos Date: Sun, 11 Oct 2026 04:00:03 +0200 Subject: [PATCH 07/11] Propagate the push error in the unlimited queue test `unlimited_queues_do_not_block` wrapped its low-priority push in `time::timeout` and discarded the inner `Result` with `let _`. Only the outer `Elapsed` case was mapped to a panic message, so a genuine `PushError` was swallowed and surfaced later as an opaque receive failure. Both layers are now propagated, matching the pattern already used by the throttling-disabled test at the top of the file. A push that errors now reports its own cause instead of being misattributed to the subsequent receive. Co-Authored-By: Claude Code --- tests/push.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/push.rs b/tests/push.rs index 3dbc2e38f..93d337351 100644 --- a/tests/push.rs +++ b/tests/push.rs @@ -269,9 +269,10 @@ async fn unlimited_queues_do_not_block() -> TestResult<()> { time::pause(); let (mut queues, handle) = support::builder::().unlimited().build()?; push_expect!(handle.push_high_priority(1u8))?; - let _ = time::timeout(Duration::from_millis(10), handle.push_low_priority(2u8)) + time::timeout(Duration::from_millis(10), handle.push_low_priority(2u8)) .await - .map_err(|error| format!("pushes should not block when unlimited: {error}"))?; + .map_err(|error| format!("pushes should not block when unlimited: {error}"))? + .map_err(|error| format!("push should not error when unlimited: {error}"))?; let (_, a) = recv_expect!(queues.recv())?; let (_, b) = recv_expect!(queues.recv())?; From 70a998be07a032be61fa01e36b2f7cbbdb07e25c Mon Sep 17 00:00:00 2001 From: leynos Date: Sun, 11 Oct 2026 04:16:45 +0200 Subject: [PATCH 08/11] Collapse the pooled teardown tests into one case table The active-runtime and no-runtime teardown tests were near-identical: each built the same fixture, armed the same post-close signal, dropped the connection, awaited the signal, and asserted teardown ran once. The only difference was where the drop happened. CodeScene flagged the pair as code duplication, and the hosted "Enforce advisory code health rules" gate failed the pull request on it. The shared body now lives in a single `rstest` parameterized test with a `DropSite` case value, and the drop strategy moves into `DropSite::drop_managed`. This preserves both behaviours exactly. The active-runtime case drops inline while the test's Tokio runtime is driving; the detached-thread case drops on a spawned thread that owns no runtime, surfacing a panic from that thread as an error. Both cases keep their own assertion that the teardown callback ran exactly once. Co-Authored-By: Claude Code --- src/client/tests/pool.rs | 47 +++++++++++++++++++++++++--------------- 1 file changed, 30 insertions(+), 17 deletions(-) diff --git a/src/client/tests/pool.rs b/src/client/tests/pool.rs index 84db2e628..d837f8392 100644 --- a/src/client/tests/pool.rs +++ b/src/client/tests/pool.rs @@ -61,34 +61,47 @@ async fn managed_connection_with_teardown() -> super::helpers::TestResult, -) -> TestResult { - let (mut managed, teardown_count) = managed_connection_with_teardown.await?; - let (sender, receiver) = tokio::sync::oneshot::channel(); - managed.notify_after_close(sender); +/// Where the pooled connection is dropped, relative to the test's runtime. +#[derive(Debug, Clone, Copy)] +enum DropSite { + /// Drop inline, while the test's Tokio runtime is still driving. + ActiveRuntime, + /// Drop on a spawned thread, which owns no Tokio runtime. + DetachedThread, +} - drop(managed); - timeout(Duration::from_secs(1), receiver).await??; - assert_teardown_once(&teardown_count); - Ok(()) +impl DropSite { + /// Drop `managed` here, surfacing a panic from the spawned thread. + fn drop_managed(self, managed: ManagedClientConnection) -> TestResult { + match self { + Self::ActiveRuntime => { + drop(managed); + Ok(()) + } + Self::DetachedThread => std::thread::spawn(move || drop(managed)).join().map_err( + |_| -> Box { + std::io::Error::other("pooled connection drop thread panicked").into() + }, + ), + } + } } +/// Teardown runs exactly once whether or not a runtime is driving the drop. #[rstest] +#[case::active_runtime(DropSite::ActiveRuntime)] +#[case::detached_thread(DropSite::DetachedThread)] #[tokio::test] -async fn managed_connection_drop_runs_teardown_without_runtime( +async fn managed_connection_drop_runs_teardown_once( + #[case] drop_site: DropSite, #[future] managed_connection_with_teardown: super::helpers::TestResult, ) -> TestResult { let (mut managed, teardown_count) = managed_connection_with_teardown.await?; let (sender, receiver) = tokio::sync::oneshot::channel(); managed.notify_after_close(sender); - std::thread::spawn(move || drop(managed)) - .join() - .map_err(|_| std::io::Error::other("pooled connection drop thread panicked"))?; - receiver.await?; + drop_site.drop_managed(managed)?; + timeout(Duration::from_secs(1), receiver).await??; assert_teardown_once(&teardown_count); Ok(()) } From 2b67a90bb324857d07c20fe375b62c7214f8b103 Mon Sep 17 00:00:00 2001 From: leynos Date: Mon, 12 Oct 2026 02:28:38 +0200 Subject: [PATCH 09/11] Bound the detached-thread pooled teardown test The detached-thread case of `managed_connection_drop_runs_teardown_once` joined its worker thread synchronously before arming the one-second notification timeout. `#[tokio::test]` drives a single-threaded runtime, so that blocking join starved the executor and the timeout could never be reached: a teardown that never completed would hang the suite rather than fail it. `DropSite::drop_managed` now returns a `DropCompletion` describing how to wait, instead of joining inline. The detached worker signals through a oneshot channel as soon as the drop completes, and `DropCompletion::wait` awaits that channel under a shared `GRACE` budget before joining the worker. A stuck drop therefore fails on the elapsed timeout with a diagnostic, while a worker panic is still reported against the test because the join happens once the worker has already finished. The bound is verified by negative control rather than by inspection alone: injecting a thirty-second sleep into the detached worker makes the case fail in 1.00s with "pooled connection drop did not finish within the grace period" instead of blocking indefinitely. Co-Authored-By: Claude Code --- src/client/tests/pool.rs | 82 +++++++++++++++++++++++++++++++++++----- 1 file changed, 72 insertions(+), 10 deletions(-) diff --git a/src/client/tests/pool.rs b/src/client/tests/pool.rs index d837f8392..cff0cdfc4 100644 --- a/src/client/tests/pool.rs +++ b/src/client/tests/pool.rs @@ -31,6 +31,10 @@ type ManagedFixture = ( Arc, ); +/// Upper bound on legitimate pooled teardown, so a stuck drop fails loudly +/// rather than hanging the suite. +const GRACE: Duration = Duration::from_secs(1); + /// Keep teardown failures attributed to the test that dropped the connection. #[track_caller] fn assert_teardown_once(teardown_count: &AtomicUsize) { @@ -70,24 +74,82 @@ enum DropSite { DetachedThread, } +/// How to wait for a drop that has been performed. +enum DropCompletion { + /// The drop ran inline on the test thread, so it has already finished. + Inline, + /// The drop runs on a detached thread that signals when it completes. + /// + /// Joining that thread blocks the caller, and `#[tokio::test]` drives a + /// single-threaded runtime, so joining inline would starve the executor and + /// stop any later `timeout` from arming. The worker signals through a + /// channel instead, letting the caller bound the wait, and is joined only + /// once it has finished. + Detached { + worker: std::thread::JoinHandle<()>, + finished: tokio::sync::oneshot::Receiver<()>, + }, +} + +impl DropCompletion { + /// Wait for the drop to finish, failing rather than hanging past `grace`. + /// + /// Waiting on the channel first is what makes a stuck teardown observable: + /// the caller fails on the elapsed timeout instead of blocking forever in a + /// join. The worker is joined afterwards, once it has already finished, so + /// that a panic inside it is still reported against this test. + async fn wait(self, grace: Duration) -> TestResult { + let Self::Detached { worker, finished } = self else { + return Ok(()); + }; + if timeout(grace, finished).await.is_err() { + return Err(std::io::Error::other( + "pooled connection drop did not finish within the grace period", + ) + .into()); + } + join_worker(worker) + } +} + +/// Join a worker that has already finished, surfacing a panic from that thread. +fn join_worker(worker: std::thread::JoinHandle<()>) -> TestResult { + worker + .join() + .map_err(|_| -> Box { + std::io::Error::other("pooled connection drop thread panicked").into() + }) +} + impl DropSite { - /// Drop `managed` here, surfacing a panic from the spawned thread. - fn drop_managed(self, managed: ManagedClientConnection) -> TestResult { + /// Drop `managed` here, returning how to wait for that drop to finish. + fn drop_managed( + self, + managed: ManagedClientConnection, + ) -> DropCompletion { match self { Self::ActiveRuntime => { drop(managed); - Ok(()) + DropCompletion::Inline + } + Self::DetachedThread => { + let (sender, finished) = tokio::sync::oneshot::channel(); + let worker = std::thread::spawn(move || { + drop(managed); + // A failed send means the caller stopped waiting, which is + // not a worker failure. + let _ = sender.send(()); + }); + DropCompletion::Detached { worker, finished } } - Self::DetachedThread => std::thread::spawn(move || drop(managed)).join().map_err( - |_| -> Box { - std::io::Error::other("pooled connection drop thread panicked").into() - }, - ), } } } /// Teardown runs exactly once whether or not a runtime is driving the drop. +/// +/// Both the drop and the teardown signal are awaited under [`GRACE`], so a +/// teardown that never completes fails this test instead of hanging it. #[rstest] #[case::active_runtime(DropSite::ActiveRuntime)] #[case::detached_thread(DropSite::DetachedThread)] @@ -100,8 +162,8 @@ async fn managed_connection_drop_runs_teardown_once( let (sender, receiver) = tokio::sync::oneshot::channel(); managed.notify_after_close(sender); - drop_site.drop_managed(managed)?; - timeout(Duration::from_secs(1), receiver).await??; + drop_site.drop_managed(managed).wait(GRACE).await?; + timeout(GRACE, receiver).await??; assert_teardown_once(&teardown_count); Ok(()) } From b49857c6f6adb6c03a7185f2afbd98d2dc30cf59 Mon Sep 17 00:00:00 2001 From: leynos Date: Mon, 12 Oct 2026 05:13:40 +0200 Subject: [PATCH 10/11] Document the timing flag helpers with usage examples `ClientTimingFlags::with_enabled`, `all`, and `is_enabled` carried only one-line summaries. AGENTS.md requires function documentation to include clear examples demonstrating usage and outcome, and the repository already follows that for crate-internal helpers such as `ActorState::new` and `process_frame_with_hooks_and_metrics`. Each method now documents its effect with a worked example showing which operations the call selects and which it leaves alone. The fences are `plaintext` because the type is `pub(crate)`: a compiled doctest cannot name it from outside the crate, so an unwrapped `` ``` block would either fail to resolve or force the example to be ignored. `plaintext` keeps the example readable and honest about being illustrative rather than executable. No behaviour changes; documentation only. Co-Authored-By: Claude Code --- src/client/tracing_timing.rs | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/src/client/tracing_timing.rs b/src/client/tracing_timing.rs index 613a7b4b6..c50e545c5 100644 --- a/src/client/tracing_timing.rs +++ b/src/client/tracing_timing.rs @@ -37,6 +37,18 @@ pub(crate) struct ClientTimingFlags(u8); impl ClientTimingFlags { /// Set one operation without changing the other five. + /// + /// # Examples + /// + /// ```plaintext + /// let timing = ClientTimingFlags::default() + /// .with_enabled(ClientOperation::Connect, true) + /// .with_enabled(ClientOperation::Close, true); + /// + /// assert!(timing.is_enabled(ClientOperation::Connect)); + /// assert!(timing.is_enabled(ClientOperation::Close)); + /// assert!(!timing.is_enabled(ClientOperation::Call)); + /// ``` pub(crate) const fn with_enabled(mut self, operation: ClientOperation, enabled: bool) -> Self { let mask = operation.mask(); if enabled { @@ -48,9 +60,29 @@ impl ClientTimingFlags { } /// Set every operation to the same timing policy. + /// + /// # Examples + /// + /// ```plaintext + /// let silent = ClientTimingFlags::all(false); + /// assert!(!silent.is_enabled(ClientOperation::Connect)); + /// + /// let verbose = ClientTimingFlags::all(true); + /// assert!(verbose.is_enabled(ClientOperation::Connect)); + /// assert!(verbose.is_enabled(ClientOperation::Close)); + /// ``` pub(crate) const fn all(enabled: bool) -> Self { Self(if enabled { 0b11_1111 } else { 0 }) } /// Report whether timing is enabled for one operation. + /// + /// # Examples + /// + /// ```plaintext + /// let timing = ClientTimingFlags::default().with_enabled(ClientOperation::Send, true); + /// + /// assert!(timing.is_enabled(ClientOperation::Send)); + /// assert!(!timing.is_enabled(ClientOperation::Receive)); + /// ``` pub(crate) const fn is_enabled(self, operation: ClientOperation) -> bool { self.0 & operation.mask() != 0 } From 4385a077681f386487fd02f016e26b33b187783d Mon Sep 17 00:00:00 2001 From: leynos Date: Mon, 12 Oct 2026 08:03:58 +0200 Subject: [PATCH 11/11] Document the const client surface and timing module CodeRabbit's pre-merge checks failed on b49857c: the new `src/client/tracing_timing.rs` carried only a one-line module comment, and the compile-time client configuration this branch introduces was absent from the users' guide. Expand the module comment into the title-plus-paragraph form used by its siblings, naming the six timed categories and explaining why the six selections live in one `u8` rather than six parallel booleans. Add a "Compile-time client configuration" section to the users' guide. It enumerates the `const` surface, records the three clamping setters that stay non-const because they clamp, warns that `reuseport` is Unix-only and needs `cfg` gating, and points at the trybuild fixture that enforces the contract. The example was compiled as an example target to confirm it builds as written. Co-Authored-By: Claude Code --- docs/users-guide.md | 61 ++++++++++++++++++++++++++++++++++++ src/client/tracing_timing.rs | 6 ++++ 2 files changed, 67 insertions(+) diff --git a/docs/users-guide.md b/docs/users-guide.md index 287141a81..27836350a 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -1637,6 +1637,67 @@ cargo run --example echo --features examples cargo run --example client_echo_login --features examples ``` +### Compile-time client configuration + +Most client configuration methods are `const fn`, so configuration can be +assembled inside a `const fn` and therefore evaluated by the compiler rather +than at runtime. Types with a const constructor go further: +`SendStreamingOutcome::new` is `const`, so an outcome value can initialize a +`const` item directly, and its `frames_sent` getter can be read the same way. + +The const surface covers the builder setters `codec_config`, `socket_options`, +`length_format`, and `tracing_config`; the builder socket shorthands `nodelay`, +`keepalive`, `send_buffer_size`, `recv_buffer_size`, `reuseaddr`, and +`reuseport`; `ClientCodecConfig` members such as `length_format`, +`max_frame_length_value`, and `length_format_value`; the `TracingConfig` +per-operation level and timing setters plus `with_all_levels` and +`with_all_timing`; `SendStreamingConfig::with_chunk_size` and `with_timeout`; +`ClientPoolConfig::fairness_policy`; the read-only getters +`ClientError::should_recycle_connection`, `ResponseStream::correlation_id`, +`is_terminated`, and `frame_count`; and +`SendStreamingOutcome::{new, frames_sent}`. `ClientPoolConfig` requires the +`pool` feature, which is not enabled by default. Each setter takes `self` by +value and returns `Self`, so a configuration chain works unchanged in a +constant context. Setters that clamp their input, such as +`ClientCodecConfig::max_frame_length`, `ClientPoolConfig::pool_size`, and +`ClientPoolConfig::idle_timeout`, remain ordinary functions, as does every +`Default::default` implementation. `reuseport` additionally exists only on Unix +outside Solaris, illumos, and Cygwin, so guard any call to it with the matching +`#[cfg(...)]`. + +```rust +use std::time::Duration; + +use wireframe::client::{SocketOptions, TracingConfig, WireframeClientBuilder}; + +const fn configure( + builder: WireframeClientBuilder, + socket_options: SocketOptions, + tracing: TracingConfig, +) -> WireframeClientBuilder { + builder + .socket_options(socket_options) + .tracing_config(tracing) + .nodelay(true) + .keepalive(Some(Duration::from_secs(30))) + .reuseaddr(true) +} + +let builder = configure( + WireframeClientBuilder::new(), + SocketOptions::default(), + TracingConfig::default().with_connect_timing(true), +); +let _ = builder; +``` + +Opening a connection is not part of this surface: `connect` and `connect_pool` +are `async` methods that perform I/O, so only the builder and its configuration +values are usable in constant contexts, and the chain must be consumed by a +normal `await`ed `connect`. The public contract is enforced by the trybuild +fixture in `tests/ui/client_const_api.rs`, which compiles the same calls inside +`const fn` bodies; run `cargo test --test compile_error` to check it. + ### Client pools Use `connect_pool` when one warm socket is too little, but opening a new TCP diff --git a/src/client/tracing_timing.rs b/src/client/tracing_timing.rs index c50e545c5..c0af2d06f 100644 --- a/src/client/tracing_timing.rs +++ b/src/client/tracing_timing.rs @@ -1,4 +1,10 @@ //! Compact timing switches for the six client operation categories. +//! +//! [`ClientOperation`] names one of six timed categories: `connect`, `send`, +//! `receive`, `call`, `streaming`, and `close`. [`ClientTimingFlags`] stores +//! the six selections as independent bits in one copyable `u8` instead of six +//! parallel booleans, and [`TracingConfig`](super::TracingConfig) drives them +//! through its per-operation setters and reads them back via `timing_enabled`. /// Client operation whose elapsed time can be recorded. #[derive(Clone, Copy, Debug)]