Type preamble handler test cases and stabilise runtime readiness

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.

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),
             },

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

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,
         }

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,24 +349,33 @@ 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));
 }
 
 #[rstest]
 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)

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<oneshot::Sender<()>>,
+    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,

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<F, T> WireframeServer<F, T, Bound>
 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<F, T, L>(
     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 {