Add slow reader/writer back-pressure testkit, tests, and docs

docs/adr-002-streaming-requests-and-shared-message-assembly.md

@@ -472,6 +472,13 @@ These utilities should build on the existing `wireframe_testing` companion
 crate, and may be re-exported as `wireframe::testkit` behind a dedicated
 feature to keep the core crate lightweight.[^testing]
 
+Implementation note for roadmap item `8.5.2`: `wireframe_testing` now exposes
+`SlowIoPacing` and `SlowIoConfig`, together with slow-I/O driver helpers for
+raw frames, default length-delimited payloads, and codec-aware payload/frame
+round trips. The pacing model is additive and duplex-based: tests can slow the
+client write direction, the client read direction, or both, while keeping the
+rest of the in-process app harness unchanged.
+
 ## Consequences
 
 - Wireframe gains an explicit “streaming request body” surface alongside the

docs/roadmap.md

@@ -315,7 +315,7 @@ and standardized per-connection memory budgets.
 
 - [x] 8.5.1. Add utilities for feeding partial frames or fragments into an
   in-process app.
-- [ ] 8.5.2. Add slow reader and writer simulation for back-pressure testing.
+- [x] 8.5.2. Add slow reader and writer simulation for back-pressure testing.
 - [ ] 8.5.3. Add deterministic assertion helpers for reassembly outcomes.
 - [ ] 8.5.4. Export utilities as `wireframe::testkit` behind a dedicated
   feature.

docs/users-guide.md

@@ -418,6 +418,62 @@ Available assertion helpers:
 - `assert_log_at_level(level, substring)` — assert a log at a specific level
   contains a substring.
 
+#### Simulating slow readers and writers
+
+Back-pressure tests often need more than partial-frame delivery. The
+`wireframe_testing` crate also provides slow-I/O helpers that pace the client
+write side, the client read side, or both directions at once.
+
+Use `SlowIoPacing` to define a chunk size and inter-chunk delay, then apply it
+through `SlowIoConfig`:
+
+```rust,no_run
+use std::{num::NonZeroUsize, time::Duration};
+use wireframe::{
+    app::{Envelope, WireframeApp},
+    codec::examples::HotlineFrameCodec,
+    serializer::{BincodeSerializer, Serializer},
+};
+use wireframe_testing::{
+    SlowIoConfig, SlowIoPacing, drive_with_slow_codec_payloads,
+};
+
+let codec = HotlineFrameCodec::new(4096);
+let app = WireframeApp::new()?.with_codec(codec.clone());
+let config = SlowIoConfig::new()
+    .with_writer_pacing(SlowIoPacing::new(
+        NonZeroUsize::new(8).expect("non-zero"),
+        Duration::from_millis(5),
+    ))
+    .with_reader_pacing(SlowIoPacing::new(
+        NonZeroUsize::new(32).expect("non-zero"),
+        Duration::from_millis(5),
+    ))
+    .with_capacity(64);
+
+let request = BincodeSerializer.serialize(&Envelope::new(
+    1,
+    Some(7),
+    vec![1, 2, 3],
+))?;
+let payloads = drive_with_slow_codec_payloads(app, &codec, vec![request], config)
+    .await?;
+```
+
+Available slow-I/O helper functions:
+
+- `drive_with_slow_frames` — pre-framed bytes, returns raw output bytes.
+- `drive_with_slow_payloads` — default length-delimited payloads, returns raw
+  output bytes.
+- `drive_with_slow_codec_payloads` — codec-aware payloads, returns decoded
+  payload byte vectors.
+- `drive_with_slow_codec_frames` — codec-aware frames, returns decoded
+  `F::Frame` values.
+
+These helpers are designed for deterministic tests under paused Tokio time. Use
+small duplex capacities together with reader pacing when you need the app's
+outbound writes to hit back-pressure quickly.
+
 #### Zero-copy payload extraction
 
 For performance-critical codecs, use `Bytes` instead of `Vec<u8>` for payload

docs/wireframe-testing-crate.md

@@ -141,6 +141,103 @@ Behavioural details:
 - I/O failures, framing errors, and server task panics are all returned as
   `io::Error` values, so tests can assert on error handling.
 
+### Slow-I/O drivers
+
+Roadmap item `8.5.2` extends the in-memory harness with explicit slow reader
+and slow writer simulation. The public surface uses one pacing type plus one
+driver config:
+
+```rust,no_run
+# fn example() -> Result<(), Box<dyn std::error::Error>> {
+use std::{num::NonZeroUsize, time::Duration};
+use wireframe_testing::{SlowIoConfig, SlowIoPacing};
+
+let writer = SlowIoPacing::new(
+    NonZeroUsize::new(8).ok_or_else(|| std::io::Error::other("chunk size must be non-zero"))?,
+    Duration::from_millis(5),
+);
+let reader = SlowIoPacing::new(
+    NonZeroUsize::new(32).ok_or_else(|| std::io::Error::other("chunk size must be non-zero"))?,
+    Duration::from_millis(5),
+);
+let config = SlowIoConfig::new()
+    .with_writer_pacing(writer)
+    .with_reader_pacing(reader)
+    .with_capacity(64);
+# let _ = config;
+# Ok(())
+# }
+```
+
+The pacing applies to the client side of the in-memory duplex stream:
+
+- `writer_pacing` throttles bytes written into the app.
+- `reader_pacing` throttles bytes drained from the app.
+- `capacity` controls how quickly the duplex buffer saturates, which is useful
+  when asserting back-pressure behaviour.
+
+Accessibility caption: Sequence diagram showing a test spawning an async
+runtime task that drives slow-I/O helpers, optionally pacing writes into the
+app and reads back out of it through Tokio time delays before returning the
+captured bytes for back-pressure assertions.
+
+```mermaid
+sequenceDiagram
+    actor Test as Test
+    participant Runtime as TokioRuntime
+    participant Helper as SlowIoHelpers
+    participant App as WireframeApp
+    participant Writer as SlowWriterPacer
+    participant Reader as SlowReaderPacer
+    participant Time as TokioTime
+
+    Test->>Runtime: spawn async test
+    Runtime->>Helper: drive_with_slow_payloads(app, payloads, config)
+
+    alt writer_pacing configured
+        Helper->>Writer: start_paced_writes(payloads, config.writer_pacing)
+        Writer->>App: send_first_chunk()
+        loop for each remaining chunk
+            Writer->>Time: sleep(config.writer_pacing.delay)
+            Time-->>Writer: wake
+            Writer->>App: send_chunk()
+        end
+    else no writer pacing
+        Helper->>App: send_all_payloads()
+    end
+
+    alt reader_pacing configured
+        Helper->>Reader: start_paced_reads(config.reader_pacing)
+        Reader->>App: read_first_chunk()
+        App-->>Reader: first_chunk_bytes
+        Reader-->>Helper: append_to_output(first_chunk_bytes)
+        loop while app_has_more_output
+            Reader->>Time: sleep(config.reader_pacing.delay)
+            Time-->>Reader: wake
+            Reader->>App: read_chunk()
+            App-->>Reader: chunk_bytes
+            Reader-->>Helper: append_to_output(chunk_bytes)
+        end
+    else no reader pacing
+        Helper->>App: read_all_output()
+        App-->>Helper: all_bytes
+    end
+
+    Helper-->>Runtime: Result<Vec<u8>>
+    Runtime-->>Test: assert_backpressure_behaviour()
+```
+
+Public entry points:
+
+- `drive_with_slow_frames` for pre-framed raw bytes.
+- `drive_with_slow_payloads` for default length-delimited payloads.
+- `drive_with_slow_codec_payloads` for codec-aware payload round trips.
+- `drive_with_slow_codec_frames` for codec-aware frame assertions.
+
+These helpers are intentionally additive rather than replacing the existing
+drivers. Existing tests keep the simpler fast-path helpers, while
+back-pressure-focused tests opt into explicit pacing.
+
 ### Buffer capacity and limits
 
 The duplex stream buffer defaults to `TEST_MAX_FRAME`, matching the shared

src/panic.rs

@@ -5,14 +5,25 @@
 
 use std::any::Any;
 
-/// Format a panic payload into a `String` using `Debug` formatting.
+/// Format a panic payload into a `String`.
+///
+/// String and string-slice panic payloads preserve their original message.
+/// Other payload types fall back to `Debug` formatting.
 ///
 /// # Examples
 /// ```
 /// # use std::any::Any;
 /// # use wireframe::panic::format_panic;
 /// let payload: Box<dyn Any + Send> = Box::new("boom");
-/// assert_eq!(format_panic(&payload), "Any { .. }");
+/// assert_eq!(format_panic(&payload), "boom");
 /// ```
 #[must_use]
-pub fn format_panic(panic: &Box<dyn Any + Send>) -> String { format!("{panic:?}") }
+pub fn format_panic(panic: &Box<dyn Any + Send>) -> String {
+    if let Some(message) = panic.downcast_ref::<&'static str>() {
+        (*message).to_owned()
+    } else if let Some(message) = panic.downcast_ref::<String>() {
+        message.clone()
+    } else {
+        format!("{panic:?}")
+    }
+}

src/server/connection_spawner.rs

@@ -229,7 +229,7 @@ mod tests {
                 .iter()
                 .find(|line| {
                     line.contains("connection task panicked")
-                        && line.contains("panic=Any")
+                        && line.contains("panic=boom")
                         && line.contains(&format!("peer_addr=Some({peer_addr})"))
                 })
                 .map(|_| ())
@@ -345,7 +345,7 @@ mod tests {
                 .iter()
                 .find(|line| {
                     line.contains("connection task panicked")
-                        && line.contains("panic=Any")
+                        && line.contains("panic=boom")
                         && line.contains(&format!("peer_addr=Some({peer_addr})"))
                 })
                 .map(|_| ())

tests/features/slow_io_backpressure.feature

@@ -0,0 +1,25 @@
+@slow_io_backpressure
+Feature: Slow reader and writer simulation
+  Slow-I/O helpers pace the client write and read sides so back-pressure can be
+  asserted deterministically under paused Tokio time.
+
+  Scenario: Slow writer delays request completion
+    Given a slow-io echo app with max frame length 4096
+    When a 64-byte request is driven with slow writer pacing of 8 bytes every 5 milliseconds
+    Then the slow-io drive remains pending
+    When slow-io virtual time advances by 100 milliseconds
+    Then the slow-io drive completes with an echoed payload of 64 bytes
+
+  Scenario: Slow reader delays response draining
+    Given a slow-io echo app with max frame length 4096
+    When a slow reader drive is configured as 256/16/5/64
+    Then the slow-io drive remains pending
+    When slow-io virtual time advances by 200 milliseconds
+    Then the slow-io drive completes with an echoed payload of 256 bytes
+
+  Scenario: Combined slow reader and writer still round-trips correctly
+    Given a slow-io echo app with max frame length 4096
+    When a combined slow-io drive is configured as 96/12/5/24/5/64
+    Then the slow-io drive remains pending
+    When slow-io virtual time advances by 200 milliseconds
+    Then the slow-io drive completes with an echoed payload of 96 bytes

tests/fixtures/mod.rs

@@ -37,6 +37,7 @@ pub mod panic;
 pub mod partial_frame_feeding;
 pub mod request_parts;
 pub mod serializer_boundaries;
+pub mod slow_io_backpressure;
 pub mod stream_end;
 pub mod test_observability;
 pub mod unified_codec;