From 0609317fcf06365c5c397e6fc28b7eccb842b248 Mon Sep 17 00:00:00 2001 From: Leynos Date: Sat, 9 Aug 2025 07:17:23 +0100 Subject: [PATCH 1/6] Add async accept-loop backoff test Exercise accept_loop with a mock listener that always errors to observe backoff doubling and capping in an async context. --- src/server/runtime.rs | 96 ++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 90 insertions(+), 6 deletions(-) diff --git a/src/server/runtime.rs b/src/server/runtime.rs index 95df19ae..94f2aa20 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -1,10 +1,11 @@ //! Runtime control for [`WireframeServer`]. -use std::sync::Arc; +use std::{io, net::SocketAddr, sync::Arc}; +use async_trait::async_trait; use futures::Future; use tokio::{ - net::TcpListener, + net::{TcpListener, TcpStream}, select, signal, time::{Duration, sleep}, @@ -21,6 +22,21 @@ use super::{ }; use crate::{app::WireframeApp, preamble::Preamble}; +#[async_trait] +pub(super) trait Listener { + async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)>; + fn local_addr(&self) -> io::Result; +} + +#[async_trait] +impl Listener for TcpListener { + async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> { + TcpListener::accept(self).await + } + + fn local_addr(&self) -> io::Result { TcpListener::local_addr(self) } +} + /// /// /// @@ -168,8 +184,8 @@ where } } -pub(super) async fn accept_loop( - listener: Arc, +pub(super) async fn accept_loop( + listener: Arc, factory: F, on_success: Option>, on_failure: Option, @@ -179,6 +195,7 @@ pub(super) async fn accept_loop( ) where F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, T: Preamble, + L: Listener + Send + Sync + 'static, { let mut delay = backoff_config.initial_delay; loop { @@ -213,13 +230,15 @@ pub(super) async fn accept_loop( mod tests { use std::sync::{ Arc, + Mutex, atomic::{AtomicUsize, Ordering}, }; + use async_trait::async_trait; use rstest::rstest; use tokio::{ sync::oneshot, - time::{Duration, timeout}, + time::{Duration, Instant, timeout}, }; use super::*; @@ -298,7 +317,7 @@ mod tests { .expect("failed to bind test listener"), ); - tracker.spawn(accept_loop::<_, ()>( + tracker.spawn(accept_loop::<_, (), _>( listener, factory, None, @@ -314,4 +333,69 @@ mod tests { let result = timeout(Duration::from_millis(100), tracker.wait()).await; assert!(result.is_ok()); } + + struct MockListener { + calls: Arc>>, + } + + impl MockListener { + fn new(calls: Arc>>) -> Self { Self { calls } } + } + + #[async_trait] + impl super::Listener for MockListener { + async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> { + self.calls.lock().expect("lock").push(Instant::now()); + Err(io::Error::other("mock error")) + } + + fn local_addr(&self) -> io::Result { + Ok("127.0.0.1:0".parse().expect("addr parse")) + } + } + + #[rstest] + #[tokio::test] + async fn test_accept_loop_exponential_backoff_async( + factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, + ) { + let calls = Arc::new(Mutex::new(Vec::new())); + let listener = Arc::new(MockListener::new(calls.clone())); + let token = CancellationToken::new(); + let tracker = TaskTracker::new(); + let backoff = BackoffConfig { + initial_delay: Duration::from_millis(5), + max_delay: Duration::from_millis(20), + }; + + tracker.spawn(accept_loop::<_, (), _>( + listener, + factory, + None, + None, + token.clone(), + tracker.clone(), + backoff, + )); + + while calls.lock().expect("lock").len() < 4 { + sleep(Duration::from_millis(1)).await; + } + + token.cancel(); + tracker.close(); + tracker.wait().await; + + let calls = calls.lock().expect("lock"); + let intervals: Vec<_> = calls.windows(2).map(|w| w[1] - w[0]).collect(); + let tolerance = Duration::from_millis(5); + let expected = [5, 10, 20]; + for (interval, ms) in intervals.iter().zip(expected) { + let target = Duration::from_millis(ms); + assert!( + *interval >= target && *interval < target + tolerance, + "interval {interval:?} not within expected range {target:?}" + ); + } + } } From 09b4d33972696c4abf97391032aad6342553ef7b Mon Sep 17 00:00:00 2001 From: Leynos Date: Tue, 12 Aug 2025 16:45:42 +0100 Subject: [PATCH 2/6] Rename listener trait and stabilise backoff test --- docs/mocking-network-outages-in-rust.md | 2 +- src/server/runtime.rs | 38 ++++++++++++++----------- 2 files changed, 23 insertions(+), 17 deletions(-) diff --git a/docs/mocking-network-outages-in-rust.md b/docs/mocking-network-outages-in-rust.md index c2f978a7..430b9d80 100644 --- a/docs/mocking-network-outages-in-rust.md +++ b/docs/mocking-network-outages-in-rust.md @@ -579,7 +579,7 @@ for mocking might be higher-level components: - **Accept Loop Simulation:** We could define a trait for the listener: ```rust - trait Listener { + trait AcceptListener { async fn accept(&self) -> io::Result<(Box, SocketAddr)>; } ``` diff --git a/src/server/runtime.rs b/src/server/runtime.rs index 94f2aa20..d3a9b92d 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -23,13 +23,13 @@ use super::{ use crate::{app::WireframeApp, preamble::Preamble}; #[async_trait] -pub(super) trait Listener { +pub(super) trait AcceptListener { async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)>; fn local_addr(&self) -> io::Result; } #[async_trait] -impl Listener for TcpListener { +impl AcceptListener for TcpListener { async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> { TcpListener::accept(self).await } @@ -195,7 +195,7 @@ pub(super) async fn accept_loop( ) where F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, T: Preamble, - L: Listener + Send + Sync + 'static, + L: AcceptListener + Send + Sync + 'static, { let mut delay = backoff_config.initial_delay; loop { @@ -238,7 +238,8 @@ mod tests { use rstest::rstest; use tokio::{ sync::oneshot, - time::{Duration, Instant, timeout}, + task::yield_now, + time::{Duration, Instant, advance, timeout}, }; use super::*; @@ -343,7 +344,7 @@ mod tests { } #[async_trait] - impl super::Listener for MockListener { + impl super::AcceptListener for MockListener { async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> { self.calls.lock().expect("lock").push(Instant::now()); Err(io::Error::other("mock error")) @@ -355,7 +356,7 @@ mod tests { } #[rstest] - #[tokio::test] + #[tokio::test(start_paused = true)] async fn test_accept_loop_exponential_backoff_async( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { @@ -378,24 +379,29 @@ mod tests { backoff, )); - while calls.lock().expect("lock").len() < 4 { - sleep(Duration::from_millis(1)).await; + yield_now().await; + + for ms in [5, 10, 20] { + advance(Duration::from_millis(ms)).await; + yield_now().await; } token.cancel(); + advance(Duration::from_millis(20)).await; + yield_now().await; tracker.close(); tracker.wait().await; let calls = calls.lock().expect("lock"); + assert_eq!(calls.len(), 4); let intervals: Vec<_> = calls.windows(2).map(|w| w[1] - w[0]).collect(); - let tolerance = Duration::from_millis(5); - let expected = [5, 10, 20]; - for (interval, ms) in intervals.iter().zip(expected) { - let target = Duration::from_millis(ms); - assert!( - *interval >= target && *interval < target + tolerance, - "interval {interval:?} not within expected range {target:?}" - ); + let expected = [ + Duration::from_millis(5), + Duration::from_millis(10), + Duration::from_millis(20), + ]; + for (interval, expected) in intervals.into_iter().zip(expected) { + assert_eq!(interval, expected); } } } From 37df8b771ae11ff88685a354ba64bf70e4e05103 Mon Sep 17 00:00:00 2001 From: Leynos Date: Tue, 12 Aug 2025 17:38:12 +0100 Subject: [PATCH 3/6] Document accept loop and assert immediate first accept --- src/server/runtime.rs | 58 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/src/server/runtime.rs b/src/server/runtime.rs index d3a9b92d..a3764105 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -184,6 +184,57 @@ where } } +/// Accepts incoming connections and spawns handler tasks. +/// +/// The loop accepts connections from `listener`, creates a new +/// [`WireframeApp`] via `factory` for each one, and spawns a task to handle +/// the connection. Failures to accept a connection trigger an exponential +/// backoff governed by `backoff_config`. The loop terminates when `shutdown` +/// is cancelled, and all spawned tasks are tracked by `tracker` for graceful +/// shutdown. +/// +/// # Parameters +/// +/// - `listener`: Source of incoming TCP connections. +/// - `factory`: Creates a fresh [`WireframeApp`] for each connection. +/// - `on_success`: Callback invoked after a successful preamble. +/// - `on_failure`: Callback invoked when the preamble fails. +/// - `shutdown`: Signal used to stop the accept loop. +/// - `tracker`: Task tracker used for graceful shutdown. +/// - `backoff_config`: Controls exponential backoff behaviour. +/// +/// # Type Parameters +/// +/// - `F`: Factory function that creates [`WireframeApp`] instances. +/// - `T`: Preamble type for connection handshaking. +/// - `L`: Listener type implementing [`AcceptListener`]. +/// +/// # Examples +/// +/// ``` +/// use std::sync::Arc; +/// +/// use tokio_util::{sync::CancellationToken, task::TaskTracker}; +/// use wireframe::{ +/// app::WireframeApp, +/// server::runtime::{AcceptListener, BackoffConfig, accept_loop}, +/// }; +/// +/// async fn run(listener: Arc) { +/// let tracker = TaskTracker::new(); +/// let token = CancellationToken::new(); +/// accept_loop::<_, (), _>( +/// listener, +/// || WireframeApp::default(), +/// None, +/// None, +/// token, +/// tracker, +/// BackoffConfig::default(), +/// ) +/// .await; +/// } +/// ``` pub(super) async fn accept_loop( listener: Arc, factory: F, @@ -381,6 +432,12 @@ mod tests { yield_now().await; + let first_call = { + let calls = calls.lock().expect("lock"); + assert_eq!(calls.len(), 1); + calls[0] + }; + for ms in [5, 10, 20] { advance(Duration::from_millis(ms)).await; yield_now().await; @@ -394,6 +451,7 @@ mod tests { let calls = calls.lock().expect("lock"); assert_eq!(calls.len(), 4); + assert_eq!(calls[0], first_call); let intervals: Vec<_> = calls.windows(2).map(|w| w[1] - w[0]).collect(); let expected = [ Duration::from_millis(5), From 11239a4b4faf2b5593ef361729214959ccf4e58a Mon Sep 17 00:00:00 2001 From: Leynos Date: Tue, 12 Aug 2025 21:05:02 +0100 Subject: [PATCH 4/6] Validate back-off config and tidy docs --- src/server/runtime.rs | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/src/server/runtime.rs b/src/server/runtime.rs index a3764105..12327c66 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -37,13 +37,10 @@ impl AcceptListener for TcpListener { fn local_addr(&self) -> io::Result { TcpListener::local_addr(self) } } +/// Configuration for exponential back-off timing in the accept loop. /// -/// -/// -/// Configuration for exponential backoff timing in the accept loop. -/// -/// Controls retry behavior when `accept()` calls fail on the server's TCP listener. -/// The backoff starts at `initial_delay` and doubles on each failure, capped at `max_delay`. +/// Controls retry behaviour when `accept()` calls fail on the server's TCP listener. +/// The back-off starts at `initial_delay` and doubles on each failure, capped at `max_delay`. /// /// # Default Values /// - `initial_delay`: 10 milliseconds @@ -189,7 +186,7 @@ where /// The loop accepts connections from `listener`, creates a new /// [`WireframeApp`] via `factory` for each one, and spawns a task to handle /// the connection. Failures to accept a connection trigger an exponential -/// backoff governed by `backoff_config`. The loop terminates when `shutdown` +/// back-off governed by `backoff_config`. The loop terminates when `shutdown` /// is cancelled, and all spawned tasks are tracked by `tracker` for graceful /// shutdown. /// @@ -201,7 +198,7 @@ where /// - `on_failure`: Callback invoked when the preamble fails. /// - `shutdown`: Signal used to stop the accept loop. /// - `tracker`: Task tracker used for graceful shutdown. -/// - `backoff_config`: Controls exponential backoff behaviour. +/// - `backoff_config`: Controls exponential back-off behaviour. /// /// # Type Parameters /// @@ -211,7 +208,7 @@ where /// /// # Examples /// -/// ``` +/// ```no_run /// use std::sync::Arc; /// /// use tokio_util::{sync::CancellationToken, task::TaskTracker}; @@ -248,6 +245,14 @@ pub(super) async fn accept_loop( T: Preamble, L: AcceptListener + Send + Sync + 'static, { + debug_assert!( + backoff_config.initial_delay >= Duration::from_millis(1), + "initial_delay must be at least 1ms", + ); + debug_assert!( + backoff_config.initial_delay <= backoff_config.max_delay, + "initial_delay must not exceed max_delay", + ); let mut delay = backoff_config.initial_delay; loop { select! { From 8bc816b7e8a5c68a13499b4ab3a4eb4deaff91a9 Mon Sep 17 00:00:00 2001 From: Leynos Date: Tue, 12 Aug 2025 21:48:24 +0100 Subject: [PATCH 5/6] Document AcceptListener and mock backoff test --- Cargo.lock | 71 +++++++++++++++++++++++++++++++++++++++++++ Cargo.toml | 1 + src/server/runtime.rs | 52 ++++++++++++++----------------- 3 files changed, 95 insertions(+), 29 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9d1f1dfc..37a7d4ef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -512,6 +512,12 @@ dependencies = [ "syn", ] +[[package]] +name = "downcast" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1435fa1053d8b2fbbe9be7e97eca7f33d37b28409959813daefc1446a14247f1" + [[package]] name = "drain_filter_polyfill" version = "0.1.3" @@ -576,6 +582,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "fragile" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28dd6caf6059519a65843af8fe2a3ae298b14b80179855aeb4adc2c1934ee619" + [[package]] name = "fs_extra" version = "1.3.0" @@ -1246,6 +1258,32 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "mockall" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39a6bfcc6c8c7eed5ee98b9c3e33adc726054389233e201c95dab2d41a3839d2" +dependencies = [ + "cfg-if", + "downcast", + "fragile", + "mockall_derive", + "predicates", + "predicates-tree", +] + +[[package]] +name = "mockall_derive" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ca3004c2efe9011bd4e461bd8256445052b9615405b4f7ea43fc8ca5c20898" +dependencies = [ + "cfg-if", + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "nibble_vec" version = "0.1.0" @@ -1434,6 +1472,32 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "predicates" +version = "3.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5d19ee57562043d37e82899fade9a22ebab7be9cef5026b07fda9cdd4293573" +dependencies = [ + "anstyle", + "predicates-core", +] + +[[package]] +name = "predicates-core" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "727e462b119fe9c93fd0eb1429a5f7647394014cf3c04ab2c0350eeb09095ffa" + +[[package]] +name = "predicates-tree" +version = "1.0.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72dd2d6d381dfb73a193c7fca536518d7caee39fc8503f74e7dc0be0531b425c" +dependencies = [ + "predicates-core", + "termtree", +] + [[package]] name = "prettyplease" version = "0.2.35" @@ -2092,6 +2156,12 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "termtree" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683" + [[package]] name = "textwrap" version = "0.16.2" @@ -2825,6 +2895,7 @@ dependencies = [ "metrics", "metrics-exporter-prometheus", "metrics-util", + "mockall", "proptest", "rstest", "serde", diff --git a/Cargo.toml b/Cargo.toml index 7c34913e..e4d6686f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,6 +41,7 @@ serial_test = "3.2.0" cucumber = "0.20.2" metrics-util = "0.20.0" tracing-test = "0.2.5" +mockall = "0.13.1" [features] default = ["metrics"] diff --git a/src/server/runtime.rs b/src/server/runtime.rs index 12327c66..24a794c5 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -22,8 +22,13 @@ use super::{ }; use crate::{app::WireframeApp, preamble::Preamble}; +/// Abstraction for sources of incoming connections consumed by the accept loop. +/// +/// Implementations must be cancellation-safe: dropping a pending `accept()` +/// future must not leak resources. +#[cfg_attr(test, mockall::automock)] #[async_trait] -pub(super) trait AcceptListener { +pub(super) trait AcceptListener: Send + Sync { async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)>; fn local_addr(&self) -> io::Result; } @@ -208,14 +213,11 @@ where /// /// # Examples /// -/// ```no_run +/// ```ignore /// use std::sync::Arc; /// /// use tokio_util::{sync::CancellationToken, task::TaskTracker}; -/// use wireframe::{ -/// app::WireframeApp, -/// server::runtime::{AcceptListener, BackoffConfig, accept_loop}, -/// }; +/// use wireframe::{app::WireframeApp /*, server::runtime::{AcceptListener, BackoffConfig, accept_loop} */}; /// /// async fn run(listener: Arc) { /// let tracker = TaskTracker::new(); @@ -290,7 +292,6 @@ mod tests { atomic::{AtomicUsize, Ordering}, }; - use async_trait::async_trait; use rstest::rstest; use tokio::{ sync::oneshot, @@ -298,7 +299,7 @@ mod tests { time::{Duration, Instant, advance, timeout}, }; - use super::*; + use super::{MockAcceptListener, *}; use crate::server::test_util::{bind_server, factory, free_listener}; #[rstest] @@ -391,33 +392,26 @@ mod tests { assert!(result.is_ok()); } - struct MockListener { - calls: Arc>>, - } - - impl MockListener { - fn new(calls: Arc>>) -> Self { Self { calls } } - } - - #[async_trait] - impl super::AcceptListener for MockListener { - async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)> { - self.calls.lock().expect("lock").push(Instant::now()); - Err(io::Error::other("mock error")) - } - - fn local_addr(&self) -> io::Result { - Ok("127.0.0.1:0".parse().expect("addr parse")) - } - } - #[rstest] #[tokio::test(start_paused = true)] async fn test_accept_loop_exponential_backoff_async( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { let calls = Arc::new(Mutex::new(Vec::new())); - let listener = Arc::new(MockListener::new(calls.clone())); + let mut listener = MockAcceptListener::new(); + let call_log = calls.clone(); + listener + .expect_accept() + .returning(move || { + call_log.lock().expect("lock").push(Instant::now()); + Err(io::Error::other("mock error")) + }) + .times(4); + listener + .expect_local_addr() + .returning(|| Ok("127.0.0.1:0".parse().expect("addr parse"))) + .times(4); + let listener = Arc::new(listener); let token = CancellationToken::new(); let tracker = TaskTracker::new(); let backoff = BackoffConfig { From 2ad259708b446657f2faa9bbbb572b026734d102 Mon Sep 17 00:00:00 2001 From: Leynos Date: Tue, 12 Aug 2025 23:41:36 +0100 Subject: [PATCH 6/6] Return pinned future from accept mock --- src/server/runtime.rs | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/src/server/runtime.rs b/src/server/runtime.rs index 24a794c5..401bb3c3 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -26,8 +26,8 @@ use crate::{app::WireframeApp, preamble::Preamble}; /// /// Implementations must be cancellation-safe: dropping a pending `accept()` /// future must not leak resources. -#[cfg_attr(test, mockall::automock)] #[async_trait] +#[cfg_attr(test, mockall::automock)] pub(super) trait AcceptListener: Send + Sync { async fn accept(&self) -> io::Result<(TcpStream, SocketAddr)>; fn local_addr(&self) -> io::Result; @@ -403,8 +403,11 @@ mod tests { listener .expect_accept() .returning(move || { - call_log.lock().expect("lock").push(Instant::now()); - Err(io::Error::other("mock error")) + let call_log = Arc::clone(&call_log); + Box::pin(async move { + call_log.lock().expect("lock").push(Instant::now()); + Err(io::Error::other("mock error")) + }) }) .times(4); listener