Refine DLQ logging and showcase unlimited builder

Author: leynos

docs/efficiency-report.md

@@ -89,21 +89,22 @@ known.
 
 ### Frame Processing Pipeline
 
-- **Bottleneck**: Frame decode/encode operations in high-throughput scenarios
-- **Critical Path**: `LengthPrefixedProcessor::decode` method
-- **Optimization Priority**: High—affects every incoming frame
+- **Bottleneck**: The frame decoding and encoding operations in
+  high-throughput scenarios
+- **Critical path**: `LengthPrefixedProcessor::decode`
+- **Optimisation priority**: High—affects every incoming frame
 
 ### Connection Handling
 
-- **Bottleneck**: Connection actor event loop and fairness tracking
-- **Critical Path**: `tokio::select!` in connection actor
-- **Optimization Priority**: Medium—affects per-connection performance
+- **Bottleneck**: The connection actor’s event loop and fairness tracking
+- **Critical path**: `tokio::select!` in the connection actor
+- **Optimisation priority**: Medium—affects per-connection performance
 
 ### Message Routing
 
-- **Bottleneck**: HashMap lookups for route resolution
-- **Critical Path**: Route handler lookup in `WireframeApp`
-- **Optimization Priority**: Low—HashMap lookups are already efficient
+- **Bottleneck**: The `HashMap` lookups for route resolution
+- **Critical path**: Route handler lookup in `WireframeApp`
+- **Optimisation priority**: Low—`HashMap` lookups are already efficient
 
 ## Implemented Optimizations
 
@@ -133,8 +134,8 @@ Ok(Some(src.split_to(len).freeze().to_vec()))
 
 ## Future Optimization Opportunities
 
-1. **Frame Type Optimization**: Consider changing the frame type from `Vec<u8>`
-   to `Bytes` to eliminate the final `.to_vec()` call entirely.
+1. **Frame type optimisation**: Replace `Vec<u8>` with `Bytes` end to end to
+   eliminate the final `.to_vec()` copy on the hot path.
 
 2. **Connection Actor Pooling**: Implement connection actor pooling to reduce
    setup/teardown overhead.

docs/hardening-wireframe-a-guide-to-production-resilience.md

@@ -260,11 +260,11 @@ let (queues, handle) = PushQueues::<Frame>::builder()
     .build()
     .expect("rate within supported bounds");
 
-// Passing `None` disables rate limiting entirely:
+// Calling `unlimited` disables rate limiting entirely:
 let (_unlimited, _handle) = PushQueues::<Frame>::builder()
     .high_capacity(8)
     .low_capacity(8)
-    .rate(None)
+    .unlimited()
     .build()
     .expect("failed to build unlimited queues");
 

src/push/mod.rs

@@ -11,6 +11,7 @@
 //! let (queues, handle) = PushQueues::<u8>::builder()
 //!     .high_capacity(8)
 //!     .low_capacity(8)
+//!     .unlimited() // disables rate limiting
 //!     .build()
 //!     .expect("failed to build PushQueues");
 //! # drop((queues, handle));

src/push/queues/builder.rs

@@ -124,8 +124,11 @@ impl<F: FrameLike> PushQueuesBuilder<F> {
     }
 
     /// Log dropped frames every `n` occurrences when the DLQ is full or closed.
+    /// 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 {
+        // Treat 0 as “disabled by count”; interval-based logging remains.
         self.dlq_log_every_n = n;
         self
     }
@@ -140,7 +143,8 @@ impl<F: FrameLike> PushQueuesBuilder<F> {
     /// Build the configured [`PushQueues`] and associated [`PushHandle`].
     ///
     /// Validation of capacities and rate occurs only during the build step,
-    /// not when setting individual fields.
+    /// not when setting individual fields. A `dlq_log_every_n` of `0` disables
+    /// count-based logging; interval-based logging still applies.
     ///
     /// # Errors
     ///

src/push/queues/errors.rs

@@ -6,7 +6,7 @@ use super::MAX_PUSH_RATE;
 
 /// Errors that can occur when pushing a frame.
 #[non_exhaustive]
-#[derive(Debug, Error)]
+#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
 pub enum PushError {
     /// The queue was at capacity and the policy was `ReturnErrorIfFull`.
     #[error("push queue full")]
@@ -18,12 +18,12 @@ pub enum PushError {
 
 /// Errors returned when creating push queues.
 #[non_exhaustive]
-#[derive(Debug, Error)]
+#[derive(Debug, Error, Clone, PartialEq, Eq)]
 pub enum PushConfigError {
-    /// The provided rate was zero or exceeded [`MAX_PUSH_RATE`].
+    /// The provided rate was zero or exceeded [`crate::push::queues::MAX_PUSH_RATE`].
     #[error("invalid rate {0}; must be between 1 and {max}", max = MAX_PUSH_RATE)]
     InvalidRate(usize),
     /// The provided capacities were zero.
-    #[error("invalid capacities; high={high}, low={low}; each must be ≥ 1")]
+    #[error("invalid capacities; high={high}, low={low}; each must be >= 1")]
     InvalidCapacity { high: usize, low: usize },
 }

src/push/queues/handle.rs

@@ -154,7 +154,7 @@ impl<F: FrameLike> PushHandle<F> {
                     let dropped = self.0.dlq_drops.fetch_add(1, Ordering::Relaxed) + 1;
                     let mut last = self.0.dlq_last_log.lock().expect("lock poisoned");
                     let now = Instant::now();
-                    if dropped.is_multiple_of(log_every_n)
+                    if (log_every_n != 0 && dropped.is_multiple_of(log_every_n))
                         || now.duration_since(*last) > log_interval
                     {
                         warn!(

src/push/queues/mod.rs

@@ -149,11 +149,7 @@ impl<F: FrameLike> PushQueues<F> {
             limiter,
             dlq_tx: dlq,
             dlq_drops: AtomicUsize::new(0),
-            dlq_last_log: Mutex::new(
-                Instant::now()
-                    .checked_sub(dlq_log_interval)
-                    .unwrap_or_else(Instant::now),
-            ),
+            dlq_last_log: Mutex::new(Instant::now()),
             dlq_log_every_n,
             dlq_log_interval,
         };

tests/advanced/concurrency_loom.rs

@@ -24,6 +24,7 @@ fn concurrent_push_delivery() {
             let (queues, handle) = PushQueues::<u8>::builder()
                 .high_capacity(1)
                 .low_capacity(1)
+                .unlimited()
                 .build()
                 .expect("failed to build PushQueues");
             let token = CancellationToken::new();

tests/advanced/interaction_fuzz.rs

@@ -15,6 +15,9 @@ use wireframe::{
     response::FrameStream,
 };
 
+#[path = "../support.rs"]
+mod support;
+
 #[derive(Debug, Clone)]
 enum Action {
     High(u8),
@@ -23,9 +26,10 @@ enum Action {
 }
 
 async fn run_actions(actions: &[Action]) -> Vec<u8> {
-    let (queues, handle) = PushQueues::<u8>::builder()
+    let (queues, handle) = support::builder::<u8>()
         .high_capacity(16)
         .low_capacity(16)
+        .unlimited()
         .build()
         .expect("failed to build PushQueues");
     let shutdown = CancellationToken::new();

tests/correlation_id.rs

@@ -18,7 +18,7 @@ async fn stream_frames_carry_request_correlation_id() {
     let (queues, handle) = PushQueues::<Envelope>::builder()
         .high_capacity(1)
         .low_capacity(1)
-        .rate(None)
+        .unlimited()
         .build()
         .expect("failed to build PushQueues");
     let shutdown = CancellationToken::new();