diff --git a/crates/wireframe-verification/src/connection_model/model.rs b/crates/wireframe-verification/src/connection_model/model.rs index 436da1136..98903f2cb 100644 --- a/crates/wireframe-verification/src/connection_model/model.rs +++ b/crates/wireframe-verification/src/connection_model/model.rs @@ -22,7 +22,7 @@ impl PlaceholderConnectionModel { /// The bound limits path exploration rather than changing which connection /// transitions are legal. #[must_use] - pub fn new(max_steps: u8) -> Self { Self { max_steps } } + pub const fn new(max_steps: u8) -> Self { Self { max_steps } } /// Clone `state` and advance the step counter by one. fn stepped(state: &ConnectionState) -> ConnectionState { diff --git a/crates/wireframe-verification/src/connection_model/properties.rs b/crates/wireframe-verification/src/connection_model/properties.rs index f1f76f97d..217f373b9 100644 --- a/crates/wireframe-verification/src/connection_model/properties.rs +++ b/crates/wireframe-verification/src/connection_model/properties.rs @@ -34,7 +34,7 @@ pub(super) fn properties() -> Vec> { reason = "Stateright property callbacks are function pointers with `&Model` as their required \ first parameter." )] -fn multi_packet_terminator_is_single( +const fn multi_packet_terminator_is_single( _model: &PlaceholderConnectionModel, state: &ConnectionState, ) -> bool { @@ -47,7 +47,10 @@ fn multi_packet_terminator_is_single( reason = "Stateright property callbacks are function pointers with `&Model` as their required \ first parameter." )] -fn high_priority_progress(_model: &PlaceholderConnectionModel, state: &ConnectionState) -> bool { +const fn high_priority_progress( + _model: &PlaceholderConnectionModel, + state: &ConnectionState, +) -> bool { state.emissions.high_priority } @@ -57,7 +60,10 @@ fn high_priority_progress(_model: &PlaceholderConnectionModel, state: &Connectio reason = "Stateright property callbacks are function pointers with `&Model` as their required \ first parameter." )] -fn low_priority_progress(_model: &PlaceholderConnectionModel, state: &ConnectionState) -> bool { +const fn low_priority_progress( + _model: &PlaceholderConnectionModel, + state: &ConnectionState, +) -> bool { state.emissions.low_priority } @@ -67,7 +73,7 @@ fn low_priority_progress(_model: &PlaceholderConnectionModel, state: &Connection reason = "Stateright property callbacks are function pointers with `&Model` as their required \ first parameter." )] -fn response_completion(_model: &PlaceholderConnectionModel, state: &ConnectionState) -> bool { +const fn response_completion(_model: &PlaceholderConnectionModel, state: &ConnectionState) -> bool { state.completed_outputs.response } @@ -77,7 +83,10 @@ fn response_completion(_model: &PlaceholderConnectionModel, state: &ConnectionSt reason = "Stateright property callbacks are function pointers with `&Model` as their required \ first parameter." )] -fn multi_packet_completion(_model: &PlaceholderConnectionModel, state: &ConnectionState) -> bool { +const fn multi_packet_completion( + _model: &PlaceholderConnectionModel, + state: &ConnectionState, +) -> bool { state.completed_outputs.multi_packet } @@ -87,7 +96,7 @@ fn multi_packet_completion(_model: &PlaceholderConnectionModel, state: &Connecti reason = "Stateright property callbacks are function pointers with `&Model` as their required \ first parameter." )] -fn shutdown_races_active_output( +const fn shutdown_races_active_output( _model: &PlaceholderConnectionModel, state: &ConnectionState, ) -> bool { diff --git a/crates/wireframe-verification/src/connection_model/state.rs b/crates/wireframe-verification/src/connection_model/state.rs index 8b222aecd..610abd054 100644 --- a/crates/wireframe-verification/src/connection_model/state.rs +++ b/crates/wireframe-verification/src/connection_model/state.rs @@ -71,16 +71,18 @@ pub(super) struct CompletedOutputs { impl ConnectionState { /// Returns `true` when no output stream is currently active. - pub(crate) fn is_output_idle(&self) -> bool { matches!(self.active_output, ActiveOutput::Idle) } + pub(crate) const fn is_output_idle(&self) -> bool { + matches!(self.active_output, ActiveOutput::Idle) + } /// Returns `true` when a new output stream may be installed (output idle and no shutdown /// pending). - pub(crate) fn can_install_output(&self) -> bool { + pub(crate) const fn can_install_output(&self) -> bool { !self.shutdown.requested && self.is_output_idle() } /// Returns `true` when the fairness policy permits emitting the queued low-priority output. - pub(crate) fn can_emit_low_priority(&self) -> bool { + pub(crate) const fn can_emit_low_priority(&self) -> bool { self.pending_outputs.low_priority_queued && (!self.pending_outputs.high_priority_queued || self.pending_outputs.fairness_allows_low) diff --git a/docs/developers-guide.md b/docs/developers-guide.md index b9e64d5fc..91d7d8956 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -1561,6 +1561,32 @@ developer-only refresh uses `TRYBUILD=overwrite cargo test --test compile_error`. Review refreshed fixtures as maintenance changes, and never run the overwrite command in CI. +#### The const-API surface + +`const` qualifications are a compile-time contract, so they belong in the +trybuild suite rather than a runtime test. The census layers made a large set +of public items `const fn`, and each family has a fixture that calls it from a +`const fn` body: + +| Fixture | Surface it pins | +| ------------------------------------ | ---------------------------------------------------- | +| `tests/ui/client_const_api.rs` | client builder and configuration setters | +| `tests/ui/app_const_api.rs` | app configuration and packet constructors | +| `tests/ui/codec_shared_const_api.rs` | codec, recovery, and middleware constructors | +| `tests/ui/fragment_const_api.rs` | fragmentation configuration | +| `tests/ui/server_const_api.rs` | push-queue builder, `ConnectionId`, codec, and error | + +When a new `const fn` joins the public surface, extend the fixture that owns +that family instead of adding a parallel one. Fixtures must construct their +values in a way the compiler accepts: a `const fn` cannot call a non-const +constructor such as `Default::default`, cannot access a private field, and +cannot evaluate a destructor that is not `const`. Pass such values in as +parameters; the callee's `const`-ness is the property under test, not the +argument. `PushQueuesBuilder`'s frame parameter is the exception worth +remembering: `FrameLike` is crate-visible, but its blanket implementation makes +the callable bound `Send + 'static`, so a fixture writes a generic parameter +and never names the trait. + Trybuild cannot show that a feature-gated item is *absent*. It forwards the harness's own feature list to the generated fixture, and this workspace's `wireframe_testing` dev-dependency enables `wireframe/test-support` through diff --git a/docs/users-guide.md b/docs/users-guide.md index d41b75bb4..a51d701fb 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -2408,6 +2408,54 @@ low-priority queue; `close` lets tests release resources when no actor is running.[^27] The connection actor consumes these queues, an optional streaming response, and a cancellation token. +### Compile-time queue and session configuration + +The queue builder's setters are `const fn`, so an existing builder can be +configured by the compiler. `high_capacity`, `low_capacity`, `rate`, +`unlimited`, `dlq_log_every_n`, and `dlq_log_interval` all take `self` by value +and return `Self`, which means a chain works unchanged inside a `const fn`. The +builder itself is created at runtime, because `PushQueuesBuilder::default` and +`PushQueues::builder` are ordinary functions: + +```rust +use std::time::Duration; + +use wireframe::push::PushQueuesBuilder; + +const fn configure( + builder: PushQueuesBuilder, +) -> PushQueuesBuilder { + builder + .high_capacity(8) + .low_capacity(4) + .rate(Some(64)) + .dlq_log_interval(Duration::from_secs(5)) +} + +let builder = configure(PushQueuesBuilder::>::default()); +let _ = builder; +``` + +`unlimited` is the const shorthand for `rate(None)`, so a chain can also remove +a rate limit that a previous layer installed. The frame parameter is declared +with only `Send + 'static`, so downstream code stays generic over its own frame +type and never names `FrameLike`. That trait is crate-visible, and its blanket +implementation satisfies the bound for every `Send + 'static` type. + +The same const treatment covers `ConnectionId::{new, as_u64}`, +`LengthDelimitedFrameCodec::max_frame_length`, `ConnectionInfo::peer_addr`, +`WireframeError::{from_io, from_codec}`, and `WireframeError::is_clean_close`. A +`WireframeError` has a destructor that cannot run in a `const fn`, so a const +function may borrow an existing error or return a newly built one, but it can +never drop one. That is why the `from_io` payload arrives as a parameter and +the result is returned by value. + +`tests/ui/server_const_api.rs` compiles these calls inside `const` items and +`const fn` bodies, so a method that loses `const` fails the build rather than +silently regressing. The fixture covers the builder setters, both +`ConnectionId` accessors, `max_frame_length`, `peer_addr`, and the two error +constructors; anything not listed there has no compile-time guard yet. + `FairnessConfig` and `FairnessTracker` limit consecutive high-priority frames and optionally enforce a time slice. Resetting the tracker after low-priority work keeps fairness predictable.[^28][^29] `ConnectionActor::run` polls the diff --git a/src/client/pool/churn.rs b/src/client/pool/churn.rs index dd742f555..e82d3a87d 100644 --- a/src/client/pool/churn.rs +++ b/src/client/pool/churn.rs @@ -222,7 +222,7 @@ pub struct PoolTaskChurnObserver { impl PoolTaskChurnObserver { /// Wrap a pool's counters. - pub(crate) fn new(counts: Arc) -> Self { Self { counts } } + pub(crate) const fn new(counts: Arc) -> Self { Self { counts } } /// Read the counters' current values. #[must_use] diff --git a/src/client/pool/client_pool.rs b/src/client/pool/client_pool.rs index e09bbf2bd..e8b09d0aa 100644 --- a/src/client/pool/client_pool.rs +++ b/src/client/pool/client_pool.rs @@ -293,7 +293,7 @@ where /// ``` #[cfg(any(test, feature = "test-support"))] #[must_use] - pub fn slot_id(&self) -> usize { self.selected_slot_index() } + pub const fn slot_id(&self) -> usize { self.selected_slot_index() } } #[cfg(test)] diff --git a/src/client/pool/core.rs b/src/client/pool/core.rs index 6ca5d3918..8588b97ae 100644 --- a/src/client/pool/core.rs +++ b/src/client/pool/core.rs @@ -241,7 +241,7 @@ where /// /// The empty case reports zero so the rotation can be described without a /// branch at every call site; no cursor value is consumed for an empty pool. -fn rotation_start(len: usize, cursor: usize) -> usize { +const fn rotation_start(len: usize, cursor: usize) -> usize { if len == 0 { 0 } else { @@ -261,6 +261,7 @@ fn rotation_indices(len: usize, start: usize) -> impl Iterator + C #[cfg(test)] #[path = "rotation_tests.rs"] mod rotation_tests; + #[cfg(test)] #[path = "core_tests/mod.rs"] mod tests; diff --git a/src/client/pool/lease.rs b/src/client/pool/lease.rs index 3b5a7d0bd..97e17c9a0 100644 --- a/src/client/pool/lease.rs +++ b/src/client/pool/lease.rs @@ -209,11 +209,13 @@ where self.pending_responses.fetch_add(1, Ordering::AcqRel); } PendingResponseChange::Remove => { - let _ = self.pending_responses.try_update( - Ordering::AcqRel, - Ordering::Acquire, - |pending| Some(pending.saturating_sub(1)), - ); + // The update always succeeds because the closure always + // returns `Some`, so an unused `Result` is discarded here. + self.pending_responses + .try_update(Ordering::AcqRel, Ordering::Acquire, |pending| { + Some(pending.saturating_sub(1)) + }) + .ok(); } PendingResponseChange::None => {} } diff --git a/src/client/pool/scheduler/mod.rs b/src/client/pool/scheduler/mod.rs index 36a95e2d7..fdf8b34d2 100644 --- a/src/client/pool/scheduler/mod.rs +++ b/src/client/pool/scheduler/mod.rs @@ -133,13 +133,13 @@ where self.fifo_queue.push_back((handle_id, sender)); WaiterEnqueueResult::Enqueued } - PoolFairnessPolicy::RoundRobin => match self.handle_waiters.get_mut(&handle_id) { - Some(queue) => { + PoolFairnessPolicy::RoundRobin => self.handle_waiters.get_mut(&handle_id).map_or( + WaiterEnqueueResult::UnregisteredHandle, + |queue| { queue.push_back(sender); WaiterEnqueueResult::Enqueued - } - None => WaiterEnqueueResult::UnregisteredHandle, - }, + }, + ), } } @@ -161,7 +161,7 @@ where } /// Release one request after its lease is delivered or its receiver closes. - fn finish_servicing_waiter(&mut self) { + const fn finish_servicing_waiter(&mut self) { self.servicing_waiters = self.servicing_waiters.saturating_sub(1); } diff --git a/src/client/pool/slot.rs b/src/client/pool/slot.rs index 47f6a3d26..7c4d965dd 100644 --- a/src/client/pool/slot.rs +++ b/src/client/pool/slot.rs @@ -206,7 +206,7 @@ where C: Send + 'static, { /// Keep the connection after its operation returned. - pub(crate) fn disarm(&mut self) { self.armed = false; } + pub(crate) const fn disarm(&mut self) { self.armed = false; } /// Invalidate the connection while an operation is still holding it. pub(crate) fn mark_broken(&mut self) { self.connection.mark_broken(); } diff --git a/src/codec/mod.rs b/src/codec/mod.rs index 7088e171a..d6580705a 100644 --- a/src/codec/mod.rs +++ b/src/codec/mod.rs @@ -124,7 +124,7 @@ impl LengthDelimitedFrameCodec { /// Return the maximum frame length accepted by this codec. #[must_use] - pub fn max_frame_length(&self) -> usize { self.max_frame_length } + pub const fn max_frame_length(&self) -> usize { self.max_frame_length } /// Build a fresh tokio codec so encoder and decoder state remain isolated. fn new_inner_codec(&self) -> LengthDelimitedCodec { @@ -225,24 +225,24 @@ impl EofContext { /// - [`EofError::MidFrame`]: Header complete but payload truncated. The length prefix was /// successfully read but the connection closed before the full payload arrived. fn build_eof_error(context: EofContext) -> io::Error { - match context.expected { - Some(expected) => { - // EOF during payload read - we have a header but incomplete payload - CodecError::Eof(EofError::MidFrame { - bytes_received: context.bytes_received.saturating_sub(LENGTH_HEADER_SIZE), - expected, - }) - .into() - } - None => { + context.expected.map_or_else( + || { // EOF during header read CodecError::Eof(EofError::MidHeader { bytes_received: context.bytes_received, header_size: LENGTH_HEADER_SIZE, }) .into() - } - } + }, + |expected| { + // EOF during payload read - we have a header but incomplete payload + CodecError::Eof(EofError::MidFrame { + bytes_received: context.bytes_received.saturating_sub(LENGTH_HEADER_SIZE), + expected, + }) + .into() + }, + ) } /// Encoder half of [`LengthDelimitedFrameCodec`]. diff --git a/src/connection/drain.rs b/src/connection/drain.rs index 7fe8b567b..2b03cd3ae 100644 --- a/src/connection/drain.rs +++ b/src/connection/drain.rs @@ -116,7 +116,7 @@ where } /// Reset burst history after low-priority or [`QueueKind::Multi`] frames. - pub(super) fn after_low(&mut self) { self.fairness.reset(); } + pub(super) const fn after_low(&mut self) { self.fairness.reset(); } /// Common logic for handling closed receivers. pub(super) fn handle_closed_receiver( diff --git a/src/connection/mod.rs b/src/connection/mod.rs index 9932cbf20..3955474ad 100644 --- a/src/connection/mod.rs +++ b/src/connection/mod.rs @@ -183,7 +183,9 @@ where } /// Replace the fairness configuration. - pub fn set_fairness(&mut self, fairness: FairnessConfig) { self.fairness.set_config(fairness); } + pub const fn set_fairness(&mut self, fairness: FairnessConfig) { + self.fairness.set_config(fairness); + } /// Enable transparent fragmentation for outbound frames. /// @@ -191,7 +193,7 @@ where /// into multiple fragments carrying a standard fragment header inside the /// payload. Callers continue to enqueue complete frames; fragmentation /// occurs just before hooks and metrics are applied. - pub fn enable_fragmentation(&mut self, config: FragmentationConfig) + pub const fn enable_fragmentation(&mut self, config: FragmentationConfig) where F: Packet, { diff --git a/src/error.rs b/src/error.rs index 4d67a1f8e..396ee8336 100644 --- a/src/error.rs +++ b/src/error.rs @@ -44,15 +44,15 @@ impl From for WireframeError { impl WireframeError { /// Convert an I/O error into a `WireframeError`. #[must_use] - pub fn from_io(error: std::io::Error) -> Self { Self::Io(error) } + pub const fn from_io(error: std::io::Error) -> Self { Self::Io(error) } /// Convert a codec error into a `WireframeError`. #[must_use] - pub fn from_codec(error: crate::codec::CodecError) -> Self { Self::Codec(error) } + pub const fn from_codec(error: crate::codec::CodecError) -> Self { Self::Codec(error) } /// Returns true if this error represents a clean connection close. #[must_use] - pub fn is_clean_close(&self) -> bool { + pub const fn is_clean_close(&self) -> bool { matches!(self, Self::Codec(codec_error) if codec_error.is_clean_close()) } } @@ -73,8 +73,8 @@ fn transport_or_codec_source( error: &WireframeError, ) -> Option<&(dyn std::error::Error + 'static)> { match error { - WireframeError::Io(error) => Some(error), - WireframeError::Codec(error) => Some(error), + WireframeError::Io(io_error) => Some(io_error), + WireframeError::Codec(codec_error) => Some(codec_error), WireframeError::DuplicateRoute(_) | WireframeError::Protocol(_) => None, } } @@ -101,7 +101,7 @@ where E: std::error::Error + 'static, { match error { - WireframeError::Protocol(error) => Some(error), + WireframeError::Protocol(protocol_error) => Some(protocol_error), WireframeError::DuplicateRoute(_) | WireframeError::Io(_) | WireframeError::Codec(_) => { None } diff --git a/src/extractor/connection_info.rs b/src/extractor/connection_info.rs index 988d6c570..39f8d1954 100644 --- a/src/extractor/connection_info.rs +++ b/src/extractor/connection_info.rs @@ -29,7 +29,7 @@ impl ConnectionInfo { /// assert_eq!(info.peer_addr(), Some(addr)); /// ``` #[must_use] - pub fn peer_addr(&self) -> Option { self.peer_addr } + pub const fn peer_addr(&self) -> Option { self.peer_addr } } impl FromMessageRequest for ConnectionInfo { diff --git a/src/extractor/request.rs b/src/extractor/request.rs index 5752fbfd3..02370e1a2 100644 --- a/src/extractor/request.rs +++ b/src/extractor/request.rs @@ -49,7 +49,7 @@ impl MessageRequest { /// assert!(req.peer_addr.is_some()); /// ``` #[must_use] - pub fn with_peer_addr(mut self, addr: Option) -> Self { + pub const fn with_peer_addr(mut self, addr: Option) -> Self { self.peer_addr = addr; self } @@ -179,7 +179,7 @@ impl<'a> Payload<'a> { /// ``` #[must_use] #[inline] - pub fn new(data: &'a [u8]) -> Self { Self { data } } + pub const fn new(data: &'a [u8]) -> Self { Self { data } } } impl<'a> From<&'a [u8]> for Payload<'a> { @@ -224,5 +224,5 @@ impl Payload<'_> { /// assert_eq!(payload.remaining(), 3); /// ``` #[must_use] - pub fn remaining(&self) -> usize { self.data.len() } + pub const fn remaining(&self) -> usize { self.data.len() } } diff --git a/src/extractor/shared_state.rs b/src/extractor/shared_state.rs index 1589c814e..2296e8494 100644 --- a/src/extractor/shared_state.rs +++ b/src/extractor/shared_state.rs @@ -33,7 +33,7 @@ where _payload: &mut Payload<'_>, ) -> Result { req.state::() - .ok_or(ExtractError::MissingState(std::any::type_name::())) + .ok_or_else(|| ExtractError::MissingState(std::any::type_name::())) } } diff --git a/src/fairness.rs b/src/fairness.rs index a893fff81..191483163 100644 --- a/src/fairness.rs +++ b/src/fairness.rs @@ -60,12 +60,12 @@ pub(crate) struct FairnessTracker { impl FairnessTracker { /// Create a tracker using Tokio's monotonic clock. - pub(crate) fn new(config: FairnessConfig) -> Self { Self::with_clock(config, TokioClock) } + pub(crate) const fn new(config: FairnessConfig) -> Self { Self::with_clock(config, TokioClock) } } impl FairnessTracker { /// Construct a tracker with an explicit clock for deterministic tests. - pub(crate) fn with_clock(config: FairnessConfig, clock: C) -> Self { + pub(crate) const fn with_clock(config: FairnessConfig, clock: C) -> Self { Self { config, clock, @@ -75,7 +75,7 @@ impl FairnessTracker { } /// Replace policy and clear prior burst history so old traffic cannot bias it. - pub(crate) fn set_config(&mut self, config: FairnessConfig) { + pub(crate) const fn set_config(&mut self, config: FairnessConfig) { self.config = config; self.reset(); } @@ -104,10 +104,10 @@ impl FairnessTracker { } /// End the current high-priority burst after low-priority or multi-queue work. - pub(crate) fn reset(&mut self) { self.clear(); } + pub(crate) const fn reset(&mut self) { self.clear(); } /// Clear counters and timer without changing configured fairness policy. - fn clear(&mut self) { + const fn clear(&mut self) { self.high_counter = 0; self.high_start = None; } diff --git a/src/message_assembler/series/mod.rs b/src/message_assembler/series/mod.rs index 77172d050..a3b1f5576 100644 --- a/src/message_assembler/series/mod.rs +++ b/src/message_assembler/series/mod.rs @@ -94,7 +94,7 @@ impl MessageSeries { /// assert!(!series.is_complete()); /// ``` #[must_use] - pub fn from_first_frame(header: &FirstFrameHeader) -> Self { + pub const fn from_first_frame(header: &FirstFrameHeader) -> Self { Self { message_key: header.message_key, next_sequence: None, @@ -273,7 +273,7 @@ impl MessageSeries { impl MessageSeries { /// Testing helper that forces the next expected sequence. #[doc(hidden)] - pub fn force_next_sequence_for_tests(&mut self, next: FrameSequence) { + pub const fn force_next_sequence_for_tests(&mut self, next: FrameSequence) { self.sequence_tracking = SequenceTracking::Tracked; self.next_sequence = Some(next); } diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index 074bf3b5a..95f0f7b88 100644 --- a/src/metrics/mod.rs +++ b/src/metrics/mod.rs @@ -29,7 +29,9 @@ pub(crate) use client_pool::{ inc_pool_acquisition, }; mod decode_rejections; + mod server_startup; + mod server_supervisor; pub use client_pool::{ @@ -64,10 +66,13 @@ pub(crate) use server_supervisor::inc_server_supervisor_abnormal_termination; /// Name of the gauge tracking active connections. pub const CONNECTIONS_ACTIVE: &str = "wireframe_connections_active"; + /// Name of the counter tracking processed frames. pub const FRAMES_PROCESSED: &str = "wireframe_frames_processed_total"; + /// Name of the counter tracking error occurrences. pub const ERRORS_TOTAL: &str = "wireframe_errors_total"; + /// Name of the counter tracking connection panics. /// /// ```plaintext @@ -76,6 +81,7 @@ pub const ERRORS_TOTAL: &str = "wireframe_errors_total"; /// wireframe_connection_panics_total 1 /// ``` pub const CONNECTION_PANICS: &str = "wireframe_connection_panics_total"; + /// Name of the counter tracking recovered client pool bookkeeping lock poison. /// /// ```plaintext @@ -187,10 +193,10 @@ pub enum Direction { impl Direction { /// Supplies the bounded metric label value, avoiding call-site strings /// that could make inbound and outbound observations incomparable. - fn as_str(self) -> &'static str { + const fn as_str(self) -> &'static str { match self { - Direction::Inbound => "inbound", - Direction::Outbound => "outbound", + Self::Inbound => "inbound", + Self::Outbound => "outbound", } } } diff --git a/src/middleware.rs b/src/middleware.rs index 0363b6317..737371885 100644 --- a/src/middleware.rs +++ b/src/middleware.rs @@ -47,7 +47,7 @@ pub struct ServiceRequest { impl ServiceRequest { /// Create a new [`ServiceRequest`] from raw frame bytes. #[must_use] - pub fn new(frame: Vec, correlation_id: Option) -> Self { + pub const fn new(frame: Vec, correlation_id: Option) -> Self { Self { inner: FrameContainer::new(frame), correlation_id, @@ -56,7 +56,7 @@ impl ServiceRequest { /// Borrow the underlying frame bytes. #[must_use] - pub fn frame(&self) -> &[u8] { self.inner.frame().as_slice() } + pub const fn frame(&self) -> &[u8] { self.inner.frame().as_slice() } /// Return the correlation identifier associated with this request, if any. #[must_use] @@ -71,7 +71,7 @@ impl ServiceRequest { /// Mutable access to the inner frame bytes. #[must_use] - pub fn frame_mut(&mut self) -> &mut Vec { self.inner.frame_mut() } + pub const fn frame_mut(&mut self) -> &mut Vec { self.inner.frame_mut() } /// Consume the request, returning the inner frame bytes. #[must_use] @@ -90,7 +90,7 @@ pub struct ServiceResponse { impl ServiceResponse { /// Create a new [`ServiceResponse`] containing the given frame bytes. #[must_use] - pub fn new(frame: Vec, correlation_id: Option) -> Self { + pub const fn new(frame: Vec, correlation_id: Option) -> Self { Self { inner: FrameContainer::new(frame), correlation_id, @@ -99,11 +99,11 @@ impl ServiceResponse { /// Borrow the inner frame bytes. #[must_use] - pub fn frame(&self) -> &[u8] { self.inner.frame().as_slice() } + pub const fn frame(&self) -> &[u8] { self.inner.frame().as_slice() } /// Mutable access to the response frame bytes. #[must_use] - pub fn frame_mut(&mut self) -> &mut Vec { self.inner.frame_mut() } + pub const fn frame_mut(&mut self) -> &mut Vec { self.inner.frame_mut() } /// Return the correlation identifier associated with this response, if any. #[must_use] @@ -235,7 +235,7 @@ impl FromFn { /// # } /// let mw = from_fn(logging); /// ``` -pub fn from_fn(f: F) -> FromFn { FromFn::new(f) } +pub const fn from_fn(f: F) -> FromFn { FromFn::new(f) } /// Service wrapper that applies a middleware function to requests. /// diff --git a/src/panic.rs b/src/panic.rs index ac6b4bc60..394761659 100644 --- a/src/panic.rs +++ b/src/panic.rs @@ -19,11 +19,12 @@ use std::any::Any; /// ``` #[must_use] pub fn format_panic(panic: &Box) -> String { - if let Some(message) = panic.downcast_ref::<&'static str>() { - (*message).to_owned() - } else if let Some(message) = panic.downcast_ref::() { - message.clone() - } else { - format!("{panic:?}") - } + panic.downcast_ref::<&'static str>().map_or_else( + || { + panic + .downcast_ref::() + .map_or_else(|| format!("{panic:?}"), Clone::clone) + }, + |message| (*message).to_owned(), + ) } diff --git a/src/push/queues/builder.rs b/src/push/queues/builder.rs index 5be20bcf7..6dc1cf5f8 100644 --- a/src/push/queues/builder.rs +++ b/src/push/queues/builder.rs @@ -99,14 +99,14 @@ impl Default for PushQueuesBuilder { impl PushQueuesBuilder { /// Set the capacity of the high-priority queue. #[must_use] - pub fn high_capacity(mut self, capacity: usize) -> Self { + pub const fn high_capacity(mut self, capacity: usize) -> Self { self.high_capacity = capacity; self } /// Set the capacity of the low-priority queue. #[must_use] - pub fn low_capacity(mut self, capacity: usize) -> Self { + pub const fn low_capacity(mut self, capacity: usize) -> Self { self.low_capacity = capacity; self } @@ -115,14 +115,14 @@ impl PushQueuesBuilder { /// /// Passing `None` disables rate limiting. #[must_use] - pub fn rate(mut self, rate: Option) -> Self { + pub const fn rate(mut self, rate: Option) -> Self { self.rate = rate; self } /// Disable rate limiting entirely. #[must_use] - pub fn unlimited(self) -> Self { self.rate(None) } + pub const fn unlimited(self) -> Self { self.rate(None) } /// Provide a dead-letter queue for discarded frames. /// @@ -137,7 +137,7 @@ impl PushQueuesBuilder { /// A value of `0` disables count-based logging; interval-based logging /// still applies. #[must_use] - pub fn dlq_log_every_n(mut self, n: usize) -> Self { + pub const fn dlq_log_every_n(mut self, n: usize) -> Self { // Treat 0 as “disabled by count”; interval-based logging remains. self.dlq_log_every_n = n; self @@ -145,7 +145,7 @@ impl PushQueuesBuilder { /// Minimum interval between DLQ drop log entries. #[must_use] - pub fn dlq_log_interval(mut self, interval: Duration) -> Self { + pub const fn dlq_log_interval(mut self, interval: Duration) -> Self { self.dlq_log_interval = interval; self } diff --git a/src/push/queues/dlq.rs b/src/push/queues/dlq.rs index c113ad166..6fb3e236f 100644 --- a/src/push/queues/dlq.rs +++ b/src/push/queues/dlq.rs @@ -66,7 +66,7 @@ impl DeadLetterState { } /// Whether a dead-letter queue is configured to receive dropped frames. - pub(crate) fn has_queue(&self) -> bool { self.sink.is_some() } + pub(crate) const fn has_queue(&self) -> bool { self.sink.is_some() } /// Hand `frame` to the dead-letter queue, recording a loss if that fails. /// @@ -87,9 +87,8 @@ impl DeadLetterState { return; }; let lost_frame = match returned { - mpsc::error::TrySendError::Full(frame) | mpsc::error::TrySendError::Closed(frame) => { - frame - } + mpsc::error::TrySendError::Full(rejected) + | mpsc::error::TrySendError::Closed(rejected) => rejected, }; let dropped = self.drops.fetch_add(1, Ordering::Relaxed) + 1; self.log_if_due(lost_frame, dropped); diff --git a/src/push/queues/handle.rs b/src/push/queues/handle.rs index 7aef9763d..9007da944 100644 --- a/src/push/queues/handle.rs +++ b/src/push/queues/handle.rs @@ -67,7 +67,7 @@ impl PushHandleProbe { impl PushHandle { /// Wrap shared queue state while retaining cloneable producer ownership. - pub(crate) fn from_arc(arc: Arc>) -> Self { Self(arc) } + pub(crate) const fn from_arc(arc: Arc>) -> Self { Self(arc) } /// Returns a probe for inspecting internal state during loom verification. /// diff --git a/src/push/queues/mod.rs b/src/push/queues/mod.rs index 2a5e4b055..e34e1ce4f 100644 --- a/src/push/queues/mod.rs +++ b/src/push/queues/mod.rs @@ -98,7 +98,7 @@ impl PushQueueConfig { /// Create a new configuration with default dead-letter queue logging /// cadence. #[must_use] - pub fn new( + pub const fn new( high_capacity: usize, low_capacity: usize, rate: Option, @@ -127,7 +127,7 @@ impl PushQueues { pub fn builder() -> PushQueuesBuilder { PushQueuesBuilder::default() } /// Returns the invalid rate if it is zero or exceeds the maximum. - fn invalid_rate(rate: Option) -> Option { + const fn invalid_rate(rate: Option) -> Option { match rate { Some(r) if r == 0 || r > MAX_PUSH_RATE => Some(r), _ => None, diff --git a/src/rewind_stream.rs b/src/rewind_stream.rs index 02b239629..e57a4c77e 100644 --- a/src/rewind_stream.rs +++ b/src/rewind_stream.rs @@ -25,7 +25,7 @@ pub struct RewindStream { impl RewindStream { /// Create a new `RewindStream` that will yield `leftover` before /// delegating to `inner`. - pub fn new(leftover: Vec, inner: S) -> Self { + pub const fn new(leftover: Vec, inner: S) -> Self { Self { leftover, pos: 0, @@ -34,10 +34,10 @@ impl RewindStream { } /// Return a reference to the underlying stream. - pub fn inner(&self) -> &S { &self.inner } + pub const fn inner(&self) -> &S { &self.inner } /// Return a mutable reference to the underlying stream. - pub fn inner_mut(&mut self) -> &mut S { &mut self.inner } + pub const fn inner_mut(&mut self) -> &mut S { &mut self.inner } /// Consume the wrapper and return the underlying stream. /// diff --git a/src/server/error.rs b/src/server/error.rs index 8efdf2e46..27f582c0b 100644 --- a/src/server/error.rs +++ b/src/server/error.rs @@ -33,6 +33,6 @@ pub enum ServerError { message: String, /// Original supervisor error, when termination returned an error. #[source] - source: Option>, + source: Option>, }, } diff --git a/src/server/runtime/server/mod.rs b/src/server/runtime/server/mod.rs index dec38d693..495e4bf72 100644 --- a/src/server/runtime/server/mod.rs +++ b/src/server/runtime/server/mod.rs @@ -271,7 +271,7 @@ where preamble_timeout, .. } = self; - let Some((app, shutdown)) = + let Some((app, prepared_shutdown)) = prepare_application_or_shutdown(factory, shutdown, startup_started).await? else { return Ok(()); @@ -293,7 +293,7 @@ where supervisor.lifecycle.clone(), ); runtime - .run_with_shutdown(shutdown, &supervisor, startup_started) + .run_with_shutdown(prepared_shutdown, &supervisor, startup_started) .await } diff --git a/src/server/runtime/server/observer.rs b/src/server/runtime/server/observer.rs index eba8bdd92..312fbae82 100644 --- a/src/server/runtime/server/observer.rs +++ b/src/server/runtime/server/observer.rs @@ -74,7 +74,10 @@ fn classify_join_error(error: JoinError) -> SupervisorJoinOutcome { if error.is_cancelled() { SupervisorJoinOutcome::Cancelled } else { - debug_assert!(error.is_panic()); + debug_assert!( + error.is_panic(), + "a join error that is neither cancelled nor a panic cannot be classified" + ); SupervisorJoinOutcome::Panicked(format_panic(&error.into_panic())) } } diff --git a/src/server/runtime/startup.rs b/src/server/runtime/startup.rs index 8d270631c..d16a7c4c2 100644 --- a/src/server/runtime/startup.rs +++ b/src/server/runtime/startup.rs @@ -38,7 +38,8 @@ where { let app = build_application(&factory, startup_started)?; let prepared = prepare_or_shutdown(app.prepare(), shutdown, startup_started).await?; - Ok(prepared.map(|(app, shutdown)| (Arc::new(app), shutdown))) + Ok(prepared + .map(|(prepared_app, prepared_shutdown)| (Arc::new(prepared_app), prepared_shutdown))) } /// Await preparation unless shutdown completes first, recording its outcome. @@ -51,16 +52,16 @@ where P: Future>, S: Future + Send, { - let mut shutdown = Box::pin(shutdown); + let mut prepared_shutdown = Box::pin(shutdown); let startup_result = select! { - () = &mut shutdown => { + () = &mut prepared_shutdown => { record_server_startup_duration( ServerStartupOutcome::Cancelled, startup_started.elapsed(), ); Ok(None) }, - result = preparation => Ok(Some((record_preparation_result(result, startup_started)?, shutdown))), + result = preparation => Ok(Some((record_preparation_result(result, startup_started)?, prepared_shutdown))), }; startup_result } diff --git a/src/session.rs b/src/session.rs index d527bcc35..2ee609c82 100644 --- a/src/session.rs +++ b/src/session.rs @@ -30,7 +30,7 @@ impl ConnectionId { /// assert_eq!(id.as_u64(), 42); /// ``` #[must_use] - pub fn new(id: u64) -> Self { Self(id) } + pub const fn new(id: u64) -> Self { Self(id) } /// Return the inner `u64` representation. /// @@ -43,7 +43,7 @@ impl ConnectionId { /// assert_eq!(id.as_u64(), 7); /// ``` #[must_use] - pub fn as_u64(&self) -> u64 { self.0 } + pub const fn as_u64(&self) -> u64 { self.0 } } impl std::fmt::Display for ConnectionId { @@ -63,12 +63,11 @@ impl SessionRegistry { mut map: impl FnMut(ConnectionId, Arc>) -> T, ) -> Vec { let mut out = Vec::with_capacity(self.0.len()); - self.0.retain(|id, weak| match weak.upgrade() { - Some(inner) => { + self.0.retain(|id, weak| { + weak.upgrade().is_some_and(|inner| { out.push(map(*id, inner)); true - } - None => false, + }) }); out } diff --git a/src/test_helpers/connection_startup.rs b/src/test_helpers/connection_startup.rs index 35230293e..319ef1428 100644 --- a/src/test_helpers/connection_startup.rs +++ b/src/test_helpers/connection_startup.rs @@ -229,7 +229,7 @@ impl TransformCountingMiddleware { /// assert_eq!(layer.tag(), b'A'); /// ``` #[must_use] - pub fn new(tag: u8, transforms: Arc) -> Self { Self { tag, transforms } } + pub const fn new(tag: u8, transforms: Arc) -> Self { Self { tag, transforms } } } impl TransformCountingMiddleware { diff --git a/src/test_helpers/frame_codec.rs b/src/test_helpers/frame_codec.rs index 7aec053f6..6c731f9d0 100644 --- a/src/test_helpers/frame_codec.rs +++ b/src/test_helpers/frame_codec.rs @@ -56,7 +56,7 @@ pub struct TestAdapter { impl TestAdapter { /// Create a new adapter with the given maximum frame length. #[must_use] - pub fn new(max_frame_length: usize) -> Self { Self { max_frame_length } } + pub const fn new(max_frame_length: usize) -> Self { Self { max_frame_length } } } impl Decoder for TestAdapter { diff --git a/src/test_helpers/mod.rs b/src/test_helpers/mod.rs index ea1b26ec4..548ac5c47 100644 --- a/src/test_helpers/mod.rs +++ b/src/test_helpers/mod.rs @@ -65,10 +65,10 @@ struct FrameFlags(u8); impl FrameFlags { /// Whether this frame terminates the logical message. - fn is_last(self) -> bool { self.0 & 0b1 == 0b1 } + const fn is_last(self) -> bool { self.0 & 0b1 == 0b1 } /// Whether the header carries its optional length/sequence value. - fn has_optional_field(self) -> bool { self.0 & 0b10 == 0b10 } + const fn has_optional_field(self) -> bool { self.0 & 0b10 == 0b10 } } /// Parse a protocol-specific frame header for tests. @@ -222,8 +222,8 @@ pub fn first_frame_payload( payload.put_u16(0); let body_len = u32::try_from(body.len()).map_err(|_| invalid_data("body length too large"))?; payload.put_u32(body_len); - if let Some(total) = total { - payload.put_u32(total); + if let Some(total_body_len) = total { + payload.put_u32(total_body_len); } payload.extend_from_slice(body); Ok(payload.to_vec()) diff --git a/src/testkit/reassembly/fragment.rs b/src/testkit/reassembly/fragment.rs index f8c497bc3..3360641c5 100644 --- a/src/testkit/reassembly/fragment.rs +++ b/src/testkit/reassembly/fragment.rs @@ -22,7 +22,7 @@ pub struct FragmentReassemblySnapshot<'a> { impl<'a> FragmentReassemblySnapshot<'a> { /// Create a snapshot from the caller's current reassembly state. #[must_use] - pub fn new( + pub const fn new( last_reassembled: Option<&'a ReassembledMessage>, last_error: Option<&'a ReassemblyError>, evicted_ids: &'a [MessageId], diff --git a/src/testkit/reassembly/message.rs b/src/testkit/reassembly/message.rs index a1cc4c186..cf47ab9a9 100644 --- a/src/testkit/reassembly/message.rs +++ b/src/testkit/reassembly/message.rs @@ -31,7 +31,7 @@ impl<'a> MessageAssemblySnapshot<'a> { reason = "snapshot constructor mirrors the observable message-assembly state" )] #[must_use] - pub fn new( + pub const fn new( last_result: Option<&'a Result, MessageAssemblyError>>, completed_messages: &'a [AssembledMessage], evicted_keys: &'a [MessageKey], @@ -49,7 +49,7 @@ impl<'a> MessageAssemblySnapshot<'a> { /// Extract the last completed message from the snapshot. #[must_use] - pub fn last_completed(&self) -> Option<&'a AssembledMessage> { + pub const fn last_completed(&self) -> Option<&'a AssembledMessage> { match self.last_result { Some(Ok(Some(message))) => Some(message), _ => None, @@ -199,8 +199,8 @@ fn describe_last_result( last_result: Option<&Result, MessageAssemblyError>>, ) -> String { match last_result { - None => "no result recorded".to_string(), - Some(Ok(None)) => "incomplete assembly".to_string(), + None => "no result recorded".to_owned(), + Some(Ok(None)) => "incomplete assembly".to_owned(), Some(Ok(Some(message))) => { format!("completed assembly for key {}", message.message_key()) } diff --git a/src/testkit/reassembly/message_error.rs b/src/testkit/reassembly/message_error.rs index b52de2d0f..0e4237be5 100644 --- a/src/testkit/reassembly/message_error.rs +++ b/src/testkit/reassembly/message_error.rs @@ -72,9 +72,10 @@ pub(super) fn matches_message_error( expected: MessageAssemblyErrorExpectation, ) -> bool { match expected { - MessageAssemblyErrorExpectation::SequenceMismatch { expected, found } => { - matches_sequence_mismatch(err, expected, found) - } + MessageAssemblyErrorExpectation::SequenceMismatch { + expected: expected_sequence, + found, + } => matches_sequence_mismatch(err, expected_sequence, found), MessageAssemblyErrorExpectation::DuplicateFrame { key, sequence } => { matches_duplicate_frame(err, key, sequence) } diff --git a/src/testkit/result.rs b/src/testkit/result.rs index c339cb9f0..48db59239 100644 --- a/src/testkit/result.rs +++ b/src/testkit/result.rs @@ -102,7 +102,7 @@ impl From for TestError { } impl From<&str> for TestError { - fn from(value: &str) -> Self { Self::Msg(value.to_string()) } + fn from(value: &str) -> Self { Self::Msg(value.to_owned()) } } impl From> for TestError { diff --git a/src/testkit/slow_io.rs b/src/testkit/slow_io.rs index 590acd0e2..bce644c9d 100644 --- a/src/testkit/slow_io.rs +++ b/src/testkit/slow_io.rs @@ -37,7 +37,9 @@ pub struct SlowIoPacing { impl SlowIoPacing { /// Create a pacing configuration for chunked transfers. #[must_use] - pub fn new(chunk_size: NonZeroUsize, delay: Duration) -> Self { Self { chunk_size, delay } } + pub const fn new(chunk_size: NonZeroUsize, delay: Duration) -> Self { + Self { chunk_size, delay } + } } /// Shared configuration for slow-I/O app driving. @@ -68,21 +70,21 @@ impl SlowIoConfig { /// Set the pacing for bytes written into the app. #[must_use] - pub fn with_writer_pacing(mut self, pacing: SlowIoPacing) -> Self { + pub const fn with_writer_pacing(mut self, pacing: SlowIoPacing) -> Self { self.writer_pacing = Some(pacing); self } /// Set the pacing for bytes read from the app. #[must_use] - pub fn with_reader_pacing(mut self, pacing: SlowIoPacing) -> Self { + pub const fn with_reader_pacing(mut self, pacing: SlowIoPacing) -> Self { self.reader_pacing = Some(pacing); self } /// Set the duplex stream capacity. #[must_use] - pub fn with_capacity(mut self, capacity: usize) -> Self { + pub const fn with_capacity(mut self, capacity: usize) -> Self { self.capacity = capacity; self } @@ -120,14 +122,14 @@ fn validate_pacing_chunk_size( direction: &str, capacity: usize, ) -> io::Result<()> { - if let Some(pacing) = pacing - && pacing.chunk_size.get() > capacity + if let Some(configured_pacing) = pacing + && configured_pacing.chunk_size.get() > capacity { return Err(io::Error::new( io::ErrorKind::InvalidInput, format!( "{direction} chunk size {} must not exceed capacity {}", - pacing.chunk_size.get(), + configured_pacing.chunk_size.get(), capacity ), )); @@ -153,8 +155,8 @@ where { match pacing { None => writer.write_all(bytes).await, - Some(pacing) => { - let step = pacing.chunk_size.get(); + Some(configured_pacing) => { + let step = configured_pacing.chunk_size.get(); let total = bytes.len(); let mut offset = 0; while offset < total { @@ -164,7 +166,7 @@ where .ok_or_else(|| io::Error::other("writer chunk slice out of bounds"))?; writer.write_all(chunk).await?; offset = end; - pause_between_chunks(pacing.delay, offset < total).await; + pause_between_chunks(configured_pacing.delay, offset < total).await; } Ok(()) } @@ -185,12 +187,12 @@ where reader.read_to_end(&mut out).await?; Ok(out) } - Some(pacing) => { + Some(configured_pacing) => { let mut out = Vec::new(); let mut should_pause_before_read = false; - let mut buf = vec![0; pacing.chunk_size.get()]; + let mut buf = vec![0; configured_pacing.chunk_size.get()]; loop { - pause_between_chunks(pacing.delay, should_pause_before_read).await; + pause_between_chunks(configured_pacing.delay, should_pause_before_read).await; let read = reader.read(&mut buf).await?; if read == 0 { break; @@ -218,19 +220,22 @@ where F: FnOnce(DuplexStream) -> Fut, Fut: std::future::Future, { - let config = config.validate()?; + let validated_config = config.validate()?; drive_with_strategies( server_fn, - config.capacity, + validated_config.capacity, |mut writer: WriteHalf| async { - write_with_optional_pacing(&mut writer, &wire_bytes, config.writer_pacing).await?; + write_with_optional_pacing(&mut writer, &wire_bytes, validated_config.writer_pacing) + .await?; Ok(writer) }, |mut reader| async { - match config.reader_pacing { + match validated_config.reader_pacing { None => read_all(reader).await, - Some(_) => read_with_optional_pacing(&mut reader, config.reader_pacing).await, + Some(_) => { + read_with_optional_pacing(&mut reader, validated_config.reader_pacing).await + } } }, ) diff --git a/src/testkit/support.rs b/src/testkit/support.rs index cfe02275b..b2a805083 100644 --- a/src/testkit/support.rs +++ b/src/testkit/support.rs @@ -98,8 +98,8 @@ where let server_fut = panic_guarded_server(server_fn, server); let writer_fut = async { - let mut writer = write_strategy(writer).await?; - writer.shutdown().await?; + let mut paced_writer = write_strategy(writer).await?; + paced_writer.shutdown().await?; io::Result::Ok(()) }; diff --git a/tests/compile_error.rs b/tests/compile_error.rs index 81fbc2df4..72f75ec50 100644 --- a/tests/compile_error.rs +++ b/tests/compile_error.rs @@ -13,6 +13,7 @@ fn compile_tests() { t.pass("tests/ui/codec_shared_const_api.rs"); t.pass("tests/ui/fragment_const_api.rs"); t.pass("tests/ui/app_const_api.rs"); + t.pass("tests/ui/server_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/server_const_api.rs b/tests/ui/server_const_api.rs new file mode 100644 index 000000000..dc768b510 --- /dev/null +++ b/tests/ui/server_const_api.rs @@ -0,0 +1,82 @@ +//! Verifies public server-runtime APIs remain callable in const contexts. +//! +//! The lint-census layer made the push-queue builder setters, `ConnectionId` +//! accessors, `ConnectionInfo::peer_addr`, and several error and codec +//! accessors `const fn`. Const callability is a compile-time promise, so a +//! fixture that compiles is the only check that can catch a regression. +//! +//! Values that no `const fn` can construct, such as `io::Error` or a live +//! connection, arrive as parameters: the point under test is the callee, not +//! the argument. + +use std::{net::SocketAddr, time::Duration}; + +use wireframe::{ + codec::{CodecError, EofError, LengthDelimitedFrameCodec}, + error::WireframeError, + extractor::ConnectionInfo, + push::PushQueuesBuilder, + session::ConnectionId, +}; + +/// Every builder setter chains, and each stays `const`, inside a `const fn`. +/// The frame parameter stays generic: `FrameLike` is only crate-visible, but +/// its blanket implementation means `F: Send + 'static` is all a caller needs. +const fn configure_builder( + builder: PushQueuesBuilder, +) -> PushQueuesBuilder { + builder + .high_capacity(8) + .low_capacity(4) + .rate(Some(64)) + .dlq_log_every_n(2) + .dlq_log_interval(Duration::from_secs(5)) +} + +/// Removing the rate limiter and restoring one both stay `const`. +const fn recalibrate_builder( + builder: PushQueuesBuilder, +) -> PushQueuesBuilder { + builder.unlimited().rate(Some(16)) +} + +const CONNECTION_ID: ConnectionId = ConnectionId::new(42); +const CONNECTION_ID_VALUE: u64 = CONNECTION_ID.as_u64(); + +/// The codec accessor is const-callable. A codec value is not `const` +/// constructible, so it arrives as a parameter. +const fn frame_limit(codec: &LengthDelimitedFrameCodec) -> usize { + codec.max_frame_length() +} + +/// `ConnectionInfo::peer_addr` is const; a live value arrives as a parameter. +const fn peer_addr_of(info: &ConnectionInfo) -> Option { info.peer_addr() } + +/// `WireframeError::from_codec` is const, and the codec error it wraps is +/// const-constructible, so the whole value is built in a `const` item. +const CLEAN_CLOSE_CODEC: CodecError = CodecError::Eof(EofError::CleanClose); +const CLEAN_CLOSE_ERROR: WireframeError = WireframeError::from_codec(CLEAN_CLOSE_CODEC); + +/// A `WireframeError` has a non-const destructor, so it can be neither dropped +/// nor consumed inside a `const fn`. Borrowing it keeps the value alive and +/// still exercises the const `is_clean_close` receiver. +const fn is_clean_close(error: &WireframeError) -> bool { error.is_clean_close() } + +/// Returning the error moves it out instead of dropping it, which is the only +/// shape that lets a `const fn` wrap a non-const-constructible payload. +const fn wrap_io(error: std::io::Error) -> WireframeError { + WireframeError::from_io(error) +} + +fn main() { + let _ = ( + configure_builder::>, + recalibrate_builder::>, + CONNECTION_ID_VALUE, + frame_limit, + peer_addr_of, + CLEAN_CLOSE_ERROR.is_clean_close(), + is_clean_close(&CLEAN_CLOSE_ERROR), + wrap_io, + ); +} diff --git a/wireframe_testing/src/benchmark_alloc_support.rs b/wireframe_testing/src/benchmark_alloc_support.rs index 67a751c08..c914669c0 100644 --- a/wireframe_testing/src/benchmark_alloc_support.rs +++ b/wireframe_testing/src/benchmark_alloc_support.rs @@ -210,7 +210,7 @@ mod tests { #[test] #[serial] fn counting_window_closes_after_failure() { - let error = count_allocations(|| Err("deliberate failure".to_string())) + let error = count_allocations(|| Err("deliberate failure".to_owned())) .expect_err("a failing operation must surface its error"); assert_eq!(error, "deliberate failure"); assert!( diff --git a/wireframe_testing/src/client_pair.rs b/wireframe_testing/src/client_pair.rs index f4ff1d4ae..70832bdf5 100644 --- a/wireframe_testing/src/client_pair.rs +++ b/wireframe_testing/src/client_pair.rs @@ -139,7 +139,9 @@ impl WireframePair { // Send the shutdown signal. if let Some(shutdown_tx) = running.shutdown_tx.take() { - let _ = shutdown_tx.send(()); + // The receiver may already have dropped; teardown is not a place + // to report that. + shutdown_tx.send(()).ok(); } // Await the server task through a mutable borrow. @@ -166,7 +168,7 @@ impl Drop for WireframePair { fn drop(&mut self) { if let Some(running) = self.running.take() { if let Some(shutdown_tx) = running.shutdown_tx { - let _ = shutdown_tx.send(()); + shutdown_tx.send(()).ok(); } if let Some(handle) = running.handle { spawn_bounded_shutdown(handle, std::time::Duration::from_millis(100)); @@ -222,13 +224,13 @@ struct PendingServer(Option); impl PendingServer { /// Take the shutdown channel and handle out of the guard, consuming it. - fn take(&mut self) -> Option { self.0.take() } + const fn take(&mut self) -> Option { self.0.take() } } impl Drop for PendingServer { fn drop(&mut self) { if let Some((shutdown_tx, handle)) = self.0.take() { - let _ = shutdown_tx.send(()); + shutdown_tx.send(()).ok(); spawn_bounded_shutdown(handle, std::time::Duration::from_millis(100)); } } @@ -304,7 +306,7 @@ where server .ready_signal(ready_tx) .run_with_shutdown(async { - let _ = shutdown_rx.await; + shutdown_rx.await.ok(); }) .await }); @@ -315,10 +317,10 @@ where // Wait for the server to signal ready. if ready_rx.await.is_err() { // Server dropped ready_tx without signalling - join the task to get the real error. - let (_, handle) = pending + let (_, startup_handle) = pending .take() .ok_or_else(|| TestError::Msg("pending server already taken".into()))?; - return match handle.await { + return match startup_handle.await { Err(join_err) => Err(TestError::Msg(format!( "server task failed to start: {join_err}" ))), @@ -336,7 +338,7 @@ where let client = builder.connect(addr).await?; // Now that the client is connected, take the pending server out of the guard. - let (shutdown_tx, handle) = pending + let (running_shutdown, running_handle) = pending .take() .ok_or_else(|| TestError::Msg("pending server already taken".into()))?; @@ -344,8 +346,8 @@ where addr, running: Some(Running { client: Some(client), - shutdown_tx: Some(shutdown_tx), - handle: Some(handle), + shutdown_tx: Some(running_shutdown), + handle: Some(running_handle), }), }) } diff --git a/wireframe_testing/src/codec_benchmarks/codec_fragmentation_benchmark_support.rs b/wireframe_testing/src/codec_benchmarks/codec_fragmentation_benchmark_support.rs index 8e8c1dfb6..199b32e53 100644 --- a/wireframe_testing/src/codec_benchmarks/codec_fragmentation_benchmark_support.rs +++ b/wireframe_testing/src/codec_benchmarks/codec_fragmentation_benchmark_support.rs @@ -112,9 +112,9 @@ pub fn measure_fragmented_wrap( fragment_payload_cap: usize, ) -> Result { let payload = payload_for_class(payload_class); - let fragment_payload_cap = + let validated_cap = NonZeroUsize::new(fragment_payload_cap).ok_or("fragment payload cap must be non-zero")?; - let fragmenter = Fragmenter::new(fragment_payload_cap); + let fragmenter = Fragmenter::new(validated_cap); let codec = LengthDelimitedFrameCodec::new(LARGE_PAYLOAD_BYTES + 4096); let started = std::time::Instant::now(); let mut total_bytes = 0_u64; diff --git a/wireframe_testing/src/helpers/codec_fixtures.rs b/wireframe_testing/src/helpers/codec_fixtures.rs index 3ae99d682..47c9825a6 100644 --- a/wireframe_testing/src/helpers/codec_fixtures.rs +++ b/wireframe_testing/src/helpers/codec_fixtures.rs @@ -95,10 +95,10 @@ pub fn valid_hotline_frame( payload: &[u8], transaction_id: impl Into, ) -> HotlineFrame { - let transaction_id = transaction_id.into().0; + let resolved_transaction_id = transaction_id.into().0; let codec = HotlineFrameCodec::new(payload.len()); let mut frame = codec.wrap_payload(Bytes::copy_from_slice(payload)); - frame.transaction_id = transaction_id; + frame.transaction_id = resolved_transaction_id; frame } @@ -122,8 +122,8 @@ pub fn valid_hotline_frame( /// ``` #[must_use] pub fn oversized_hotline_wire(max_frame_length: impl Into) -> Vec { - let max_frame_length = max_frame_length.into().0; - let oversized_len = max_frame_length.saturating_add(1); + let resolved_max = max_frame_length.into().0; + let oversized_len = resolved_max.saturating_add(1); let payload = vec![0xab; oversized_len]; build_hotline_wire(HotlineWireFixture { payload: &payload, @@ -213,14 +213,14 @@ pub fn truncated_hotline_header() -> Vec { pub fn truncated_hotline_payload(payload_len: impl Into) -> Vec { // Clamp to at least 1 so the header always claims more payload than // the buffer provides, guaranteeing a genuinely truncated frame. - let payload_len = payload_len.into().0.max(1); + let declared_len = payload_len.into().0.max(1); let header = HotlineHeader { - data_size: u32_from_usize(payload_len), - total_size: u32_from_usize(payload_len.saturating_add(HEADER_LEN)), + data_size: u32_from_usize(declared_len), + total_size: u32_from_usize(declared_len.saturating_add(HEADER_LEN)), transaction_id: TransactionId(0), }; - let half_payload = payload_len >> 1; + let half_payload = declared_len >> 1; let mut buf = Vec::with_capacity(HEADER_LEN + half_payload); append_hotline_header(&mut buf, header); buf.extend_from_slice(&vec![0xcc; half_payload]); @@ -251,10 +251,10 @@ pub fn correlated_hotline_wire( transaction_id: impl Into, payloads: &[&[u8]], ) -> Vec { - let transaction_id = transaction_id.into().0; + let resolved_transaction_id = transaction_id.into().0; let mut buf = Vec::new(); for payload in payloads { - buf.extend_from_slice(&valid_hotline_wire(payload, transaction_id)); + buf.extend_from_slice(&valid_hotline_wire(payload, resolved_transaction_id)); } buf } @@ -282,14 +282,14 @@ pub fn sequential_hotline_wire( base_transaction_id: impl Into, payloads: &[&[u8]], ) -> Vec { - let base_transaction_id = base_transaction_id.into().0; + let first_transaction_id = base_transaction_id.into().0; let mut buf = Vec::new(); for (i, payload) in payloads.iter().enumerate() { #[expect( clippy::cast_possible_truncation, reason = "fixture payloads slice length will not exceed u32::MAX" )] - let tid = base_transaction_id.wrapping_add(i as u32); + let tid = first_transaction_id.wrapping_add(i as u32); buf.extend_from_slice(&valid_hotline_wire(payload, tid)); } buf @@ -323,12 +323,12 @@ enum TotalSizePolicy { impl TotalSizePolicy { /// Calculate the declared frame length for the fixture's validation scenario. fn total_size(self, data_size: usize) -> u32 { - let total_size = data_size.saturating_add(HEADER_LEN); - let total_size = match self { - Self::Matching => total_size, - Self::OffByOne => total_size.saturating_add(1), + let base_total = data_size.saturating_add(HEADER_LEN); + let declared_total = match self { + Self::Matching => base_total, + Self::OffByOne => base_total.saturating_add(1), }; - u32_from_usize(total_size) + u32_from_usize(declared_total) } } diff --git a/wireframe_testing/src/helpers/drive.rs b/wireframe_testing/src/helpers/drive.rs index b36c8d1a3..dffefea94 100644 --- a/wireframe_testing/src/helpers/drive.rs +++ b/wireframe_testing/src/helpers/drive.rs @@ -51,7 +51,7 @@ where .catch_unwind() .await; match result { - Ok(result) => result, + Ok(server_result) => server_result, Err(panic) => { let panic_msg = wireframe::panic::format_panic(&panic); Err(io::Error::other(format!("server task failed: {panic_msg}"))) diff --git a/wireframe_testing/src/helpers/runtime.rs b/wireframe_testing/src/helpers/runtime.rs index 926f9c70c..deffb1ecc 100644 --- a/wireframe_testing/src/helpers/runtime.rs +++ b/wireframe_testing/src/helpers/runtime.rs @@ -43,14 +43,14 @@ where C: Send + 'static, E: Packet, { - let capacity = capacity.unwrap_or(super::DEFAULT_CAPACITY); - if capacity == 0 { + let resolved_capacity = capacity.unwrap_or(super::DEFAULT_CAPACITY); + if resolved_capacity == 0 { return Err(io::Error::new( io::ErrorKind::InvalidInput, "capacity must be greater than zero", )); } - if capacity > MAX_CAPACITY { + if resolved_capacity > MAX_CAPACITY { return Err(io::Error::new( io::ErrorKind::InvalidInput, format!("capacity must not exceed {MAX_CAPACITY} bytes"), @@ -60,7 +60,7 @@ where drive_internal( |server| async move { app.handle_connection_result(server).await }, frames, - capacity, + resolved_capacity, ) .await } diff --git a/wireframe_testing/src/helpers/tests/helper_tests.rs b/wireframe_testing/src/helpers/tests/helper_tests.rs index 3042f7ba1..2c0a3ee64 100644 --- a/wireframe_testing/src/helpers/tests/helper_tests.rs +++ b/wireframe_testing/src/helpers/tests/helper_tests.rs @@ -43,7 +43,7 @@ async fn drive_with_payloads_wraps_frames() -> io::Result<()> { format!("failed to create app: {error}"), ) })?; - let app = app + let routed_app = app .route( 1, Arc::new(|_: &Envelope| -> BoxFuture<'static, ()> { Box::pin(async {}) }), @@ -64,7 +64,7 @@ async fn drive_with_payloads_wraps_frames() -> io::Result<()> { ) })?; - let out = drive_with_payloads(app, vec![encoded]).await?; + let out = drive_with_payloads(routed_app, vec![encoded]).await?; let frames = decode_frames(&out)?; if frames.len() != 1 { return Err(io::Error::new( diff --git a/wireframe_testing/src/integration_helpers.rs b/wireframe_testing/src/integration_helpers.rs index f925f4b13..419c6cb40 100644 --- a/wireframe_testing/src/integration_helpers.rs +++ b/wireframe_testing/src/integration_helpers.rs @@ -164,7 +164,7 @@ impl CommonTestEnvelope { /// assert_eq!(envelope.id, 7); /// ``` #[must_use] - pub fn new(id: u32, correlation_id: Option, payload: Vec) -> Self { + pub const fn new(id: u32, correlation_id: Option, payload: Vec) -> Self { Self { id, correlation_id, @@ -183,7 +183,7 @@ impl CommonTestEnvelope { /// assert_eq!(envelope.correlation_id, None); /// ``` #[must_use] - pub fn with_payload(id: u32, payload: Vec) -> Self { + pub const fn with_payload(id: u32, payload: Vec) -> Self { Self { id, correlation_id: None, @@ -202,7 +202,7 @@ impl CommonTestEnvelope { /// assert_eq!(envelope.payload, Vec::::new()); /// ``` #[must_use] - pub fn with_id(id: u32) -> Self { + pub const fn with_id(id: u32) -> Self { Self { id, correlation_id: None, @@ -287,9 +287,9 @@ pub type CommonHandler = Arc< /// let handler = echo_handler(&counter); /// ``` pub fn echo_handler(counter: &Arc) -> CommonHandler { - let counter = counter.clone(); + let shared_counter = counter.clone(); Arc::new(move |_: &CommonTestEnvelope| { - let c = counter.clone(); + let c = shared_counter.clone(); Box::pin(async move { c.fetch_add(1, Ordering::SeqCst); }) diff --git a/wireframe_testing/src/logging.rs b/wireframe_testing/src/logging.rs index 0c86198ce..4c779ce68 100644 --- a/wireframe_testing/src/logging.rs +++ b/wireframe_testing/src/logging.rs @@ -138,11 +138,11 @@ mod tests { .join(); assert!(join_result.is_err(), "poisoning thread should panic"); - let mut handle = LoggerHandle::default(); + let mut recovered = LoggerHandle::default(); assert!( - handle.pop().is_none(), + recovered.pop().is_none(), "poison recovery should drain stale log records" ); - handle.clear(); + recovered.clear(); } } diff --git a/wireframe_testing/src/observability/assertions.rs b/wireframe_testing/src/observability/assertions.rs index c0077f525..901d16b6c 100644 --- a/wireframe_testing/src/observability/assertions.rs +++ b/wireframe_testing/src/observability/assertions.rs @@ -47,13 +47,13 @@ impl ObservabilityHandle { labels: impl Into, expected: u64, ) -> Result<(), String> { - let labels = labels.into(); - let actual = find_counter(&self.captured, name, &labels.as_str_pairs()); + let resolved = labels.into(); + let actual = find_counter(&self.captured, name, &resolved.as_str_pairs()); if actual == expected { Ok(()) } else { Err(format!( - "counter {name} with labels {labels:?}: expected {expected}, got {actual}" + "counter {name} with labels {resolved:?}: expected {expected}, got {actual}" )) } } @@ -81,13 +81,13 @@ impl ObservabilityHandle { name: &str, labels: impl Into, ) -> Result<(), String> { - let labels = labels.into(); + let resolved = labels.into(); let recorded = self .captured .iter() .filter(|(key, ..)| key.key().name() == name) .filter(|(key, ..)| { - labels + resolved .as_str_pairs() .iter() .all(|(label, value)| { @@ -101,7 +101,7 @@ impl ObservabilityHandle { Ok(()) } else { Err(format!( - "histogram {name} with labels {labels:?} did not record a value" + "histogram {name} with labels {resolved:?} did not record a value" )) } } @@ -193,8 +193,8 @@ impl ObservabilityHandle { /// ``` pub fn assert_log_contains(&mut self, substring: &str) -> Result<(), String> { self.assert_log_matches( - |record| record.args().to_string().contains(substring), - |record| record.args().to_string(), + |record| record.args().contains(substring), + |record| record.args().to_owned(), |records| format!("no log record containing {substring:?}; captured: {records:?}"), ) } @@ -220,7 +220,7 @@ impl ObservabilityHandle { /// ``` pub fn assert_log_at_level(&mut self, level: Level, substring: &str) -> Result<(), String> { self.assert_log_matches( - |record| record.level() == level && record.args().to_string().contains(substring), + |record| record.level() == level && record.args().contains(substring), |record| format!("[{}] {}", record.level(), record.args()), |records| format!("no {level} log containing {substring:?}; captured: {records:?}"), ) diff --git a/wireframe_testing/src/observability/mod.rs b/wireframe_testing/src/observability/mod.rs index bb2e95cad..4f2665307 100644 --- a/wireframe_testing/src/observability/mod.rs +++ b/wireframe_testing/src/observability/mod.rs @@ -134,7 +134,7 @@ impl ObservabilityHandle { /// log::warn!("example"); /// assert!(obs.logs().pop().is_some()); /// ``` - pub fn logs(&mut self) -> &mut LoggerHandle { &mut self.logger } + pub const fn logs(&mut self) -> &mut LoggerHandle { &mut self.logger } /// Access the metrics recorder for use with /// [`metrics::with_local_recorder`]. @@ -150,7 +150,7 @@ impl ObservabilityHandle { /// }); /// ``` #[must_use] - pub fn recorder(&self) -> &DebuggingRecorder { &self.recorder } + pub const fn recorder(&self) -> &DebuggingRecorder { &self.recorder } /// Take a snapshot of the current metrics state. /// @@ -223,8 +223,8 @@ impl ObservabilityHandle { /// ``` #[must_use] pub fn counter(&self, name: &str, labels: impl Into) -> u64 { - let labels = labels.into(); - find_counter(&self.captured, name, &labels.as_str_pairs()) + let resolved = labels.into(); + find_counter(&self.captured, name, &resolved.as_str_pairs()) } /// Query a counter value by name, aggregating across all label sets. diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/contended.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/contended.rs index 821607e62..6e8583b50 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/contended.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/contended.rs @@ -141,15 +141,15 @@ pub(super) async fn contended_round_bounded( observer: &PoolTaskChurnObserver, ) -> Result { let burst = contended_round(workload, pool, observer); - match tokio::time::timeout(CONTENDED_BURST_TIMEOUT, burst).await { - Ok(result) => result, - Err(_) => Err(format!( + let Ok(result) = tokio::time::timeout(CONTENDED_BURST_TIMEOUT, burst).await else { + return Err(format!( "contended burst did not complete within {CONTENDED_BURST_TIMEOUT:?}: every contender \ is stalled with waiters still queued (pool acquire-path hang, last observed before \ #535 and not reproduced since; see docs/runtime-ownership-baselines.md and issue \ #647)" - )), - } + )); + }; + result } /// Empty the pool, prove contenders had to wait for it, then run the rounds. @@ -211,19 +211,21 @@ fn contend( let mut tasks = Vec::with_capacity(contenders); for _ in 0..contenders { let mut handle = pool.handle(); - let barrier = Arc::clone(&barrier); - let granted = Arc::clone(&granted); - let failures = Arc::clone(&failures); + // Each contender needs its own `Arc` clone; the originals stay alive so + // nothing is dropped while other contenders are still starting. + let contender_barrier = Arc::clone(&barrier); + let contender_granted = Arc::clone(&granted); + let contender_failures = Arc::clone(&failures); tasks.push(tokio::spawn(async move { - barrier.wait().await; + contender_barrier.wait().await; match acquire_round(&mut handle, rounds_per_contender).await { Ok(count) => { - granted.fetch_add( + contender_granted.fetch_add( usize::try_from(count).unwrap_or(usize::MAX), Ordering::SeqCst, ); } - Err(err) => push_failure(&failures, &err), + Err(err) => push_failure(&contender_failures, &err), } })); } @@ -263,9 +265,8 @@ async fn awaits_forced_waits( } }; - match tokio::time::timeout(CONTENDED_WAIT_TIMEOUT, pending).await { - Ok(waits) => Ok(waits), - Err(_) => Err(format!( + let Ok(waits) = tokio::time::timeout(CONTENDED_WAIT_TIMEOUT, pending).await else { + return Err(format!( "pool was exhausted, so {contenders} contenders had to wait for capacity, but only {} \ waits were recorded within {CONTENDED_WAIT_TIMEOUT:?}; the exhaustion or its \ instrumentation is not working", @@ -273,8 +274,9 @@ async fn awaits_forced_waits( .snapshot() .capacity_waits .saturating_sub(waits_before) - )), - } + )); + }; + Ok(waits) } /// Record one contender's acquisition failure without losing the others. diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/distribution/mod.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/distribution/mod.rs index fa00efb4c..6c9b76b56 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/distribution/mod.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/distribution/mod.rs @@ -97,7 +97,7 @@ impl ContendedLatency { } /// Fold one burst's churn into the run's total. - pub(super) fn push_churn(&mut self, churn: PoolTaskChurn) { + pub(super) const fn push_churn(&mut self, churn: PoolTaskChurn) { self.churn = self.churn.saturating_add(churn); } @@ -111,7 +111,7 @@ impl ContendedLatency { /// assert_eq!(ContendedLatency::default().successes(), 0); /// ``` #[must_use] - pub fn successes(&self) -> usize { self.samples.len() } + pub const fn successes(&self) -> usize { self.samples.len() } /// Bursts that failed, whether or not their message was retained. /// diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/distribution/tests.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/distribution/tests.rs index bb5e69f96..58b3d87d5 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/distribution/tests.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/distribution/tests.rs @@ -119,7 +119,7 @@ fn failures_are_counted_in_full_and_retained_in_bounded_memory() { assert_eq!(latency.failures().len(), RETAINED_FAILURES); assert_eq!( latency.failures().first(), - Some(&"burst 0 failed".to_string()), + Some(&"burst 0 failed".to_owned()), "retention keeps the earliest messages, which name the first occurrence" ); } diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/phase.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/phase.rs index 298106688..c249dedb6 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/phase.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/measurement/phase.rs @@ -142,7 +142,7 @@ pub(super) async fn acquisition_phase( /// those rounds could run. Folding the second into the first would put /// `pool_size` eager loopback handshakes back inside the figure the scope exists /// to report. -pub(super) fn finish( +pub(super) const fn finish( outcome: PhaseOutcome, elapsed: Duration, construction: Duration, diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs index 52e8e1a56..28cfb93e7 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs @@ -105,7 +105,7 @@ impl AcquisitionMode { } /// Build the pool configuration this mode requires. - pub(super) fn configure( + pub(super) const fn configure( self, config: wireframe::client::ClientPoolConfig, ) -> wireframe::client::ClientPoolConfig { @@ -198,7 +198,7 @@ impl PoolWorkload { /// Reject workloads whose regime cannot produce the documented outcome. pub(super) fn validate(self) -> Result<(), String> { if self.pool_size == 0 { - return Err("pool_size must be at least 1".to_string()); + return Err("pool_size must be at least 1".to_owned()); } if self.mode.is_contended() && self.mode.contenders(self.contenders) <= self.pool_size { return Err(format!( diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/connection_actor_benchmark_support/workload.rs b/wireframe_testing/src/runtime_ownership_benchmarks/connection_actor_benchmark_support/workload.rs index f0357cb18..fa04c7866 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/connection_actor_benchmark_support/workload.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/connection_actor_benchmark_support/workload.rs @@ -148,12 +148,12 @@ impl ActorWorkload { /// Reject workloads the harness cannot measure as described. pub(super) fn validate(self) -> Result<(), String> { if self.queue_capacity == 0 { - return Err("queue_capacity must be at least 1".to_string()); + return Err("queue_capacity must be at least 1".to_owned()); } let headroom = self .queue_capacity .checked_mul(2) - .ok_or_else(|| "queue_capacity is too large to double".to_string())?; + .ok_or_else(|| "queue_capacity is too large to double".to_owned())?; if self.frames > headroom { return Err(format!( "frames {} exceeds total queue capacity {}; preloading would block on \ diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/connection_startup_benchmark_support/workload.rs b/wireframe_testing/src/runtime_ownership_benchmarks/connection_startup_benchmark_support/workload.rs index 0bb46a15d..738ad8bed 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/connection_startup_benchmark_support/workload.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/connection_startup_benchmark_support/workload.rs @@ -166,7 +166,7 @@ impl StartupWorkload { /// Reject workloads whose accounting cannot be asserted. pub(super) fn validate(self) -> Result<(), String> { if self.connections == 0 { - return Err("connections must be at least 1".to_string()); + return Err("connections must be at least 1".to_owned()); } // `StartupFixturePlan::new` floors both counts at one, so a zero here // would silently measure a different application than the label claims. diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement.rs b/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement/mod.rs similarity index 99% rename from wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement.rs rename to wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement/mod.rs index b5f37e713..7114027bd 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement/mod.rs @@ -72,7 +72,7 @@ impl SendOutcome { /// Saturating rather than wrapping, in keeping with the counters' own /// saturating arithmetic: a run long enough to overflow is a misconfigured /// workload, and a wrapped total would look deceptively small. - pub(super) fn accumulate(self, running: Self) -> Self { + pub(super) const fn accumulate(self, running: Self) -> Self { Self { accepted: running.accepted.saturating_add(self.accepted), drained: running.drained.saturating_add(self.drained), diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement/overflow.rs b/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement/overflow.rs index 7da16abb6..2323c302f 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement/overflow.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/measurement/overflow.rs @@ -27,7 +27,7 @@ async fn fill_to_capacity( match sent { Ok(()) => filled += 1, Err(PushError::Closed) => { - return Err("queue closed while filling to capacity".to_string()); + return Err("queue closed while filling to capacity".to_owned()); } // `QueueFull` is the expected stop; `PushError` is non-exhaustive, // so any future variant also stops here rather than being mistaken @@ -63,7 +63,7 @@ pub(super) async fn send_overflow(workload: PushWorkload) -> Result outcome.accepted += 1, Err(PushError::QueueFull) => outcome.reported_queue_full = true, Err(PushError::Closed) => { - return Err("queue closed during overflow sends".to_string()); + return Err("queue closed during overflow sends".to_owned()); } // Non-exhaustive enum: an unrecognized error is reported as a // diagnostic rather than being silently counted as a drop. diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/scenario.rs b/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/scenario.rs index 627f717cc..ce34a0173 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/scenario.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/push_benchmark_support/scenario.rs @@ -193,7 +193,7 @@ impl PushScenario { /// assert!(PushScenario::OverflowDrop.policy().is_some()); /// ``` #[must_use] - pub fn policy(self) -> Option { + pub const fn policy(self) -> Option { match self { Self::OverflowReturnError => Some(PushPolicy::ReturnErrorIfFull), Self::OverflowDrop => Some(PushPolicy::DropIfFull), @@ -301,10 +301,10 @@ impl PushWorkload { /// Returns a description of whichever constraint the workload breaks. pub(super) fn validate(self) -> Result<(), String> { if self.queue_capacity == 0 { - return Err("queue_capacity must be at least 1".to_string()); + return Err("queue_capacity must be at least 1".to_owned()); } if self.sends == 0 { - return Err("sends must be at least 1".to_string()); + return Err("sends must be at least 1".to_owned()); } if self.scenario.is_uncontended() && self.sends > self.queue_capacity { return Err(format!(