diff --git a/docs/server/configuration.md b/docs/server/configuration.md index f40edf2b..44df5fab 100644 --- a/docs/server/configuration.md +++ b/docs/server/configuration.md @@ -20,8 +20,9 @@ server.run().await?; ## Accept loop backoff The accept loop retries failed `accept()` calls using exponential backoff. -`accept_backoff(initial_delay, max_delay)` sets both bounds in one call. These -values are stored in `BackoffConfig`: +`accept_backoff(cfg)` sets both bounds using a `BackoffConfig` value. The +builder normalises the supplied configuration via `BackoffConfig::normalised`, +so out-of-range values are adjusted rather than preserved: - `initial_delay` – starting delay for the first retry, clamped to at least 1 millisecond. @@ -37,11 +38,13 @@ values are stored in `BackoffConfig`: ```rust use std::time::Duration; -use wireframe::{app::WireframeApp, server::WireframeServer}; +use wireframe::{app::WireframeApp, server::{WireframeServer, BackoffConfig}}; + +let cfg = BackoffConfig { + initial_delay: Duration::from_millis(5), + max_delay: Duration::from_millis(500), +}; let server = WireframeServer::new(|| WireframeApp::default()) - .accept_backoff(Duration::from_millis(5), Duration::from_millis(500)); + .accept_backoff(cfg); ``` - -`accept_initial_delay` and `accept_max_delay` allow adjusting each parameter -individually. diff --git a/src/server/config/binding.rs b/src/server/config/binding.rs index b136c74e..75ed6beb 100644 --- a/src/server/config/binding.rs +++ b/src/server/config/binding.rs @@ -88,6 +88,7 @@ where on_preamble_success: self.on_preamble_success, on_preamble_failure: self.on_preamble_failure, ready_tx: self.ready_tx, + backoff_config: self.backoff_config, state: Bound { listener: Arc::new(tokio), }, @@ -173,6 +174,7 @@ where on_preamble_success: self.on_preamble_success, on_preamble_failure: self.on_preamble_failure, ready_tx: self.ready_tx, + backoff_config: self.backoff_config, state: Bound { listener: Arc::new(tokio), }, diff --git a/src/server/config/mod.rs b/src/server/config/mod.rs index 01e844f5..f882e4c0 100644 --- a/src/server/config/mod.rs +++ b/src/server/config/mod.rs @@ -16,7 +16,7 @@ use core::marker::PhantomData; use tokio::sync::oneshot; -use super::{ServerState, Unbound, WireframeServer}; +use super::{BackoffConfig, ServerState, Unbound, WireframeServer}; use crate::{app::WireframeApp, preamble::Preamble}; macro_rules! builder_setter { @@ -76,6 +76,7 @@ where on_preamble_success: None, on_preamble_failure: None, ready_tx: None, + backoff_config: BackoffConfig::default(), state: Unbound, _preamble: PhantomData, } @@ -122,6 +123,23 @@ where ready_signal, ready_tx, tx: oneshot::Sender<()> => Some(tx) ); + builder_setter!( + /// Configure accept-loop backoff behaviour. + /// + /// The supplied configuration is passed to + /// [`BackoffConfig::normalised`] (`cfg.normalised()`) before being + /// stored. Normalisation clamps `initial_delay` to at least 1 ms and no + /// greater than `max_delay`. If `initial_delay` exceeds `max_delay`, + /// the values are swapped. Normalisation applies any other adjustments + /// `BackoffConfig::normalised` defines so out-of-range values are + /// corrected rather than preserved. + /// + /// Invariants: + /// - `initial_delay` must be >= 1 ms + /// - `initial_delay` must be <= `max_delay` + accept_backoff, backoff_config, cfg: BackoffConfig => cfg.normalised() + ); + /// Returns the configured number of worker tasks for the server. /// /// # Examples diff --git a/src/server/config/preamble.rs b/src/server/config/preamble.rs index 449961f2..9ad3fadf 100644 --- a/src/server/config/preamble.rs +++ b/src/server/config/preamble.rs @@ -46,6 +46,7 @@ where on_preamble_success: None, on_preamble_failure: None, ready_tx: self.ready_tx, + backoff_config: self.backoff_config, state: self.state, _preamble: PhantomData, } diff --git a/src/server/config/tests.rs b/src/server/config/tests.rs index 485b7ac4..9dbeb49e 100644 --- a/src/server/config/tests.rs +++ b/src/server/config/tests.rs @@ -16,14 +16,24 @@ use std::{ use rstest::rstest; use super::*; -use crate::server::test_util::{ - TestPreamble, - bind_server, - factory, - free_listener, - listener_addr, - server_with_preamble, -}; +use bincode::error::DecodeError; +use crate::server::{ + test_util::{ + TestPreamble, + bind_server, + factory, + free_listener, + listener_addr, + server_with_preamble, + }, + BackoffConfig, +use tokio::net::{TcpListener, TcpStream}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum PreambleHandlerKind { + Success, + Failure, +} fn expected_default_worker_count() -> usize { // Mirror the default worker logic to keep tests aligned with `WireframeServer::new`. @@ -96,37 +106,61 @@ async fn test_local_addr_after_bind( } #[rstest] -#[case("success")] -#[case("failure")] +#[case::success(PreambleHandlerKind::Success)] +#[case::failure(PreambleHandlerKind::Failure)] #[tokio::test] async fn test_preamble_handler_registration( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, - #[case] handler_type: &str, + #[case] handler: PreambleHandlerKind, ) { let counter = Arc::new(AtomicUsize::new(0)); let c = counter.clone(); let server = server_with_preamble(factory); - let server = match handler_type { - "success" => server.on_preamble_decode_success(move |_p: &TestPreamble, _| { + let server = match handler { + PreambleHandlerKind::Success => server.on_preamble_decode_success(move |_p: &TestPreamble, _| { let c = c.clone(); Box::pin(async move { c.fetch_add(1, Ordering::SeqCst); Ok(()) }) }), - "failure" => server.on_preamble_decode_failure(move |_err: &DecodeError| { + PreambleHandlerKind::Failure => server.on_preamble_decode_failure(move |_err: &DecodeError| { c.fetch_add(1, Ordering::SeqCst); }), - _ => panic!("Invalid handler type"), }; assert_eq!(counter.load(Ordering::SeqCst), 0); - match handler_type { - "success" => assert!(server.on_preamble_success.is_some()), - "failure" => assert!(server.on_preamble_failure.is_some()), - _ => unreachable!(), + match handler { + PreambleHandlerKind::Success => { + assert!(server.on_preamble_success.is_some()); + let handler = server + .on_preamble_success + .as_ref() + .expect("success handler missing"); + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind listener"); + let addr = listener.local_addr().expect("listener addr"); + let _client = TcpStream::connect(addr) + .await + .expect("client connect failed"); + let (mut stream, _) = listener.accept().await.expect("accept stream"); + let preamble = TestPreamble { id: 0, message: String::new() }; + handler(&preamble, &mut stream) + .await + .expect("handler failed"); + } + PreambleHandlerKind::Failure => { + assert!(server.on_preamble_failure.is_some()); + let handler = server + .on_preamble_failure + .as_ref() + .expect("failure handler missing"); + handler(&DecodeError::UnexpectedEnd); + } } + assert_eq!(counter.load(Ordering::SeqCst), 1); } #[rstest] @@ -204,11 +238,12 @@ async fn test_bind_to_multiple_addresses( fn test_accept_backoff_configuration( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { - let initial = Duration::from_millis(5); - let max = Duration::from_millis(500); - let server = WireframeServer::new(factory).accept_backoff(initial, max); - assert_eq!(server.backoff_config.initial_delay, initial); - assert_eq!(server.backoff_config.max_delay, max); + let cfg = BackoffConfig { + initial_delay: Duration::from_millis(5), + max_delay: Duration::from_millis(500), + }; + let server = WireframeServer::new(factory).accept_backoff(cfg); + assert_eq!(server.backoff_config, cfg); } /// Behaviour test verifying exponential delay doubling and capping. @@ -264,7 +299,8 @@ fn test_accept_initial_delay_configuration( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { let delay = Duration::from_millis(20); - let server = WireframeServer::new(factory).accept_initial_delay(delay); + let cfg = BackoffConfig { initial_delay: delay, ..BackoffConfig::default() }; + let server = WireframeServer::new(factory).accept_backoff(cfg); assert_eq!(server.backoff_config.initial_delay, delay); } @@ -273,21 +309,29 @@ fn test_accept_max_delay_configuration( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { let delay = Duration::from_millis(2000); - let server = WireframeServer::new(factory).accept_max_delay(delay); + let cfg = BackoffConfig { max_delay: delay, ..BackoffConfig::default() }; + let server = WireframeServer::new(factory).accept_backoff(cfg); assert_eq!(server.backoff_config.max_delay, delay); } #[rstest] fn test_backoff_validation(factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static) { - let server = WireframeServer::new(factory.clone()).accept_initial_delay(Duration::ZERO); + let server = WireframeServer::new(factory.clone()) + .accept_backoff(BackoffConfig { initial_delay: Duration::ZERO, ..BackoffConfig::default() }); assert_eq!( server.backoff_config.initial_delay, Duration::from_millis(1) ); let server = WireframeServer::new(factory) - .accept_initial_delay(Duration::from_millis(100)) - .accept_max_delay(Duration::from_millis(50)); + .accept_backoff(BackoffConfig { + initial_delay: Duration::from_millis(100), + max_delay: Duration::from_millis(50), + }); + assert_eq!( + server.backoff_config.initial_delay, + Duration::from_millis(50) + ); assert_eq!(server.backoff_config.max_delay, Duration::from_millis(100)); } @@ -305,8 +349,12 @@ fn test_backoff_default_values(factory: impl Fn() -> WireframeApp + Send + Sync fn test_initial_delay_exceeds_default_max( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { - let server = WireframeServer::new(factory).accept_initial_delay(Duration::from_secs(2)); - assert_eq!(server.backoff_config.initial_delay, Duration::from_secs(2)); + let cfg = BackoffConfig { + initial_delay: Duration::from_secs(2), + max_delay: Duration::from_secs(1), + }; + let server = WireframeServer::new(factory).accept_backoff(cfg); + assert_eq!(server.backoff_config.initial_delay, Duration::from_secs(1)); assert_eq!(server.backoff_config.max_delay, Duration::from_secs(2)); } @@ -314,15 +362,20 @@ fn test_initial_delay_exceeds_default_max( fn test_accept_backoff_parameter_swapping( factory: impl Fn() -> WireframeApp + Send + Sync + Clone + 'static, ) { - let server = WireframeServer::new(factory.clone()) - .accept_backoff(Duration::from_millis(5), Duration::from_millis(1)); + let server = WireframeServer::new(factory.clone()).accept_backoff(BackoffConfig { + initial_delay: Duration::from_millis(5), + max_delay: Duration::from_millis(1), + }); assert_eq!( server.backoff_config.initial_delay, Duration::from_millis(1) ); assert_eq!(server.backoff_config.max_delay, Duration::from_millis(5)); - let server = WireframeServer::new(factory).accept_backoff(Duration::ZERO, Duration::ZERO); + let server = WireframeServer::new(factory).accept_backoff(BackoffConfig { + initial_delay: Duration::ZERO, + max_delay: Duration::ZERO, + }); assert_eq!( server.backoff_config.initial_delay, Duration::from_millis(1) diff --git a/src/server/mod.rs b/src/server/mod.rs index c73473ee..7ba0b448 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -119,6 +119,7 @@ where /// Because only one notification may be sent, a new `ready_tx` must be /// provided each time the server is started. pub(crate) ready_tx: Option>, + pub(crate) backoff_config: BackoffConfig, /// Typestate tracking whether the server has been bound to a listener. /// [`Unbound`] servers require binding before they can run. pub(crate) state: S, diff --git a/src/server/runtime.rs b/src/server/runtime.rs index f8f608b6..d98c4835 100644 --- a/src/server/runtime.rs +++ b/src/server/runtime.rs @@ -54,7 +54,7 @@ impl AcceptListener for TcpListener { /// # Invariants /// - `initial_delay` must not exceed `max_delay` /// - `initial_delay` must be at least 1 millisecond -#[derive(Clone, Copy, Debug)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct BackoffConfig { pub initial_delay: Duration, pub max_delay: Duration, @@ -69,6 +69,18 @@ impl Default for BackoffConfig { } } +impl BackoffConfig { + #[must_use] + pub fn normalised(mut self) -> Self { + self.initial_delay = self.initial_delay.max(Duration::from_millis(1)); + self.max_delay = self.max_delay.max(Duration::from_millis(1)); + if self.initial_delay > self.max_delay { + std::mem::swap(&mut self.initial_delay, &mut self.max_delay); + } + self + } +} + impl WireframeServer where F: Fn() -> WireframeApp + Send + Sync + Clone + 'static, @@ -181,15 +193,9 @@ where on_preamble_failure, ready_tx, state: Bound { listener }, + backoff_config, .. } = self; - - if let Some(tx) = ready_tx - && tx.send(()).is_err() - { - tracing::warn!("Failed to send readiness signal: receiver dropped"); - } - let shutdown_token = CancellationToken::new(); let tracker = TaskTracker::new(); @@ -207,10 +213,15 @@ where on_failure, token, t, - BackoffConfig::default(), + backoff_config, )); } + // Signal readiness after all workers have been spawned. + if ready_tx.is_some_and(|tx| tx.send(()).is_err()) { + tracing::warn!("Failed to send readiness signal: receiver dropped"); + } + select! { () = shutdown => shutdown_token.cancel(), () = tracker.wait() => {}, @@ -284,12 +295,12 @@ pub(super) async fn accept_loop( L: AcceptListener + Send + Sync + 'static, { debug_assert!( - backoff_config.initial_delay >= Duration::from_millis(1), - "initial_delay must be at least 1ms", + backoff_config.initial_delay <= backoff_config.max_delay, + "BackoffConfig invariant violated: initial_delay > max_delay" ); debug_assert!( - backoff_config.initial_delay <= backoff_config.max_delay, - "initial_delay must not exceed max_delay", + backoff_config.initial_delay >= Duration::from_millis(1), + "BackoffConfig invariant violated: initial_delay < 1ms" ); let mut delay = backoff_config.initial_delay; loop {