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/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/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/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..db49600e0 100644 --- a/src/client/pool/managed.rs +++ b/src/client/pool/managed.rs @@ -4,14 +4,23 @@ //! `bb8` and guarantees connection teardown hooks still run when the pooled //! connection is dropped or reaped. -use std::{ - mem::ManuallyDrop, - ops::{Deref, DerefMut}, -}; - +#[cfg(test)] +use tokio::sync::oneshot; 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, +}; + +/// Complete a test wait only after the close future has returned. +#[cfg(test)] +fn signal_close_completion(completion_sender: Option>) { + if let Some(sender) = completion_sender { + sender.send(()).unwrap_or_default(); + } +} /// One physical client connection managed by `bb8`. pub(crate) struct ManagedClientConnection @@ -19,10 +28,13 @@ 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, + /// Signals tests only after the detached close future completes. + #[cfg(test)] + close_completion: Option>, } impl ManagedClientConnection @@ -31,36 +43,37 @@ 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, + #[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) 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,13 +82,17 @@ 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() { + #[cfg(test)] + let close_completion = self.close_completion.take(); handle.spawn(async move { client.close().await; + #[cfg(test)] + signal_close_completion(close_completion); }); return; } @@ -85,6 +102,8 @@ where .build() { runtime.block_on(client.close()); + #[cfg(test)] + signal_close_completion(self.close_completion.take()); } } } 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/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. diff --git a/src/client/tests/pool.rs b/src/client/tests/pool.rs index 816bc1690..cff0cdfc4 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, @@ -24,6 +26,147 @@ use crate::{ }; type TestResult = Result<(), Box>; +type ManagedFixture = ( + ManagedClientConnection, + 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) { + 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. +#[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_callback_count); + async move { + callback_count.fetch_add(1, Ordering::SeqCst); + } + }) + }) + .await?; + Ok((ManagedClientConnection::new(client), teardown_count)) +} + +/// 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, +} + +/// 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, returning how to wait for that drop to finish. + fn drop_managed( + self, + managed: ManagedClientConnection, + ) -> DropCompletion { + match self { + Self::ActiveRuntime => { + drop(managed); + 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 } + } + } + } +} + +/// 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)] +#[tokio::test] +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); + + drop_site.drop_managed(managed).wait(GRACE).await?; + timeout(GRACE, receiver).await??; + assert_teardown_once(&teardown_count); + Ok(()) +} #[rustfmt::skip] #[fixture] 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_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..c0af2d06f --- /dev/null +++ b/src/client/tracing_timing.rs @@ -0,0 +1,214 @@ +//! 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)] +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. + /// + /// # 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 { + self.0 |= mask; + } else { + self.0 &= !mask; + } + self + } + + /// 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 + } +} + +#[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_strategy = prop_oneof![ + Just(ClientOperation::Connect), + Just(ClientOperation::Send), + Just(ClientOperation::Receive), + Just(ClientOperation::Call), + Just(ClientOperation::Streaming), + Just(ClientOperation::Close), + ]; + + prop_oneof![ + (operation_strategy, 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/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())?; diff --git a/tests/ui/client_const_api.rs b/tests/ui/client_const_api.rs new file mode 100644 index 000000000..060aca086 --- /dev/null +++ b/tests/ui/client_const_api.rs @@ -0,0 +1,115 @@ +//! 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(); + +// 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); + + configured + }}; +} + +const fn configure_socket_options(options: SocketOptions, keepalive: Duration) -> SocketOptions { + configure_socket_settings!(options, keepalive) +} + +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_socket( + builder: WireframeClientBuilder, + socket_options: SocketOptions, + keepalive: Duration, +) -> WireframeClientBuilder { + configure_socket_settings!(builder.socket_options(socket_options), keepalive) +} + +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, + configure_socket_options, + configure_codec, + configure_tracing, + configure_streaming, + configure_builder_socket, + configure_builder_codec_and_tracing, + ); +}