Add loom-based concurrency tests for push queues

Cargo.toml

@@ -68,6 +68,9 @@ tokio = { version = "1.47.1", default-features = false, features = [
     "test-util",
 ] }
 
+[target.'cfg(loom)'.dependencies]
+loom = "0.7.2"
+
 [target.'cfg(not(loom))'.dependencies]
 metrics-exporter-prometheus = { version = "0.17.2", optional = true, features = ["http-listener"] }
 
@@ -106,3 +109,8 @@ required-features = ["examples"]
 name = "cucumber"
 harness = false
 required-features = ["advanced-tests", "cucumber-tests"]
+
+[[test]]
+name = "concurrency_loom"
+path = "tests/advanced/concurrency_loom.rs"
+required-features = ["advanced-tests"]

docs/asynchronous-outbound-messaging-design.md

@@ -478,6 +478,32 @@ flowchart TD
 This API gives developers fine-grained control over both the priority and the
 back-pressure behaviour of their pushed messages.
 
+#### 4.1.1 Loom-based concurrency verification
+
+To reason about concurrent producers we added a loom-specific probe to
+`PushHandle`. The probe exposes the dead-letter queue drop counter when tests
+are compiled with `--cfg loom`. The `tests/advanced/concurrency_loom.rs` suite
+drives `PushHandle::try_push` from multiple `loom::thread`s to assert that drop
+counts reset after the logging threshold across both priority queues, that
+queue-full errors remain deterministic, and that the probe reports zero when
+the DLQ is absent or idle. This keeps the production API unchanged whilst
+enabling exhaustive interleaving checks during the advanced test workflow.
+
+```rust
+#[cfg(loom)]
+{
+    use wireframe::push::queues::PushHandleProbe;
+
+    let probe: PushHandleProbe<_> = handle.probe();
+    let dlq_drops = probe.dlq_drop_count();
+    assert_eq!(dlq_drops, 0);
+}
+```
+
+```
+RUSTFLAGS="--cfg loom" cargo test --features advanced-tests --test concurrency_loom
+```
+
 ### 4.2 The `SessionRegistry`
 
 To allow background tasks to discover and message active connections, a

docs/roadmap.md

@@ -174,7 +174,7 @@ production environments.
   - [x] Implement fuzz testing for the protocol parser
     (`tests/advanced/interaction_fuzz.rs`).
 
-  - [ ] Use `loom` for concurrency testing of shared state
+  - [x] Use `loom` for concurrency testing of shared state
     (`tests/advanced/concurrency_loom.rs`).
 
 ## Phase 6: Multi-Packet Streaming Responses (Priority Focus)

src/push/queues/handle.rs

@@ -1,19 +1,32 @@
 //! Cloneable handle used by producers to push frames to a connection.
 
-use std::{
-    sync::{
+#[cfg(loom)]
+mod sync {
+    pub use std::sync::{Arc, Weak};
+
+    pub use loom::sync::{
+        Mutex,
+        atomic::{AtomicUsize, Ordering},
+    };
+}
+
+#[cfg(not(loom))]
+mod sync {
+    pub use std::sync::{
         Arc,
         Mutex,
         Weak,
         atomic::{AtomicUsize, Ordering},
-    },
-    time::{Duration, Instant},
-};
+    };
+}
+
+use std::time::{Duration, Instant};
 
 use leaky_bucket::RateLimiter;
 use log::{debug, warn};
 use tokio::{sync::mpsc, time::sleep};
 
+use self::sync::{Arc, AtomicUsize, Mutex, Ordering, Weak};
 use super::{FrameLike, PushError, PushPolicy, PushPriority};
 
 /// Shared state for [`PushHandle`].
@@ -40,9 +53,34 @@ pub(crate) struct PushHandleInner<F> {
 #[derive(Clone)]
 pub struct PushHandle<F>(Arc<PushHandleInner<F>>);
 
+/// Instrumentation helper exposing internal counters when running under loom.
+#[cfg(loom)]
+pub struct PushHandleProbe<F> {
+    inner: Arc<PushHandleInner<F>>,
+}
+
+#[cfg(loom)]
+impl<F> PushHandleProbe<F> {
+    /// Return the number of frames dropped into the DLQ since the last log flush.
+    #[must_use]
+    pub fn dlq_drop_count(&self) -> usize { self.inner.dlq_drops.load(Ordering::SeqCst) }
+}
+
 impl<F: FrameLike> PushHandle<F> {
     pub(crate) fn from_arc(arc: Arc<PushHandleInner<F>>) -> Self { Self(arc) }
 
+    /// Returns a probe for inspecting internal state during loom verification.
+    ///
+    /// Exposes dead-letter drop counters so loom tests can assert behaviour
+    /// without altering the production API.
+    #[cfg(loom)]
+    #[must_use]
+    pub fn probe(&self) -> PushHandleProbe<F> {
+        PushHandleProbe {
+            inner: self.0.clone(),
+        }
+    }
+
     /// Internal helper to push a frame with the requested priority.
     ///
     /// IMPORTANT: We honour the rate limiter before attempting to reserve

src/push/queues/mod.rs

@@ -6,12 +6,18 @@
 //! level. An optional rate limiter caps throughput at [`MAX_PUSH_RATE`] pushes
 //! per second.
 
+#[cfg(not(loom))]
+use std::sync::{Mutex, atomic::AtomicUsize};
 use std::{
-    sync::{Arc, Mutex, atomic::AtomicUsize},
+    sync::Arc,
     time::{Duration, Instant},
 };
 
 use leaky_bucket::RateLimiter;
+#[cfg(loom)]
+use loom::sync::Mutex;
+#[cfg(loom)]
+use loom::sync::atomic::AtomicUsize;
 use static_assertions::const_assert;
 use tokio::sync::mpsc;
 

tests/advanced/concurrency_loom.rs

@@ -1,76 +1,171 @@
-#![cfg(feature = "advanced-tests")]
-//! Concurrency tests for push delivery using loom.
+#![cfg(all(feature = "advanced-tests", loom))]
+//! Concurrency tests for push queues using loom.
 //!
-//! These tests model concurrent push execution to validate fairness and
-//! correct shutdown behaviour under various interleavings.
+//! These tests exercise the `PushHandle` shared state without Tokio. `loom`
+//! explores interleavings to ensure DLQ accounting and queue-full errors remain
+//! deterministic under concurrent producers.
 
-use loom::model;
-use tokio::runtime::Builder;
-use tokio_util::sync::CancellationToken;
-use wireframe::{
-    connection::ConnectionActor,
-    push::PushQueues,
-};
+use loom::{model, thread};
+use rstest::rstest;
+use tokio::sync::mpsc;
+use wireframe::push::{PushError, PushPolicy, PushPriority, PushQueues};
+
+#[rstest]
+#[case(PushPriority::High)]
+#[case(PushPriority::Low)]
+fn concurrent_drops_reset_dlq_counter(#[case] priority: PushPriority) {
+    model(move || {
+        let (dlq_tx, mut dlq_rx) = mpsc::channel(4);
+        let (queues, handle) = PushQueues::<u8>::builder()
+            .high_capacity(1)
+            .low_capacity(1)
+            .dlq(Some(dlq_tx))
+            .dlq_log_every_n(2)
+            .unlimited()
+            .build()
+            .expect("failed to build PushQueues");
+        let _queues = queues;
+
+        handle
+            .try_push(0, priority, PushPolicy::ReturnErrorIfFull)
+            .expect("initial push should succeed");
+
+        let probe = handle.probe();
+        let h1 = handle.clone();
+        let h2 = handle.clone();
+
+        let t1_priority = priority;
+        let t2_priority = priority;
+
+        let t1 = thread::spawn(move || {
+            h1.try_push(1, t1_priority, PushPolicy::WarnAndDropIfFull)
+                .expect("first drop should succeed");
+        });
+        let t2 = thread::spawn(move || {
+            h2.try_push(2, t2_priority, PushPolicy::WarnAndDropIfFull)
+                .expect("second drop should succeed");
+        });
+
+        t1.join().expect("first drop thread panicked");
+        t2.join().expect("second drop thread panicked");
+
+        assert_eq!(
+            probe.dlq_drop_count(),
+            0,
+            "counter should reset after reaching the logging threshold"
+        );
+
+        let mut drops = Vec::new();
+        while let Ok(frame) = dlq_rx.try_recv() {
+            drops.push(frame);
+        }
+        drops.sort_unstable();
+        assert_eq!(drops, vec![1, 2]);
+    });
+}
 
 #[test]
-fn concurrent_push_delivery() {
+fn concurrent_queue_full_errors_are_reported() {
     model(|| {
-        let rt = Builder::new_current_thread()
-            .enable_all()
+        let (queues, handle) = PushQueues::<u8>::builder()
+            .high_capacity(1)
+            .low_capacity(1)
+            .unlimited()
             .build()
-            .expect("failed to build tokio runtime");
-
-        rt.block_on(async {
-            let (queues, handle) = PushQueues::<u8>::builder()
-                .high_capacity(1)
-                .low_capacity(1)
-                .unlimited()
-                .build()
-                .expect("failed to build PushQueues");
-            let token = CancellationToken::new();
-
-            let out = loom::sync::Arc::new(loom::sync::Mutex::new(Vec::new()));
-            let out_clone = out.clone();
-            let mut actor: ConnectionActor<_, ()> =
-                ConnectionActor::new(queues, handle.clone(), None, token.clone());
-
-            let actor_task = tokio::spawn(async move {
-                let mut buf = Vec::new();
-                actor
-                    .run(&mut buf)
-                    .await
-                    .expect("connection actor failed to run");
-                out_clone
-                    .lock()
-                    .expect("mutex poisoned")
-                    .extend(buf);
-            });
-
-            let h1 = handle.clone();
-            let t1 = tokio::spawn(async move {
-                h1
-                    .push_high_priority(1u8)
-                    .await
-                    .expect("failed to push high priority frame");
-            });
-
-            let h2 = handle.clone();
-            let t2 = tokio::spawn(async move {
-                h2
-                    .push_low_priority(2u8)
-                    .await
-                    .expect("failed to push low priority frame");
-            });
-
-            t1.await.expect("high priority task join failed");
-            t2.await.expect("low priority task join failed");
-            token.cancel();
-            actor_task.await.expect("actor task join failed");
-
-            let buf = out.lock().expect("mutex poisoned");
-            assert!(buf.contains(&1));
-            assert!(buf.contains(&2));
+            .expect("failed to build PushQueues");
+        let _queues = queues;
+
+        handle
+            .try_push(0, PushPriority::High, PushPolicy::ReturnErrorIfFull)
+            .expect("initial push should succeed");
+
+        let h1 = handle.clone();
+        let h2 = handle.clone();
+        let h3 = handle.clone();
+
+        let t1 = thread::spawn(move || {
+            let res = h1.try_push(1, PushPriority::High, PushPolicy::ReturnErrorIfFull);
+            assert!(
+                matches!(res, Err(PushError::QueueFull)),
+                "expected queue full error for first producer"
+            );
+        });
+        let t2 = thread::spawn(move || {
+            let res = h2.try_push(2, PushPriority::High, PushPolicy::ReturnErrorIfFull);
+            assert!(
+                matches!(res, Err(PushError::QueueFull)),
+                "expected queue full error for second producer"
+            );
+        });
+        let t3 = thread::spawn(move || {
+            let res = h3.try_push(3, PushPriority::High, PushPolicy::ReturnErrorIfFull);
+            assert!(
+                matches!(res, Err(PushError::QueueFull)),
+                "expected queue full error for third producer"
+            );
         });
+
+        t1.join().expect("first producer thread panicked");
+        t2.join().expect("second producer thread panicked");
+        t3.join().expect("third producer thread panicked");
+    });
+}
+
+#[test]
+fn dlq_probe_ignores_absent_channel() {
+    model(|| {
+        let (queues, handle) = PushQueues::<u8>::builder()
+            .high_capacity(1)
+            .low_capacity(1)
+            .dlq(None)
+            .dlq_log_every_n(2)
+            .unlimited()
+            .build()
+            .expect("failed to build PushQueues");
+        let _queues = queues;
+
+        handle
+            .try_push(0, PushPriority::High, PushPolicy::ReturnErrorIfFull)
+            .expect("initial push should succeed");
+
+        let probe = handle.probe();
+
+        handle
+            .try_push(1, PushPriority::High, PushPolicy::WarnAndDropIfFull)
+            .expect("drop should succeed even without a DLQ");
+
+        assert_eq!(
+            probe.dlq_drop_count(),
+            0,
+            "counter remains zero when DLQ is disabled"
+        );
     });
 }
 
+#[test]
+fn dlq_probe_reports_zero_when_dlq_idle() {
+    model(|| {
+        let (dlq_tx, mut dlq_rx) = mpsc::channel(2);
+        let (queues, handle) = PushQueues::<u8>::builder()
+            .high_capacity(1)
+            .low_capacity(1)
+            .dlq(Some(dlq_tx))
+            .dlq_log_every_n(2)
+            .unlimited()
+            .build()
+            .expect("failed to build PushQueues");
+        let _queues = queues;
+
+        let probe = handle.probe();
+
+        assert_eq!(
+            probe.dlq_drop_count(),
+            0,
+            "counter remains zero before any drops"
+        );
+        assert!(
+            dlq_rx.try_recv().is_err(),
+            "DLQ should be empty before any drops"
+        );
+    });
+}