diff --git a/.github/workflows/coverage-main.yml b/.github/workflows/coverage-main.yml index ce326423..ceff4fc2 100644 --- a/.github/workflows/coverage-main.yml +++ b/.github/workflows/coverage-main.yml @@ -22,7 +22,12 @@ jobs: CARGO_TERM_COLOR: always BUILD_PROFILE: debug steps: - - uses: actions/checkout@v7 + # CodeScene derives github.com/leynos/wireframe from this checkout's + # origin. + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: + repository: leynos/wireframe + persist-credentials: false - name: Setup Rust uses: leynos/shared-actions/.github/actions/setup-rust@f4764bea8d813b1a8f7ebc37a44907d3c3b1e0e4 - name: Test and Measure Coverage @@ -38,5 +43,8 @@ jobs: uses: leynos/shared-actions/.github/actions/upload-codescene-coverage@f4764bea8d813b1a8f7ebc37a44907d3c3b1e0e4 with: format: lcov + mode: upload + # Keep this project identity aligned with ci.yml's PR gate. + project-url: https://api.codescene.io/v2/projects/68308 access-token: ${{ env.CS_ACCESS_TOKEN }} installer-checksum: ${{ vars.CODESCENE_CLI_SHA256 }} diff --git a/docs/contents.md b/docs/contents.md index 9a4ddd73..c6bf6d4a 100644 --- a/docs/contents.md +++ b/docs/contents.md @@ -96,6 +96,11 @@ the-road-to-wireframe-1-0-feature-set-philosophy-and-capability-maturity.md - [Testing helpers](wireframe-testing-crate.md) In-process server and client pair helpers provided by the `wireframe_testing` companion crate. +## Migration guides + +- [v0.3.0 to v0.4.0 migration guide](v0-3-0-to-v0-4-0-migration-guide.md) + Moving from builder connection handling to reusable prepared applications. + ## Operations and resilience - [Resilience guide](hardening-wireframe-a-guide-to-production-resilience.md) diff --git a/docs/developers-guide.md b/docs/developers-guide.md index ffdbb520..5b75b06f 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -67,8 +67,22 @@ boundaries for Epic 635: client-pool scheduler a single persistent owner task and index-based slot leases beneath one `PoolCore` root. -These records are proposed, not yet accepted; the review checklist derived from -ADR 011's rules lands with their implementation epic. +The first implementation slice is now in place: consuming +`WireframeApp::prepare().await` returns an immutable `PreparedApp` or a typed +`PrepareError`. Preparation consumes route and middleware registrations and +builds each route chain once. Connection tasks borrow the prepared route table, +so a single prepared application can serve multiple connections without +repeating middleware transforms. `WireframeApp` remains the registration +builder, and its direct connection methods are compatibility APIs. + +Server factory evaluation and readiness semantics remain unchanged in this +slice. The server-runtime work tracked by issue +[#642](https://github.com/leynos/wireframe/issues/642) will prepare the factory +result before server readiness; connection-local state and the +`ConnectionRuntime` follow in issue +[#643](https://github.com/leynos/wireframe/issues/643). The records remain +proposed, and the review checklist derived from ADR 011's rules lands with +their implementation epic. ### Server supervisor lifecycle @@ -234,6 +248,23 @@ Install Whitaker through the standalone installer described in the [Whitaker user's guide](whitaker-users-guide.md) so local linting matches continuous integration (CI). +### CodeScene coverage baseline + +The `Coverage (main)` workflow in `.github/workflows/coverage-main.yml` runs on +pushes to `main`. After the test suite succeeds, it generates a ratcheted LCOV +report and uploads that report to CodeScene. The workflow checks out +`leynos/wireframe`, so CodeScene records the coverage under the repository +identity `github.com/leynos/wireframe`, and targets project `68308` explicitly. + +The upload reads `CS_ACCESS_TOKEN` from the repository secret into the job +environment, then passes that value through the upload action's required +`access-token` input. This workflow input is permitted because the value still +comes from the repository secret; never hard-code the token in workflow or +source files, and never log it. The pull-request workflow's CodeScene coverage +check uses the same project and repository identity. It consumes the report +published for `main` as the baseline for its changed-line gate, so the main +workflow must publish successfully before that gate can evaluate a pull request. + ## Mutation testing Scheduled mutation testing runs in CI via diff --git a/docs/roadmap.md b/docs/roadmap.md index 4b743213..17151cd5 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -877,3 +877,21 @@ and usability. - [ ] 19.4.1. Ensure all public items have clear, useful documentation examples. - [ ] 19.4.2. Publish documentation to `docs.rs`. + +## 20. Prepared application and runtime ownership (in progress) + +This phase makes the builder-to-runtime ownership boundary explicit while +sequencing the remaining server and connection-runtime work separately. + +### 20.1. Prepared application transition + +- [x] 20.1.1. Add the consuming `WireframeApp::prepare().await` transition and + immutable `PreparedApp`, building route middleware chains once and providing + prepared connection drivers. See issue + [#641](https://github.com/leynos/wireframe/issues/641) and + [ADR 012](adr-012-prepared-application-and-connection-runtime.md). +- [ ] 20.1.2. Prepare the application factory before server readiness. See + issue [#642](https://github.com/leynos/wireframe/issues/642). +- [ ] 20.1.3. Extract connection-local runtime ownership and lifecycle + finalization. See issue + [#643](https://github.com/leynos/wireframe/issues/643). diff --git a/docs/users-guide.md b/docs/users-guide.md index b8f75147..f963c66a 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -44,8 +44,11 @@ For invariants and naming rules used across internal modules, see the A `WireframeApp` collects route handlers and middleware. Each handler is stored as an `Arc` pointing to an async function that receives a packet reference and -returns `()`. The builder caches these registrations until `handle_connection` -constructs the middleware chain for an accepted stream.[^2] +returns `()`. For manually accepted streams, consume the builder with +`prepare().await` after registration to construct immutable middleware chains +once and obtain a `PreparedApp`. The transition returns +`Result`; a preparation failure therefore produces +no partially usable runtime.[^2] ```no_run use std::sync::Arc; @@ -175,11 +178,31 @@ fn inspect_transport_source(error: &WireframeError) -> Option<&dyn Error> { } ``` -Once a stream is accepted—either from a manual accept loop or via -`WireframeServer`—`handle_connection(stream)` builds (or reuses) the middleware -chain, wraps the transport in the configured frame codec (length-delimited by -default), enforces per-frame read timeouts, and writes responses. Serialization -helpers `send_response` and `send_response_framed` (or +For a manually accepted stream, prepare the application once and then call +`PreparedApp::handle_connection_result(stream)` (or the logging +`PreparedApp::handle_connection(stream)` wrapper) for every connection. The +prepared application reuses its middleware chains, wraps each transport in the +configured frame codec (length-delimited by default), enforces per-frame read +timeouts, and writes responses. The deprecated +`WireframeApp::handle_connection` methods rebuild route chains for +compatibility and should not be used in new code. + +```rust,no_run +use tokio::io::duplex; +use wireframe::app::{PreparedApp, WireframeApp}; + +# async fn example() -> Result<(), Box> { +let prepared: PreparedApp = WireframeApp::new()?.prepare().await?; +let (client, server) = duplex(64); +drop(client); +prepared.handle_connection_result(server).await?; +# Ok(()) +# } +``` + +`WireframeServer` still accepts a builder factory and retains its existing +per-connection factory evaluation semantics until the server-runtime migration +lands. Serialization helpers `send_response` and `send_response_framed` (or `send_response_framed_with_codec` for custom codecs) return typed `SendError` variants when encoding or I/O fails, and the connection closes after ten consecutive deserialization errors.[^6][^7] @@ -1359,11 +1382,13 @@ additional ergonomics on top of the core primitives.[^13] `WireframeApp` supports optional setup and teardown callbacks that run once per connection. Setup can return arbitrary state retained until teardown executes -after the stream finishes processing.[^2] During `handle_connection` the -framework caches middleware chains, enforces read timeouts, and records metrics -for inbound frames, serialization failures, and handler errors before logging -warnings.[^6][^7] `PacketParts::inherit_correlation` ensures response packets -carry the correct correlation identifier even when middleware omits it.[^8] +after the stream finishes processing.[^2] Preparation moves those callback +definitions into the immutable `PreparedApp`; `PreparedApp::handle_connection` +reuses the prepared middleware chains, enforces read timeouts, and records +metrics for inbound frames, serialization failures, and handler errors before +logging warnings.[^6][^7] `PacketParts::inherit_correlation` ensures response +packets carry the correct correlation identifier even when middleware omits +it.[^8] Immediate responses are available through `send_response` and `send_response_framed`, both of which report serialization or I/O problems via @@ -2543,8 +2568,25 @@ When the optional `metrics` feature is enabled, Wireframe updates the `wireframe_connections_active` gauge, frame counters tagged by direction, error counters tagged by kind, and a counter for panicking connection tasks. All helpers become no-ops when the feature is disabled so instrumentation can stay -in place.[^33] `handle_connection`, the connection actor, and the panic wrapper -call these helpers to maintain consistent telemetry.[^6][^7][^31][^20] +in place.[^33] `PreparedApp::handle_connection`, the connection actor, and the +panic wrapper call these helpers to maintain consistent telemetry.[^6][^7][^31][^20] + +Prepared application lifecycle metrics are also emitted when the `metrics` +feature is enabled: + +- `wireframe_application_preparations_total` counts each preparation attempt. + Its bounded `outcome` label is `"success"` when an immutable `PreparedApp` is + produced and `"failure"` when preparation returns an error. +- `wireframe_application_preparation_duration_seconds` records the duration of + each preparation attempt, using the same `outcome` label values. +- `wireframe_prepared_connection_uses_total` counts each connection handled by + `PreparedApp::handle_connection_result` or its logging wrapper. It has no + labels. + +These metrics are emitted around preparation and prepared-application +connection handling, so repeated connections show reuse of the already-built +route services without repeating middleware transforms. All three helpers are +no-ops when the `metrics` feature is disabled. ## Mutation testing diff --git a/docs/v0-3-0-to-v0-4-0-migration-guide.md b/docs/v0-3-0-to-v0-4-0-migration-guide.md new file mode 100644 index 00000000..632d2671 --- /dev/null +++ b/docs/v0-3-0-to-v0-4-0-migration-guide.md @@ -0,0 +1,191 @@ +# v0.3.0 to v0.4.0 migration guide + +This guide covers the prepared-application transition for applications that +drive accepted streams directly. It explains how to move route and middleware +setup out of connection handling while retaining the existing server factory +workflow. + +## Prepared application transition + +`WireframeApp` remains the mutable builder. Register routes, middleware, +protocol hooks, and connection configuration on the builder, then consume it +with `prepare().await`: + +```rust,no_run +use std::sync::Arc; + +use wireframe::app::{Envelope, Handler, WireframeApp}; + +# async fn example() -> Result<(), Box> { +let handler: Handler = Arc::new(|_envelope| Box::pin(async {})); +let app = WireframeApp::new()?.route(1, handler)?; +let prepared = app.prepare().await?; +# let _ = prepared; +# Ok(()) +# } +``` + +Preparation consumes the builder. It transforms every registered route's +middleware chain once and returns an immutable `PreparedApp` containing those +services and the runtime configuration. The builder's route-registration +methods are therefore unavailable after the transition; register all routes and +middleware before calling `prepare`. + +`prepare` returns `Result`. Preparation is currently +infallible, but the typed error provides a stable place for callers to handle +future fallible middleware or runtime preparation steps. A failed preparation +does not expose a partially prepared application. + +## Reuse the prepared application + +Use the prepared connection methods for every accepted stream. Borrowing the +same `PreparedApp` lets multiple connections share the already-built route +services: + +```rust,no_run +use tokio::io::duplex; +use wireframe::app::{PreparedApp, WireframeApp}; + +# async fn example() -> Result<(), Box> { +let prepared: PreparedApp = WireframeApp::new()?.prepare().await?; + +let (client_one, server_one) = duplex(64); +drop(client_one); +prepared.handle_connection_result(server_one).await?; + +let (client_two, server_two) = duplex(64); +drop(client_two); +prepared.handle_connection_result(server_two).await?; +# Ok(()) +# } +``` + +`PreparedApp::handle_connection_result` returns stream-processing and handler +I/O errors. `PreparedApp::handle_connection` is the logging convenience wrapper +when the caller does not need to inspect that result. The prepared application +is immutable and has no route-registration surface. + +## Update test drivers + +Tests that use the `wireframe_testing` companion crate can prepare a builder +and drive one connection with `prepare_and_drive_with_frames`, or prepare once +and reuse the result with `drive_prepared_with_frames`: + +```rust,no_run +use wireframe::app::WireframeApp; +use wireframe_testing::{drive_prepared_with_frames, prepare_and_drive_with_frames}; + +# async fn example() -> std::io::Result<()> { +let app = WireframeApp::new().map_err(std::io::Error::other)?; +let _response = prepare_and_drive_with_frames(app, Vec::new()).await?; + +let app = WireframeApp::new().map_err(std::io::Error::other)?; +let prepared = app + .prepare() + .await + .map_err(|error| std::io::Error::other(error.to_string()))?; +let _response = drive_prepared_with_frames(&prepared, Vec::new()).await?; +# Ok(()) +# } +``` + +Both prepared helpers preserve custom `FrameCodec` types. Existing builder or +mutable drivers remain available as deprecated compatibility paths; migrate +tests to the prepared helpers when they need to prove one-time middleware +transformation or reuse prepared route services. + +## Migrate byte-handling APIs + +The prepared-application transition is independent of the zero-copy byte +migration. The v0.4 byte-facing APIs use `bytes::Bytes` (or the `PayloadBytes` +wrapper) for read-only hand-offs and an explicit edit-on-demand operation for +mutation. Middleware and hook editor APIs are not yet finalized; their +migration is deferred to roadmap items 12.1.2 and 12.2.1. The compatibility +helper names described below are defined by +[ADR 009](adr-009-vec-u8-migration-rollout.md). + +### Middleware + +The public edit-on-demand API for middleware requests and responses is not yet +finalized. Continue using the current `frame_mut()` and `into_inner()` +compatibility methods while this migration is tracked by roadmap item 12.1.2. +Do not assume a response-editor method or introduce an editor method until that +API is implemented and documented. Read-only middleware should avoid editing +the frame altogether. + +### Protocol and client hooks + +The hook editor API is also deferred to roadmap item 12.2.1. Keep existing +`Vec` hook implementations until that API is finalized; client preamble +leftovers intentionally remain `Vec` in this release. The compatibility +policy is defined in [ADR 009](adr-009-vec-u8-migration-rollout.md). + +### Serializers + +Serializer output moves from an owned vector to the stable byte wrapper. Use +`PayloadBytes::from_vec` only at an existing compatibility boundary, and keep +the zero-copy value through the codec hand-off: + +```text +# Before: serialization materializes a Vec for every outbound message. +let bytes: Vec = serializer.serialize(&message)?; +let frame = codec.wrap_payload(bytes::Bytes::from(bytes)); + +# After: the serializer returns the stable shared byte representation. +let bytes: PayloadBytes = serializer.serialize(&message)?; +let frame = codec.wrap_payload(bytes.into_bytes()); + +# Compatibility only: an older caller that still requires Vec. +let bytes: Vec = serializer.serialize_to_vec(&message)?; +``` + +`serialize_to_vec` is a temporary compatibility shim where provided; new code +should consume `PayloadBytes` directly. `PayloadBytes::into_vec` is likewise an +escape hatch, not the normal transport path. + +### Custom codecs + +Codecs should store payloads as `Bytes` when possible and override +`frame_payload_bytes` to return a cheap clone. The `wrap_payload` argument is +already `Bytes`, so only the frame type and extraction methods need changing: + +```rust +// Before: a custom frame owns a Vec payload. +struct MyEnvelope { + payload: Vec, +} + +// After: the frame shares its payload buffer with the codec driver. +use bytes::Bytes; + +struct MyEnvelope { + payload: Bytes, +} + +impl FrameCodec for MyCodec { + type Frame = MyEnvelope; + + fn frame_payload(frame: &MyEnvelope) -> &[u8] { &frame.payload } + + fn frame_payload_bytes(frame: &MyEnvelope) -> Bytes { frame.payload.clone() } + + fn wrap_payload(&self, payload: Bytes) -> MyEnvelope { MyEnvelope { payload } } +} +``` + +Keep `Vec` conversion at the edge of legacy callers with +`PayloadBytes::from_vec` or `PayloadBytes::into_vec`; do not add per-codec +conversion constructors. See +[ADR 008](adr-008-zero-copy-public-byte-container.md) for the read-only and +edit-on-demand design, and the +[zero-copy migration roadmap](zero-copy-frame-and-payload-migration-roadmap.md) +for the staged rollout. + +## Server factory compatibility + +`WireframeServer` continues to accept an `AppFactory` and retains its existing +factory-evaluation semantics in this release. Applications that construct a +fresh builder per connection therefore do not automatically share a prepared +application. Preparing the application factory before server readiness, and +moving server connection tasks onto a prepared root, are tracked separately in +[issue #642](https://github.com/leynos/wireframe/issues/642). diff --git a/docs/wireframe-testing-crate.md b/docs/wireframe-testing-crate.md index 338c6598..38ab2d15 100644 --- a/docs/wireframe-testing-crate.md +++ b/docs/wireframe-testing-crate.md @@ -59,8 +59,19 @@ rstest = "0.18.2" ## Codec-aware drivers -The helpers remain centred on a single in-memory driver that runs -`WireframeApp::handle_connection` against a `tokio::io::duplex` stream. The +The helpers use an in-memory driver over a `tokio::io::duplex` stream. New +tests should prepare a builder before driving a connection: `prepare().await` +consumes the `WireframeApp`, applies each route's middleware transforms once, +and returns an immutable `PreparedApp`. The prepared value can then be borrowed +by `drive_prepared_with_frames` for multiple connections without rebuilding its +route services. Both prepared helpers preserve the prepared codec type, so +custom `FrameCodec` implementations can use this migration path as well. +`PrepareError` is the typed preparation error; the convenience helper +`prepare_and_drive_with_frames` maps it to the helper's `io::Result` surface. + +The existing builder-oriented drivers remain compatibility paths for tests that +have not migrated. They call the deprecated `WireframeApp::handle_connection` +methods and therefore rebuild route chains for each driven connection. The driver is responsible for framing inbound and outbound data using the selected `FrameCodec` and for surfacing server panics as `io::Error` values prefixed with `server task failed`. @@ -88,7 +99,8 @@ length-delimited framing. ```rust,no_run use std::io; -use wireframe::app::{Packet, WireframeApp}; +use wireframe::app::{Packet, PreparedApp, WireframeApp}; +use wireframe::codec::FrameCodec; pub async fn drive_with_frames( app: WireframeApp, @@ -125,6 +137,26 @@ where S: TestSerializer, C: Send + 'static, E: Packet; + +pub async fn prepare_and_drive_with_frames( + app: WireframeApp, + frames: Vec>, +) -> io::Result> +where + S: TestSerializer, + C: Send + 'static, + E: Packet, + F: FrameCodec; + +pub async fn drive_prepared_with_frames( + app: &PreparedApp, + frames: Vec>, +) -> io::Result> +where + S: TestSerializer, + C: Send + 'static, + E: Packet, + F: FrameCodec; ``` Codec-aware helpers should be added as non-breaking extensions, so tests can @@ -140,8 +172,12 @@ Behavioural details: `drive_with_frames`. - `drive_with_bincode` encodes a message with bincode and then length-prefixes the output before driving the app. +- `prepare_and_drive_with_frames` prepares a builder and drives one connection. +- `drive_prepared_with_frames` borrows a `PreparedApp`, so tests can reuse the + same prepared route services across connections. - Mutable variants (`drive_with_frames_mut` and `drive_with_payloads_mut`) - accept `&mut WireframeApp` so tests can reuse a configured instance. + accept `&mut WireframeApp` and remain available for compatibility coverage; + they use the deprecated builder connection path. - I/O failures, framing errors, and server task panics are all returned as `io::Error` values, so tests can assert on error handling. diff --git a/examples/metadata_routing.rs b/examples/metadata_routing.rs index c0183a0d..da77a68a 100644 --- a/examples/metadata_routing.rs +++ b/examples/metadata_routing.rs @@ -108,6 +108,7 @@ async fn run() -> io::Result<()> { .map_err(|error| io::Error::other(error.to_string()))?; let mut codec = app.length_codec(); + let app = app.prepare().await.map_err(io::Error::other)?; let (mut client, server) = duplex(1024); let server_task = tokio::spawn(async move { app.handle_connection_result(server).await }); diff --git a/examples/packet_enum.rs b/examples/packet_enum.rs index 15515d29..57e4c6d7 100644 --- a/examples/packet_enum.rs +++ b/examples/packet_enum.rs @@ -121,7 +121,7 @@ fn parse_server_addr() -> std::io::Result { /// Initialize tracing, bind the listener, and serve until shutdown is signalled. async fn run() -> std::io::Result<()> { runtime_bootstrap::init_tracing(); - let app = runtime_bootstrap::build_runtime_app(build_app)?; + let app = runtime_bootstrap::build_runtime_app(build_app).await?; let listener = runtime_bootstrap::bind_listener(parse_server_addr()?).await?; runtime_bootstrap::serve_until_shutdown( listener, diff --git a/examples/ping_pong.rs b/examples/ping_pong.rs index d248be60..29e734e5 100644 --- a/examples/ping_pong.rs +++ b/examples/ping_pong.rs @@ -179,7 +179,7 @@ fn parse_server_addr() -> std::io::Result { /// same lifecycle as the example process. async fn run() -> std::io::Result<()> { runtime_bootstrap::init_tracing(); - let app = runtime_bootstrap::build_runtime_app(build_app)?; + let app = runtime_bootstrap::build_runtime_app(build_app).await?; let listener = runtime_bootstrap::bind_listener(parse_server_addr()?).await?; runtime_bootstrap::serve_until_shutdown( listener, diff --git a/examples/support/runtime_bootstrap.rs b/examples/support/runtime_bootstrap.rs index df5ddeb9..cc65758a 100644 --- a/examples/support/runtime_bootstrap.rs +++ b/examples/support/runtime_bootstrap.rs @@ -12,17 +12,18 @@ use crate::server_loop; /// Keeping the alias here ensures each example wires the same envelope and /// serializer contract into its runtime and connection tasks. type ExampleApp = wireframe::app::WireframeApp; - +/// Immutable runtime application shared by all accepted example connections. +type PreparedExampleApp = wireframe::app::PreparedApp; /// Initialize tracing for examples, ignoring duplicate global subscriber setup. pub(crate) fn init_tracing() { let _ = tracing_subscriber::fmt::try_init(); } /// Convert an example app builder into a shared runtime app handle. -pub(crate) fn build_runtime_app( +pub(crate) async fn build_runtime_app( build_app: impl FnOnce() -> wireframe::app::Result, -) -> std::io::Result> { - build_app() - .map(Arc::new) - .map_err(|error| std::io::Error::other(error.to_string())) +) -> std::io::Result> { + let app = build_app().map_err(|error| std::io::Error::other(error.to_string()))?; + let app = app.prepare().await.map_err(std::io::Error::other)?; + Ok(Arc::new(app)) } /// Bind a TCP listener for an already parsed socket address. @@ -31,7 +32,7 @@ pub(crate) async fn bind_listener(addr: SocketAddr) -> std::io::Result, stream: TcpStream) { +pub(crate) fn spawn_connection(app: Arc, stream: TcpStream) { tokio::spawn(async move { if let Err(error) = app.handle_connection_result(stream).await { error!("connection handling failed: {error}"); @@ -42,7 +43,7 @@ pub(crate) fn spawn_connection(app: Arc, stream: TcpStream) { /// Accept connections until shutdown and dispatch each stream to the app. pub(crate) async fn serve_until_shutdown( listener: TcpListener, - app: Arc, + app: Arc, shutdown_message: &'static str, ) -> std::io::Result<()> { while let Some(stream) = server_loop::accept_until_shutdown(&listener, shutdown_message).await? diff --git a/src/app/builder/core.rs b/src/app/builder/core.rs index 0c3c0d93..9f7025ec 100644 --- a/src/app/builder/core.rs +++ b/src/app/builder/core.rs @@ -2,7 +2,7 @@ use std::{collections::HashMap, sync::Arc}; -use tokio::sync::{OnceCell, mpsc}; +use tokio::sync::mpsc; use crate::{ app::{ @@ -17,7 +17,6 @@ use crate::{ codec::{FrameCodec, LengthDelimitedFrameCodec}, hooks::WireframeProtocol, message_assembler::MessageAssembler, - middleware::HandlerService, serializer::{BincodeSerializer, Serializer}, }; @@ -34,8 +33,6 @@ pub struct WireframeApp< > { /// Handler factories keyed by the protocol message identifier. pub(in crate::app) handlers: HashMap>, - /// Lazily built middleware chains, shared after the first connection uses them. - pub(in crate::app) routes: OnceCell>>>, /// Middleware applied in registration order around each handler. pub(in crate::app) middleware: Vec>>, /// Serializer retained by every connection built from this application. @@ -76,7 +73,6 @@ where let codec = F::default(); Self { handlers: HashMap::new(), - routes: OnceCell::new(), middleware: Vec::new(), serializer: S::default(), app_data: AppDataStore::default(), @@ -159,7 +155,7 @@ where { /// Helper to rebuild the app when changing type parameters. /// - /// The `WireframeApp` builder carries 14 fields that must be moved together + /// The `WireframeApp` builder carries 13 fields that must be moved together /// when swapping serializer or codec types. Centralizing the reconstruction /// here keeps the transitions consistent and avoids repeating the same /// field list across each type-changing method. For smaller builders with @@ -175,7 +171,6 @@ where { WireframeApp { handlers: self.handlers, - routes: OnceCell::new(), middleware: self.middleware, serializer: params.serializer, app_data: self.app_data, @@ -205,7 +200,6 @@ where { WireframeApp { handlers: self.handlers, - routes: OnceCell::new(), middleware: self.middleware, serializer: self.serializer, app_data: self.app_data, diff --git a/src/app/builder/routing.rs b/src/app/builder/routing.rs index f30907c8..ddbd42fe 100644 --- a/src/app/builder/routing.rs +++ b/src/app/builder/routing.rs @@ -30,7 +30,6 @@ where return Err(WireframeError::DuplicateRoute(id)); } self.handlers.insert(id, handler); - self.routes = tokio::sync::OnceCell::new(); Ok(self) } @@ -44,7 +43,6 @@ where M: Middleware + 'static, { self.middleware.push(Box::new(mw)); - self.routes = tokio::sync::OnceCell::new(); Ok(self) } } diff --git a/src/app/error.rs b/src/app/error.rs index c89c58e1..481fc43b 100644 --- a/src/app/error.rs +++ b/src/app/error.rs @@ -21,5 +21,22 @@ pub enum SendError { Codec(#[from] CodecError), } +/// Errors produced while preparing an application for connection handling. +/// +/// Preparation is currently infallible. The reserved middleware variant keeps +/// the transition typed so future fallible transforms can preserve their +/// source error without changing the public method signature. +#[derive(Debug, Error)] +#[non_exhaustive] +pub enum PrepareError { + /// A future middleware transform failed while preparing a route. + #[error("route middleware transformation failed: {source}")] + MiddlewareTransform { + /// The transform failure that prevented preparation. + #[source] + source: Box, + }, +} + /// Result type used throughout the builder API. pub type Result = crate::Result; diff --git a/src/app/inbound_handler.rs b/src/app/inbound_handler.rs index 775bcde0..e49103ca 100644 --- a/src/app/inbound_handler.rs +++ b/src/app/inbound_handler.rs @@ -1,74 +1,138 @@ -//! Inbound connection handling and response utilities for `WireframeApp`. +//! Inbound connection handling and route preparation utilities. + +mod core; use std::{collections::HashMap, sync::Arc}; -use futures::StreamExt; -use log::{debug, warn}; -use tokio::{ - io::{self, AsyncRead, AsyncWrite}, - time::{Duration, timeout}, -}; -use tokio_util::codec::Framed; +use log::warn; +use tokio::io::{self, AsyncRead, AsyncWrite}; use super::{ builder::WireframeApp, - codec_driver::FramePipeline, - combined_codec::{CombinedCodec, ConnectionCodec}, envelope::{Envelope, Packet}, - frame_handling, + lifecycle::{ConnectionSetup, ConnectionTeardown}, + memory_budgets::MemoryBudgets, + middleware_types::{Handler, Middleware}, }; use crate::{ - codec::{FrameCodec, MAX_FRAME_LENGTH, clamp_frame_length}, + codec::FrameCodec, frame::FrameMetadata, - message::{DecodeWith, DeserializeContext, EncodeWith}, - message_assembler::MessageAssemblyState, + message::{DecodeWith, EncodeWith}, + message_assembler::MessageAssembler, middleware::HandlerService, serializer::Serializer, }; -/// Remove stale outbound and message-assembly state after an idle interval. -fn purge_expired( - pipeline: &mut FramePipeline, - message_assembly: &mut Option, -) { - pipeline.purge_expired(); - frame_handling::purge_expired_assemblies(message_assembly); -} /// Maximum consecutive deserialization failures before closing a connection. -const MAX_DESER_FAILURES: u32 = 10; +pub(super) const MAX_DESER_FAILURES: u32 = 10; -/// Per-frame processing state bundled for `handle_frame`. -struct FrameHandlingContext<'a, E, W, F> +/// Immutable inputs required to drive one connection through prepared routes. +pub(crate) struct ConnectionProcessingContext<'a, S, C, E, F> where + S: Serializer + Send + Sync, + C: Send + 'static, E: Packet, - W: AsyncRead + AsyncWrite + Unpin, F: FrameCodec, { - /// Framed transport borrowed for response writes during frame handling. - framed: &'a mut Framed>, - /// Connection-wide malformed-frame counter shared with all stages. - deser_failures: &'a mut u32, /// Immutable middleware chains used to dispatch decoded envelopes. - routes: &'a HashMap>, - /// Outbound processing state for fragmenting and counting responses. - pipeline: &'a mut FramePipeline, - /// Connection-local state for assembling multi-frame messages. - message_assembly: &'a mut Option, + pub(crate) routes: &'a HashMap>, + /// Serializer shared by all frames on the connection. + pub(crate) serializer: &'a S, + /// Codec configuration shared by all frames on the connection. + pub(crate) codec: &'a F, + /// Optional hook that creates per-connection state. + pub(crate) on_connect: Option<&'a Arc>>, + /// Optional hook that releases per-connection state after processing. + pub(crate) on_disconnect: Option<&'a Arc>>, + /// Optional assembly strategy for multi-frame protocol messages. + pub(crate) message_assembler: Option<&'a Arc>, + /// Fragmentation settings used to initialize the frame pipeline. + pub(crate) fragmentation: Option, + /// Optional byte budgets enforced while processing the connection. + pub(crate) memory_budgets: Option, + /// Maximum interval to wait for the next inbound frame. + pub(crate) read_timeout_ms: u64, } -/// State needed to turn a raw frame into a dispatchable envelope. -struct DispatchBuildContext<'a, F> +/// Drive a connection using immutable application inputs and prepared routes. +pub(crate) async fn process_connection( + stream: W, + context: ConnectionProcessingContext<'_, S, C, E, F>, +) -> io::Result<()> where + S: Serializer + FrameMetadata + Send + Sync, + C: Send + 'static, + E: Packet, F: FrameCodec, + W: AsyncRead + AsyncWrite + Send + Unpin + 'static, + Envelope: DecodeWith + EncodeWith, { - /// Raw frame borrowed while decoding its envelope metadata and payload. - frame: &'a F::Frame, - /// Pipeline needed to reassemble fragmented input and emit responses. - pipeline: &'a mut FramePipeline, - /// Mutable assembly state retained across inbound frames. - message_assembly: &'a mut Option, - /// Failure counter used to enforce the malformed-input limit. - deser_failures: &'a mut u32, + let ConnectionProcessingContext { + routes, + serializer, + codec, + on_connect, + on_disconnect, + message_assembler, + fragmentation, + memory_budgets, + read_timeout_ms, + } = context; + let state = if let Some(setup) = on_connect { + Some(setup().await) + } else { + None + }; + + let processing_result = core::process_stream( + stream, + core::StreamProcessingContext { + routes, + serializer, + codec, + message_assembler, + fragmentation, + memory_budgets, + read_timeout_ms, + }, + ) + .await; + + if let (Some(teardown), Some(state)) = (on_disconnect, state) { + teardown(state).await; + } + + if let Err(error) = processing_result { + warn!( + "connection terminated with error: correlation_id={:?}, error={error:?}", + None:: + ); + return Err(error); + } + + Ok(()) +} + +/// Construct each route's middleware chain from registered builder inputs. +/// +/// Middleware is folded in reverse registration order, preserving the first +/// registered middleware as the outermost service layer. +pub(crate) async fn build_route_chains( + handlers: &HashMap>, + middleware: &[Box>], +) -> HashMap> +where + E: Packet, +{ + let mut routes = HashMap::new(); + for (&id, handler) in handlers { + let mut service = HandlerService::new(id, handler.clone()); + for mw in middleware.iter().rev() { + service = mw.transform(service).await; + } + routes.insert(id, service); + } + routes } impl WireframeApp @@ -79,261 +143,51 @@ where F: FrameCodec, Envelope: DecodeWith + EncodeWith, { - /// Try parsing the frame using [`FrameMetadata::parse`], falling back to - /// full deserialization on failure. - fn parse_envelope( - &self, - payload: &[u8], - ) -> std::result::Result<(Envelope, usize), Box> { - match self.serializer.parse(payload) { - Ok((parsed_envelope, metadata_bytes_consumed)) => { - if !self.serializer.should_deserialize_after_parse() { - return Ok((parsed_envelope, metadata_bytes_consumed)); - } - - let context = DeserializeContext { - frame_metadata: payload.get(..metadata_bytes_consumed), - message_id: Some(parsed_envelope.id), - correlation_id: parsed_envelope.correlation_id, - metadata_bytes_consumed: Some(metadata_bytes_consumed), - }; - self.serializer - .deserialize_with_context::(payload, &context) - } - Err(_) => self.serializer.deserialize::(payload), - } - } - - /// Handle an accepted connection end-to-end, returning any processing error. + /// Handle a connection through a compatibility route preparation path. /// /// # Errors /// /// Returns an [`io::Error`] if stream processing or handler execution fails. + #[deprecated(note = "prepare the app once, then call PreparedApp::handle_connection_result")] pub async fn handle_connection_result(&self, stream: W) -> io::Result<()> where W: AsyncRead + AsyncWrite + Send + Unpin + 'static, { - let state = if let Some(setup) = &self.on_connect { - Some((setup)().await) - } else { - None - }; - - let routes = self - .routes - .get_or_init(|| async { Arc::new(self.build_chains().await) }) - .await - .clone(); - - if let Err(e) = self.process_stream(stream, &routes).await { - warn!( - "connection terminated with error: correlation_id={:?}, error={e:?}", - None:: - ); - return Err(e); - } - - if let (Some(teardown), Some(state)) = (&self.on_disconnect, state) { - teardown(state).await; - } - - Ok(()) + let routes = build_route_chains(&self.handlers, &self.middleware).await; + process_connection( + stream, + ConnectionProcessingContext { + routes: &routes, + serializer: &self.serializer, + codec: &self.codec, + on_connect: self.on_connect.as_ref(), + on_disconnect: self.on_disconnect.as_ref(), + message_assembler: self.message_assembler.as_ref(), + fragmentation: self.fragmentation, + memory_budgets: self.memory_budgets, + read_timeout_ms: self.read_timeout_ms, + }, + ) + .await } - /// Handle an accepted connection end-to-end, logging errors and swallowing the result. + /// Handle a connection through the compatibility preparation path. + #[deprecated(note = "prepare the app once, then call PreparedApp::handle_connection")] pub async fn handle_connection(&self, stream: W) where W: AsyncRead + AsyncWrite + Send + Unpin + 'static, { - if let Err(e) = self.handle_connection_result(stream).await { + #[expect( + deprecated, + reason = "compatibility wrapper delegates to its fallible counterpart" + )] + if let Err(error) = self.handle_connection_result(stream).await { warn!( - "connection handling completed with error: correlation_id={:?}, error={e:?}", + "connection handling completed with error: correlation_id={:?}, error={error:?}", None:: ); } } - - /// Build middleware chains once, preserving reverse wrapping order. - async fn build_chains(&self) -> HashMap> { - let mut routes = HashMap::new(); - for (&id, handler) in &self.handlers { - let mut service = HandlerService::new(id, handler.clone()); - for mw in self.middleware.iter().rev() { - service = mw.transform(service).await; - } - routes.insert(id, service); - } - routes - } - - /// Read frames until EOF, timeout, or a transport/handler error occurs. - async fn process_stream( - &self, - stream: W, - routes: &Arc>>, - ) -> io::Result<()> - where - W: AsyncRead + AsyncWrite + Unpin, - { - let codec = self.codec.clone(); - let combined = CombinedCodec::new(codec.decoder(), codec.encoder()); - let mut framed = Framed::new(stream, combined); - let requested_frame_length = codec.max_frame_length(); - let max_frame_length = clamp_frame_length(requested_frame_length); - if requested_frame_length > MAX_FRAME_LENGTH { - warn!( - "codec max frame length exceeds guardrail; clamping to {MAX_FRAME_LENGTH} bytes \ - (requested={requested_frame_length})" - ); - } - framed.read_buffer_mut().reserve(max_frame_length); - let effective_budgets = - frame_handling::resolve_effective_budgets(self.memory_budgets, requested_frame_length); - let mut deser_failures = 0u32; - let mut message_assembly = self.message_assembler.as_ref().map(|_| { - frame_handling::new_message_assembly_state( - self.fragmentation, - requested_frame_length, - Some(effective_budgets), - ) - }); - let mut pipeline = FramePipeline::new(self.fragmentation); - let timeout_dur = Duration::from_millis(self.read_timeout_ms); - - loop { - let pressure = frame_handling::evaluate_memory_pressure( - message_assembly.as_ref(), - Some(effective_budgets), - ); - frame_handling::apply_memory_pressure(pressure, || { - purge_expired(&mut pipeline, &mut message_assembly); - }) - .await?; - - match timeout(timeout_dur, framed.next()).await { - Ok(Some(Ok(frame))) => { - self.handle_frame( - &frame, - FrameHandlingContext { - framed: &mut framed, - deser_failures: &mut deser_failures, - routes, - message_assembly: &mut message_assembly, - pipeline: &mut pipeline, - }, - &codec, - ) - .await?; - } - Ok(Some(Err(e))) => return Err(e), - Ok(None) => break, - Err(_) => { - debug!("read timeout elapsed; continuing to wait for next frame"); - purge_expired(&mut pipeline, &mut message_assembly); - } - } - } - - Ok(()) - } - - /// Decode one frame, apply reassembly, and dispatch its response. - async fn handle_frame( - &self, - frame: &F::Frame, - ctx: FrameHandlingContext<'_, E, W, F>, - codec: &F, - ) -> io::Result<()> - where - W: AsyncRead + AsyncWrite + Unpin, - { - let FrameHandlingContext { - framed, - deser_failures, - routes, - message_assembly, - pipeline, - } = ctx; - - crate::metrics::inc_frames(crate::metrics::Direction::Inbound); - let Some(env) = self.build_dispatchable_envelope(DispatchBuildContext { - frame, - pipeline, - message_assembly, - deser_failures, - })? - else { - return Ok(()); - }; - - if let Some(service) = routes.get(&env.id) { - frame_handling::forward_response( - env, - service, - frame_handling::ResponseContext:: { - serializer: &self.serializer, - framed, - pipeline, - codec, - }, - ) - .await?; - } else { - warn!( - "no handler for message id: id={}, correlation_id={:?}", - env.id, env.correlation_id - ); - } - - Ok(()) - } - - /// Run decode, fragment reassembly, and message assembly in order. - fn build_dispatchable_envelope( - &self, - ctx: DispatchBuildContext<'_, F>, - ) -> io::Result> { - let DispatchBuildContext { - frame, - pipeline, - message_assembly, - deser_failures, - } = ctx; - let mut failure_tracker = - frame_handling::DeserFailureTracker::new(deser_failures, MAX_DESER_FAILURES); - let Some(env) = frame_handling::decode_envelope::( - self.parse_envelope(F::frame_payload(frame)), - frame, - &mut failure_tracker, - )? - else { - return Ok(None); - }; - let Some(env) = frame_handling::reassemble_if_needed( - pipeline, - deser_failures, - env, - MAX_DESER_FAILURES, - )? - else { - return Ok(None); - }; - let Some(env) = frame_handling::assemble_if_needed( - frame_handling::AssemblyRuntime::new(self.message_assembler.as_ref(), message_assembly), - deser_failures, - env, - MAX_DESER_FAILURES, - )? - else { - return Ok(None); - }; - - // Reset failure counter only after the entire inbound pipeline - // (decode, reassemble, assemble) succeeds, so that assembly-stage - // failures accumulate towards the threshold. - *deser_failures = 0; - Ok(Some(env)) - } } #[cfg(test)] diff --git a/src/app/inbound_handler/core.rs b/src/app/inbound_handler/core.rs new file mode 100644 index 00000000..84dc51aa --- /dev/null +++ b/src/app/inbound_handler/core.rs @@ -0,0 +1,329 @@ +//! Shared frame and stream processing for prepared application routes. + +use std::{collections::HashMap, sync::Arc}; + +use futures::StreamExt; +use log::{debug, warn}; +use tokio::{ + io::{self, AsyncRead, AsyncWrite}, + time::{Duration, timeout}, +}; +use tokio_util::codec::Framed; + +use super::{ + super::{ + codec_driver::FramePipeline, + combined_codec::{CombinedCodec, ConnectionCodec}, + envelope::{Envelope, Packet}, + frame_handling, + memory_budgets::MemoryBudgets, + }, + MAX_DESER_FAILURES, +}; +use crate::{ + codec::{FrameCodec, MAX_FRAME_LENGTH, clamp_frame_length}, + frame::FrameMetadata, + message::{DecodeWith, DeserializeContext, EncodeWith}, + message_assembler::{MessageAssembler, MessageAssemblyState}, + middleware::HandlerService, + serializer::Serializer, +}; + +/// Per-frame processing state bundled for `handle_frame`. +struct FrameHandlingContext<'a, S, E, W, F> +where + S: Serializer + Send + Sync, + E: Packet, + W: AsyncRead + AsyncWrite + Unpin, + F: FrameCodec, +{ + /// Framed transport used to write responses during frame handling. + framed: &'a mut Framed>, + /// Connection-wide malformed-frame counter shared with all stages. + deser_failures: &'a mut u32, + /// Immutable middleware chains used to dispatch decoded envelopes. + routes: &'a HashMap>, + /// Serializer used to decode envelopes and encode responses. + serializer: &'a S, + /// Codec configuration used by the response path. + codec: &'a F, + /// Optional assembly strategy for multi-frame protocol messages. + message_assembler: Option<&'a Arc>, + /// Outbound processing state for fragmenting and counting responses. + pipeline: &'a mut FramePipeline, + /// Connection-local state for assembling multi-frame messages. + message_assembly: &'a mut Option, +} + +/// Immutable stream-wide configuration shared by all inbound frames. +pub(super) struct StreamProcessingContext<'a, S, E, F> +where + S: Serializer + Send + Sync, + E: Packet, + F: FrameCodec, +{ + /// Immutable middleware chains used to dispatch decoded envelopes. + pub(super) routes: &'a HashMap>, + /// Serializer shared by all frames on the connection. + pub(super) serializer: &'a S, + /// Codec configuration shared by all frames on the connection. + pub(super) codec: &'a F, + /// Optional assembly strategy for multi-frame protocol messages. + pub(super) message_assembler: Option<&'a Arc>, + /// Fragmentation settings used to initialize the frame pipeline. + pub(super) fragmentation: Option, + /// Optional byte budgets enforced while processing the connection. + pub(super) memory_budgets: Option, + /// Maximum interval to wait for the next inbound frame. + pub(super) read_timeout_ms: u64, +} + +/// State needed to turn a raw frame into a dispatchable envelope. +struct DispatchBuildContext<'a, F> +where + F: FrameCodec, +{ + /// Raw frame borrowed while decoding its envelope metadata and payload. + frame: &'a F::Frame, + /// Pipeline needed to reassemble fragmented input and emit responses. + pipeline: &'a mut FramePipeline, + /// Mutable assembly state retained across inbound frames. + message_assembly: &'a mut Option, + /// Failure counter used to enforce the malformed-input limit. + deser_failures: &'a mut u32, +} + +/// Remove stale outbound and message-assembly state after an idle interval. +fn purge_expired( + pipeline: &mut FramePipeline, + message_assembly: &mut Option, +) { + pipeline.purge_expired(); + frame_handling::purge_expired_assemblies(message_assembly); +} + +/// Parse envelope metadata, falling back to full deserialization when needed. +pub(super) fn parse_envelope( + serializer: &S, + payload: &[u8], +) -> std::result::Result<(Envelope, usize), Box> +where + S: Serializer + FrameMetadata + Send + Sync, + Envelope: DecodeWith, +{ + match serializer.parse(payload) { + Ok((parsed_envelope, metadata_bytes_consumed)) => { + if !serializer.should_deserialize_after_parse() { + return Ok((parsed_envelope, metadata_bytes_consumed)); + } + + let context = DeserializeContext { + frame_metadata: payload.get(..metadata_bytes_consumed), + message_id: Some(parsed_envelope.id), + correlation_id: parsed_envelope.correlation_id, + metadata_bytes_consumed: Some(metadata_bytes_consumed), + }; + serializer.deserialize_with_context::(payload, &context) + } + Err(_) => serializer.deserialize::(payload), + } +} + +/// Read and process frames until the stream closes or an I/O error occurs. +pub(super) async fn process_stream( + stream: W, + context: StreamProcessingContext<'_, S, E, F>, +) -> io::Result<()> +where + S: Serializer + FrameMetadata + Send + Sync, + E: Packet, + F: FrameCodec, + W: AsyncRead + AsyncWrite + Unpin, + Envelope: DecodeWith + EncodeWith, +{ + let StreamProcessingContext { + routes, + serializer, + codec, + message_assembler, + fragmentation, + memory_budgets, + read_timeout_ms, + } = context; + // Each connection needs isolated framing state: cloning resets the + // counters `SeqFrameCodec` and `TaggedFrameCodec::wrap_payload` consume. + let codec = codec.clone(); + let combined = CombinedCodec::new(codec.decoder(), codec.encoder()); + let mut framed = Framed::new(stream, combined); + let requested_frame_length = codec.max_frame_length(); + let max_frame_length = clamp_frame_length(requested_frame_length); + if requested_frame_length > MAX_FRAME_LENGTH { + warn!( + "codec max frame length exceeds guardrail; clamping to {MAX_FRAME_LENGTH} bytes \ + (requested={requested_frame_length})" + ); + } + framed.read_buffer_mut().reserve(max_frame_length); + let effective_budgets = + frame_handling::resolve_effective_budgets(memory_budgets, requested_frame_length); + let mut deser_failures = 0u32; + let mut message_assembly = message_assembler.map(|_| { + frame_handling::new_message_assembly_state( + fragmentation, + requested_frame_length, + Some(effective_budgets), + ) + }); + let mut pipeline = FramePipeline::new(fragmentation); + let timeout_dur = Duration::from_millis(read_timeout_ms); + + loop { + let pressure = frame_handling::evaluate_memory_pressure( + message_assembly.as_ref(), + Some(effective_budgets), + ); + frame_handling::apply_memory_pressure(pressure, || { + purge_expired(&mut pipeline, &mut message_assembly); + }) + .await?; + + match timeout(timeout_dur, framed.next()).await { + Ok(Some(Ok(frame))) => { + handle_frame( + &frame, + FrameHandlingContext { + framed: &mut framed, + deser_failures: &mut deser_failures, + routes, + serializer, + codec: &codec, + message_assembler, + message_assembly: &mut message_assembly, + pipeline: &mut pipeline, + }, + ) + .await?; + } + Ok(Some(Err(error))) => return Err(error), + Ok(None) => break, + Err(_) => { + debug!("read timeout elapsed; continuing to wait for next frame"); + purge_expired(&mut pipeline, &mut message_assembly); + } + } + } + + Ok(()) +} + +/// Decode one frame, apply reassembly, and dispatch the resulting envelope. +async fn handle_frame( + frame: &F::Frame, + context: FrameHandlingContext<'_, S, E, W, F>, +) -> io::Result<()> +where + S: Serializer + FrameMetadata + Send + Sync, + E: Packet, + F: FrameCodec, + W: AsyncRead + AsyncWrite + Unpin, + Envelope: DecodeWith + EncodeWith, +{ + let FrameHandlingContext { + framed, + deser_failures, + routes, + serializer, + codec, + message_assembler, + message_assembly, + pipeline, + } = context; + + crate::metrics::inc_frames(crate::metrics::Direction::Inbound); + let Some(envelope) = build_dispatchable_envelope( + serializer, + message_assembler, + DispatchBuildContext:: { + frame, + pipeline, + message_assembly, + deser_failures, + }, + )? + else { + return Ok(()); + }; + + if let Some(service) = routes.get(&envelope.id) { + frame_handling::forward_response( + envelope, + service, + frame_handling::ResponseContext:: { + serializer, + framed, + pipeline, + codec, + }, + ) + .await?; + } else { + warn!( + "no handler for message id: id={}, correlation_id={:?}", + envelope.id, envelope.correlation_id + ); + } + + Ok(()) +} + +/// Build a dispatchable envelope through decode, reassembly, and assembly. +fn build_dispatchable_envelope( + serializer: &S, + message_assembler: Option<&Arc>, + context: DispatchBuildContext<'_, F>, +) -> io::Result> +where + S: Serializer + FrameMetadata + Send + Sync, + F: FrameCodec, + Envelope: DecodeWith, +{ + let DispatchBuildContext { + frame, + pipeline, + message_assembly, + deser_failures, + } = context; + let mut failure_tracker = + frame_handling::DeserFailureTracker::new(deser_failures, MAX_DESER_FAILURES); + let Some(envelope) = frame_handling::decode_envelope::( + parse_envelope(serializer, F::frame_payload(frame)), + frame, + &mut failure_tracker, + )? + else { + return Ok(None); + }; + let Some(envelope) = frame_handling::reassemble_if_needed( + pipeline, + deser_failures, + envelope, + MAX_DESER_FAILURES, + )? + else { + return Ok(None); + }; + let Some(envelope) = frame_handling::assemble_if_needed( + frame_handling::AssemblyRuntime::new(message_assembler, message_assembly), + deser_failures, + envelope, + MAX_DESER_FAILURES, + )? + else { + return Ok(None); + }; + + // Reset only after the entire pipeline succeeds, so assembly failures + // accumulate towards the close threshold. + *deser_failures = 0; + Ok(Some(envelope)) +} diff --git a/src/app/inbound_handler/tests.rs b/src/app/inbound_handler/tests.rs index e1982193..dfde8090 100644 --- a/src/app/inbound_handler/tests.rs +++ b/src/app/inbound_handler/tests.rs @@ -5,7 +5,7 @@ use tokio_util::codec::{Decoder, Encoder}; use wireframe_testing::logger; use super::*; -use crate::serializer::BincodeSerializer; +use crate::{app::frame_handling, serializer::BincodeSerializer}; #[derive(Clone, Debug)] struct BadFrame { @@ -78,7 +78,7 @@ fn decode_envelope_tracks_failures_and_logs_correlation_id() { let mut failure_tracker = frame_handling::DeserFailureTracker::new(&mut deser_failures, MAX_DESER_FAILURES); let result = frame_handling::decode_envelope::( - app.parse_envelope(BadCodec::frame_payload(&frame)), + core::parse_envelope(&app.serializer, BadCodec::frame_payload(&frame)), &frame, &mut failure_tracker, ); @@ -89,7 +89,7 @@ fn decode_envelope_tracks_failures_and_logs_correlation_id() { let mut failure_tracker = frame_handling::DeserFailureTracker::new(&mut deser_failures, MAX_DESER_FAILURES); let err = frame_handling::decode_envelope::( - app.parse_envelope(BadCodec::frame_payload(&frame)), + core::parse_envelope(&app.serializer, BadCodec::frame_payload(&frame)), &frame, &mut failure_tracker, ) diff --git a/src/app/mod.rs b/src/app/mod.rs index 46e5eb71..289592a7 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -23,10 +23,12 @@ mod memory_budgets; mod middleware_types; mod outbound_encoding; mod outbound_response; +mod prepared_app; pub use builder::WireframeApp; pub use envelope::{Envelope, Packet, PacketParts}; -pub use error::{Result, SendError}; +pub use error::{PrepareError, Result, SendError}; pub use lifecycle::{ConnectionSetup, ConnectionTeardown}; pub use memory_budgets::{BudgetBytes, MemoryBudgets}; pub use middleware_types::{Handler, Middleware}; +pub use prepared_app::PreparedApp; diff --git a/src/app/prepared_app.rs b/src/app/prepared_app.rs new file mode 100644 index 00000000..e56aa23f --- /dev/null +++ b/src/app/prepared_app.rs @@ -0,0 +1,380 @@ +//! Immutable application data prepared for connection handling. + +use std::{ + collections::HashMap, + sync::Arc, + time::{Duration, Instant}, +}; + +use tokio::{ + io::{self, AsyncRead, AsyncWrite}, + sync::mpsc, +}; +use tracing::Instrument as _; + +use super::{ + PrepareError, + builder::WireframeApp, + envelope::Packet, + inbound_handler::{ConnectionProcessingContext, build_route_chains, process_connection}, + lifecycle::{ConnectionSetup, ConnectionTeardown}, + memory_budgets::MemoryBudgets, +}; +use crate::{ + app_data_store::AppDataStore, + codec::{FrameCodec, LengthDelimitedFrameCodec}, + frame::FrameMetadata, + hooks::WireframeProtocol, + message::{DecodeWith, EncodeWith}, + message_assembler::MessageAssembler, + metrics::{self, PreparationOutcome}, + middleware::HandlerService, + serializer::{BincodeSerializer, Serializer}, +}; + +/// An immutable application template with fully transformed route services. +/// +/// Obtain this type by consuming a [`WireframeApp`] with +/// [`WireframeApp::prepare`]. It deliberately has no route-registration API, +/// so all handler and middleware transforms have completed before connections +/// start using the application. +pub struct PreparedApp< + S: Serializer + Send + Sync = BincodeSerializer, + C: Send + 'static = (), + E: Packet = super::Envelope, + F: FrameCodec = LengthDelimitedFrameCodec, +> { + /// Fully transformed route services keyed by protocol message identifier. + pub(in crate::app) routes: HashMap>, + /// Serializer retained by every connection driven by this application. + pub(in crate::app) serializer: S, + /// Codec template used to configure each connection's framed transport. + pub(in crate::app) codec: F, + // Retain this template-owned state until the connection-local runtime in + // https://github.com/leynos/wireframe/issues/643 consumes it. + #[expect( + dead_code, + reason = "tracked by issue #643: ConnectionRuntime will consume application data" + )] + /// Type-erased application state retained for the connection-runtime slice. + pub(in crate::app) app_data: AppDataStore, + /// Optional hook that creates per-connection state. + pub(in crate::app) on_connect: Option>>, + /// Optional hook that releases per-connection state after processing. + pub(in crate::app) on_disconnect: Option>>, + /// Optional protocol hook used to customize frame-level processing. + pub(in crate::app) protocol: + Option>>, + /// Optional assembler for protocol messages spread across several frames. + pub(in crate::app) message_assembler: Option>, + // Retain this template-owned configuration until the connection-local + // runtime in https://github.com/leynos/wireframe/issues/643 consumes it. + #[expect( + dead_code, + reason = "tracked by issue #643: ConnectionRuntime will consume the push DLQ" + )] + /// Optional dead-letter sink for pushes that cannot be delivered. + pub(in crate::app) push_dlq: Option>>, + /// Optional limits and timeout for transparent frame fragmentation. + pub(in crate::app) fragmentation: Option, + /// Maximum interval to wait for the next inbound frame. + pub(in crate::app) read_timeout_ms: u64, + /// Optional byte caps protecting message and connection memory usage. + pub(in crate::app) memory_budgets: Option, +} + +/// Supplies elapsed durations for preparation instrumentation. +trait PreparationTimeSource { + /// Opaque point captured at the beginning of a preparation transition. + type StartedAt; + + /// Capture a point from which preparation duration is measured. + fn start(&self) -> Self::StartedAt; + + /// Return the elapsed duration since a captured preparation point. + fn elapsed(&self, started_at: Self::StartedAt) -> Duration; +} + +/// Production time source backed by the monotonic standard-library clock. +struct SystemPreparationTimeSource; + +impl PreparationTimeSource for SystemPreparationTimeSource { + type StartedAt = Instant; + + fn start(&self) -> Self::StartedAt { Instant::now() } + + fn elapsed(&self, started_at: Self::StartedAt) -> Duration { started_at.elapsed() } +} + +/// Supplies elapsed durations for prepared-connection instrumentation. +trait ConnectionTimeSource { + /// Opaque point captured at the beginning of prepared connection handling. + type StartedAt; + + /// Capture a point from which connection duration is measured. + fn start(&self) -> Self::StartedAt; + + /// Return the elapsed duration since a captured connection point. + fn elapsed(&self, started_at: Self::StartedAt) -> Duration; +} + +/// Production time source backed by the monotonic standard-library clock. +struct SystemConnectionTimeSource; + +impl ConnectionTimeSource for SystemConnectionTimeSource { + type StartedAt = Instant; + + fn start(&self) -> Self::StartedAt { Instant::now() } + + fn elapsed(&self, started_at: Self::StartedAt) -> Duration { started_at.elapsed() } +} + +impl WireframeApp +where + S: Serializer + Send + Sync, + C: Send + 'static, + E: Packet, + F: FrameCodec, +{ + /// Consume builder registrations and prepare immutable route services. + /// + /// Middleware transforms run once for every registered route during this + /// transition. The returned template can then drive multiple connections + /// without rebuilding its route chains. + /// + /// # Examples + /// + /// ``` + /// use wireframe::app::{PreparedApp, WireframeApp}; + /// + /// # async fn example() -> Result<(), Box> { + /// let prepared: PreparedApp = WireframeApp::new()?.prepare().await?; + /// # let _ = prepared; + /// # Ok(()) + /// # } + /// ``` + /// + /// # Errors + /// + /// Returns [`PrepareError`] if a future fallible preparation step fails. + pub async fn prepare(self) -> Result, PrepareError> { + self.prepare_with_time_source(&SystemPreparationTimeSource) + .await + } + + /// Prepare the application with an injectable instrumentation time source. + async fn prepare_with_time_source( + self, + time_source: &T, + ) -> Result, PrepareError> + where + T: PreparationTimeSource, + { + let started_at = time_source.start(); + let result = self.build_prepared().await; + let outcome = if result.is_ok() { + PreparationOutcome::Success + } else { + PreparationOutcome::Failure + }; + metrics::record_application_preparation(outcome, time_source.elapsed(started_at)); + result + } + + /// Build the prepared representation before publishing it to the caller. + async fn build_prepared(self) -> Result, PrepareError> { + let routes = build_route_chains(&self.handlers, &self.middleware).await; + + Ok(PreparedApp { + routes, + serializer: self.serializer, + codec: self.codec, + app_data: self.app_data, + on_connect: self.on_connect, + on_disconnect: self.on_disconnect, + protocol: self.protocol, + message_assembler: self.message_assembler, + push_dlq: self.push_dlq, + fragmentation: self.fragmentation, + read_timeout_ms: self.read_timeout_ms, + memory_budgets: self.memory_budgets, + }) + } +} + +impl PreparedApp +where + S: Serializer + FrameMetadata + Send + Sync, + C: Send + 'static, + E: Packet, + F: FrameCodec, + super::Envelope: DecodeWith + EncodeWith, +{ + /// Handle an accepted connection using the prepared route services. + /// + /// # Examples + /// + /// ``` + /// use tokio::io::duplex; + /// use wireframe::app::{PreparedApp, WireframeApp}; + /// + /// # async fn example() -> Result<(), Box> { + /// let prepared: PreparedApp = WireframeApp::new()?.prepare().await?; + /// let (client, server) = duplex(64); + /// drop(client); + /// prepared.handle_connection_result(server).await?; + /// # Ok(()) + /// # } + /// ``` + /// + /// # Errors + /// + /// Returns an [`io::Error`] if stream processing or handler execution fails. + pub async fn handle_connection_result(&self, stream: W) -> io::Result<()> + where + W: AsyncRead + AsyncWrite + Send + Unpin + 'static, + { + self.handle_connection_with_time_source(stream, &SystemConnectionTimeSource) + .await + } + + /// Handle a connection with an injectable instrumentation time source. + async fn handle_connection_with_time_source( + &self, + stream: W, + time_source: &T, + ) -> io::Result<()> + where + W: AsyncRead + AsyncWrite + Send + Unpin + 'static, + T: ConnectionTimeSource, + { + metrics::inc_prepared_connection_uses(); + let started_at = time_source.start(); + let span = tracing::info_span!( + "prepared_connection", + outcome = tracing::field::Empty, + elapsed_ms = tracing::field::Empty + ); + let result = process_connection( + stream, + ConnectionProcessingContext { + routes: &self.routes, + serializer: &self.serializer, + codec: &self.codec, + on_connect: self.on_connect.as_ref(), + on_disconnect: self.on_disconnect.as_ref(), + message_assembler: self.message_assembler.as_ref(), + fragmentation: self.fragmentation, + memory_budgets: self.memory_budgets, + read_timeout_ms: self.read_timeout_ms, + }, + ) + .instrument(span.clone()) + .await; + span.record("outcome", if result.is_ok() { "success" } else { "error" }); + let elapsed_ms = + u64::try_from(time_source.elapsed(started_at).as_millis()).unwrap_or(u64::MAX); + span.record("elapsed_ms", elapsed_ms); + result + } + + /// Handle an accepted connection and log any processing failure. + /// + /// # Examples + /// + /// ``` + /// use tokio::io::duplex; + /// use wireframe::app::{PreparedApp, WireframeApp}; + /// + /// # async fn example() -> Result<(), Box> { + /// let prepared: PreparedApp = WireframeApp::new()?.prepare().await?; + /// let (client, server) = duplex(64); + /// drop(client); + /// prepared.handle_connection(server).await; + /// # Ok(()) + /// # } + /// ``` + pub async fn handle_connection(&self, stream: W) + where + W: AsyncRead + AsyncWrite + Send + Unpin + 'static, + { + if let Err(error) = self.handle_connection_result(stream).await { + log::warn!( + "connection handling completed with error: correlation_id={:?}, error={error:?}", + None:: + ); + } + } +} + +impl PreparedApp +where + S: Serializer + Send + Sync, + C: Send + 'static, + E: Packet, + F: FrameCodec, +{ + /// Get a clone of the configured protocol, if any. + /// + /// # Examples + /// + /// ```no_run + /// use wireframe::app::{PreparedApp, WireframeApp}; + /// + /// # async fn example() -> Result<(), Box> { + /// let prepared: PreparedApp = WireframeApp::new()?.prepare().await?; + /// assert!(prepared.protocol().is_none()); + /// # Ok(()) + /// # } + /// ``` + #[must_use] + pub fn protocol( + &self, + ) -> Option>> { + self.protocol.clone() + } + + /// Return protocol hooks derived from the installed protocol. + /// + /// # Examples + /// + /// ```no_run + /// use wireframe::app::{PreparedApp, WireframeApp}; + /// + /// # async fn example() -> Result<(), Box> { + /// let prepared: PreparedApp = WireframeApp::new()?.prepare().await?; + /// let hooks = prepared.protocol_hooks(); + /// # let _ = hooks; + /// # Ok(()) + /// # } + /// ``` + #[must_use] + pub fn protocol_hooks(&self) -> crate::hooks::ProtocolHooks { + self.protocol + .as_ref() + .map(crate::hooks::ProtocolHooks::from_protocol) + .unwrap_or_default() + } + + /// Get the configured message assembler, if any. + /// + /// # Examples + /// + /// ```no_run + /// use wireframe::app::{PreparedApp, WireframeApp}; + /// + /// # async fn example() -> Result<(), Box> { + /// let prepared: PreparedApp = WireframeApp::new()?.prepare().await?; + /// assert!(prepared.message_assembler().is_none()); + /// # Ok(()) + /// # } + /// ``` + #[must_use] + pub fn message_assembler(&self) -> Option<&Arc> { + self.message_assembler.as_ref() + } +} + +#[cfg(test)] +#[path = "prepared_app_tests.rs"] +mod tests; diff --git a/src/app/prepared_app_tests.rs b/src/app/prepared_app_tests.rs new file mode 100644 index 00000000..b78e59f4 --- /dev/null +++ b/src/app/prepared_app_tests.rs @@ -0,0 +1,81 @@ +//! Tests for prepared-application instrumentation seams. + +use std::{cell::Cell, time::Duration}; + +use super::*; + +/// Deterministic preparation time source that records the calls it serves. +struct FixedPreparationTimeSource { + starts: Cell, + elapsed: Cell, +} + +impl PreparationTimeSource for FixedPreparationTimeSource { + type StartedAt = (); + + fn start(&self) -> Self::StartedAt { self.starts.set(self.starts.get() + 1); } + + fn elapsed(&self, _started_at: Self::StartedAt) -> Duration { + self.elapsed.set(self.elapsed.get() + 1); + Duration::from_millis(1) + } +} + +/// Deterministic connection time source that records the calls it serves. +struct FixedConnectionTimeSource { + starts: Cell, + elapsed: Cell, +} + +impl ConnectionTimeSource for FixedConnectionTimeSource { + type StartedAt = (); + + fn start(&self) -> Self::StartedAt { self.starts.set(self.starts.get() + 1); } + + fn elapsed(&self, _started_at: Self::StartedAt) -> Duration { + self.elapsed.set(self.elapsed.get() + 1); + Duration::from_millis(1) + } +} + +/// Preparation records timing through the injected source exactly once. +#[tokio::test] +async fn prepare_uses_injected_time_source() { + let app: WireframeApp = WireframeApp::new().expect("app should initialize"); + let time_source = FixedPreparationTimeSource { + starts: Cell::new(0), + elapsed: Cell::new(0), + }; + + let _prepared = app + .prepare_with_time_source(&time_source) + .await + .expect("preparation should succeed"); + + assert_eq!(time_source.starts.get(), 1); + assert_eq!(time_source.elapsed.get(), 1); +} + +/// Prepared connection tracing records timing through the injected source. +#[tokio::test] +async fn prepared_connection_uses_injected_time_source() { + let prepared: PreparedApp = WireframeApp::new() + .expect("app should initialize") + .prepare() + .await + .expect("preparation should succeed"); + let time_source = FixedConnectionTimeSource { + starts: Cell::new(0), + elapsed: Cell::new(0), + }; + let (client, server) = tokio::io::duplex(64); + drop(client); + + prepared + .handle_connection_with_time_source(server, &time_source) + .await + .expect("clean EOF should complete"); + + assert_eq!(time_source.starts.get(), 1); + assert_eq!(time_source.elapsed.get(), 1); +} diff --git a/src/metrics.rs b/src/metrics.rs index 781297df..14675b11 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -15,8 +15,10 @@ //! println!("{}", handle.render()); //! ``` +use std::time::Duration; + #[cfg(feature = "metrics")] -use metrics::{counter, gauge}; +use metrics::{counter, gauge, histogram}; /// Name of the gauge tracking active connections. pub const CONNECTIONS_ACTIVE: &str = "wireframe_connections_active"; @@ -52,6 +54,18 @@ pub const POOL_BOOKKEEPING_POISON_RECOVERIES: &str = /// ``` pub const CODEC_ERRORS: &str = "wireframe_codec_errors_total"; +/// Name of the counter tracking application preparation outcomes. +/// +/// The bounded `outcome` label is either `"success"` or `"failure"`. +pub const APPLICATION_PREPARATIONS: &str = "wireframe_application_preparations_total"; + +/// Name of the counter tracking connections handled by prepared applications. +pub const PREPARED_CONNECTION_USES: &str = "wireframe_prepared_connection_uses_total"; + +/// Name of the histogram recording application preparation duration in seconds. +pub const APPLICATION_PREPARATION_DURATION: &str = + "wireframe_application_preparation_duration_seconds"; + /// Name of the counter tracking server-supervisor cancellation requests. /// /// The `reason` label is always either `"graceful"`, when the supplied @@ -96,6 +110,25 @@ impl ServerCancellationReason { } } +/// Bounded outcomes for application preparation metrics. +#[derive(Clone, Copy)] +pub(crate) enum PreparationOutcome { + /// Preparation completed and produced an immutable application template. + Success, + /// Preparation failed before exposing an application template. + Failure, +} + +impl PreparationOutcome { + /// Return the stable metric label value for this preparation outcome. + pub(crate) const fn as_str(self) -> &'static str { + match self { + Self::Success => "success", + Self::Failure => "failure", + } + } +} + /// Direction of frame processing. #[derive(Clone, Copy)] pub enum Direction { @@ -239,6 +272,30 @@ pub fn inc_codec_error(error_type: &'static str, recovery_policy: &'static str) #[cfg(not(feature = "metrics"))] pub fn inc_codec_error(_error_type: &'static str, _recovery_policy: &'static str) {} +/// Record an application preparation outcome and its elapsed duration. +#[cfg(feature = "metrics")] +pub(crate) fn record_application_preparation(outcome: PreparationOutcome, elapsed: Duration) { + counter!(APPLICATION_PREPARATIONS, "outcome" => outcome.as_str()).increment(1); + histogram!(APPLICATION_PREPARATION_DURATION, "outcome" => outcome.as_str()) + .record(elapsed.as_secs_f64()); +} + +/// Record an application preparation outcome and its elapsed duration. +/// +/// This function is a no-op when the `metrics` feature is disabled. +#[cfg(not(feature = "metrics"))] +pub(crate) fn record_application_preparation(_outcome: PreparationOutcome, _elapsed: Duration) {} + +/// Record a connection handled by an immutable prepared application. +#[cfg(feature = "metrics")] +pub(crate) fn inc_prepared_connection_uses() { counter!(PREPARED_CONNECTION_USES).increment(1); } + +/// Record a connection handled by an immutable prepared application. +/// +/// This function is a no-op when the `metrics` feature is disabled. +#[cfg(not(feature = "metrics"))] +pub(crate) fn inc_prepared_connection_uses() {} + /// Record a server-supervisor cancellation request with a bounded reason. #[cfg(feature = "metrics")] pub(crate) fn inc_server_supervisor_cancellation(reason: ServerCancellationReason) { diff --git a/src/server/connection_spawner.rs b/src/server/connection_spawner.rs index c4f77def..cab15e28 100644 --- a/src/server/connection_spawner.rs +++ b/src/server/connection_spawner.rs @@ -100,7 +100,13 @@ where Envelope: DecodeWith + EncodeWith, { match factory.build() { - Ok(app) => { + Ok(app) => + { + #[expect( + deprecated, + reason = "the server retains per-connection factory evaluation until the runtime \ + slice" + )] if let Err(e) = app.handle_connection_result(stream).await { warn!("connection task error: {e:?}"); } diff --git a/src/testkit/fragment_drive.rs b/src/testkit/fragment_drive.rs index c8565eff..b4a9fe33 100644 --- a/src/testkit/fragment_drive.rs +++ b/src/testkit/fragment_drive.rs @@ -117,6 +117,10 @@ where /// /// Returns any I/O, fragmentation, or codec error encountered during /// encoding, transport, or decoding. +#[expect( + deprecated, + reason = "compatibility helper drives the legacy builder API" +)] pub async fn drive_with_fragments_mut( app: &mut WireframeApp, codec: &F, diff --git a/src/testkit/partial_frame.rs b/src/testkit/partial_frame.rs index a9d0e634..5ec636ee 100644 --- a/src/testkit/partial_frame.rs +++ b/src/testkit/partial_frame.rs @@ -76,6 +76,10 @@ where /// /// Returns any I/O or codec error encountered during encoding, transport, or /// decoding. +#[expect( + deprecated, + reason = "compatibility helper drives the legacy builder API" +)] pub async fn drive_with_partial_frames_mut( app: &mut WireframeApp, codec: &F, diff --git a/src/testkit/support.rs b/src/testkit/support.rs index 31dbd9d7..cfe02275 100644 --- a/src/testkit/support.rs +++ b/src/testkit/support.rs @@ -300,6 +300,10 @@ pub(crate) fn extract_payloads(frames: &[F::Frame]) -> Vec(app: WireframeApp, server: DuplexStream) where S: TestSerializer, diff --git a/tests/common/fragment_helpers/app.rs b/tests/common/fragment_helpers/app.rs index 4801cf07..05dec85e 100644 --- a/tests/common/fragment_helpers/app.rs +++ b/tests/common/fragment_helpers/app.rs @@ -50,6 +50,7 @@ pub fn make_app( } /// Spawn an app and return the client connection and server task handle. +#[expect(deprecated, reason = "fragment helper drives the legacy builder API")] pub fn spawn_app( app: WireframeApp, ) -> ( diff --git a/tests/compile_error.rs b/tests/compile_error.rs index 8adb8de1..16d31ff1 100644 --- a/tests/compile_error.rs +++ b/tests/compile_error.rs @@ -5,4 +5,5 @@ fn compile_tests() { let t = trybuild::TestCases::new(); t.pass("tests/ui/wireframe_result_default_no_protocol.rs"); t.compile_fail("tests/ui/wireframe_result_default_rejects_unit_protocol.rs"); + t.compile_fail("tests/ui/prepared_app_rejects_route.rs"); } diff --git a/tests/example_codecs.rs b/tests/example_codecs.rs index 3df1261b..ef1b21dc 100644 --- a/tests/example_codecs.rs +++ b/tests/example_codecs.rs @@ -101,6 +101,7 @@ fn mysql_codec_rejects_oversized_payload() { } #[tokio::test] +#[expect(deprecated, reason = "test covers the legacy builder connection API")] async fn hotline_codec_round_trips_through_app() { let codec = HotlineFrameCodec::new(64); let app = WireframeApp::::new() diff --git a/tests/fixtures/budget_cleanup.rs b/tests/fixtures/budget_cleanup.rs index 349da446..16b36764 100644 --- a/tests/fixtures/budget_cleanup.rs +++ b/tests/fixtures/budget_cleanup.rs @@ -132,6 +132,10 @@ impl BudgetCleanupWorld { } /// Start the app under test using the supplied budget and timeout config. + #[expect( + deprecated, + reason = "fixture covers the legacy builder connection API" + )] pub fn start_app(&mut self, config: CleanupConfig) -> TestResult { let Some(fragment_limit) = NonZeroUsize::new(BUFFER_CAPACITY.saturating_mul(16)) else { return Err("buffer-derived fragment limit should be non-zero".into()); diff --git a/tests/fixtures/budget_transitions.rs b/tests/fixtures/budget_transitions.rs index 16fea6f5..474753b2 100644 --- a/tests/fixtures/budget_transitions.rs +++ b/tests/fixtures/budget_transitions.rs @@ -133,6 +133,10 @@ impl BudgetTransitionsWorld { } /// Start the app under test using the supplied budget and timeout config. + #[expect( + deprecated, + reason = "fixture covers the legacy builder connection API" + )] pub fn start_app(&mut self, config: TransitionConfig) -> TestResult { let Some(fragment_limit) = NonZeroUsize::new(BUFFER_CAPACITY.saturating_mul(16)) else { return Err("buffer-derived fragment limit should be non-zero".into()); diff --git a/tests/fixtures/codec_stateful.rs b/tests/fixtures/codec_stateful.rs index 7f312db9..fa52e6b2 100644 --- a/tests/fixtures/codec_stateful.rs +++ b/tests/fixtures/codec_stateful.rs @@ -154,6 +154,10 @@ struct StatefulServer { handle: JoinHandle<()>, } +#[expect( + deprecated, + reason = "fixture covers the legacy builder connection API" +)] async fn serve_stateful_connections( listener: TcpListener, app: WireframeApp, diff --git a/tests/fixtures/derived_memory_budgets.rs b/tests/fixtures/derived_memory_budgets.rs index 0145464d..0f579e74 100644 --- a/tests/fixtures/derived_memory_budgets.rs +++ b/tests/fixtures/derived_memory_budgets.rs @@ -225,6 +225,10 @@ impl DerivedMemoryBudgetsWorld { self.start_with_app(app, rx) } + #[expect( + deprecated, + reason = "fixture covers the legacy builder connection API" + )] fn start_with_app( &mut self, app: WireframeApp, diff --git a/tests/fixtures/memory_budget_backpressure.rs b/tests/fixtures/memory_budget_backpressure.rs index aea7e1a4..cf475413 100644 --- a/tests/fixtures/memory_budget_backpressure.rs +++ b/tests/fixtures/memory_budget_backpressure.rs @@ -126,6 +126,10 @@ impl MemoryBudgetBackpressureWorld { } /// Start the app under test using the supplied budget and timeout config. + #[expect( + deprecated, + reason = "fixture covers the legacy builder connection API" + )] pub fn start_app(&mut self, config: BackpressureConfig) -> TestResult { let Some(fragment_limit) = NonZeroUsize::new(BUFFER_CAPACITY.saturating_mul(16)) else { return Err("buffer-derived fragment limit should be non-zero".into()); diff --git a/tests/fixtures/memory_budget_hard_cap.rs b/tests/fixtures/memory_budget_hard_cap.rs index 7237400e..47d1c950 100644 --- a/tests/fixtures/memory_budget_hard_cap.rs +++ b/tests/fixtures/memory_budget_hard_cap.rs @@ -129,6 +129,10 @@ impl MemoryBudgetHardCapWorld { } /// Start the app under test using the supplied budget and timeout config. + #[expect( + deprecated, + reason = "fixture covers the legacy builder connection API" + )] pub fn start_app(&mut self, config: HardCapConfig) -> TestResult { let Some(fragment_limit) = NonZeroUsize::new(BUFFER_CAPACITY.saturating_mul(16)) else { return Err("buffer-derived fragment limit should be non-zero".into()); diff --git a/tests/fixtures/message_assembly_inbound.rs b/tests/fixtures/message_assembly_inbound.rs index 9f4aac68..734cdf19 100644 --- a/tests/fixtures/message_assembly_inbound.rs +++ b/tests/fixtures/message_assembly_inbound.rs @@ -115,6 +115,10 @@ impl MessageAssemblyInboundWorld { /// /// Returns an error if the fragmentation config, app builder, or runtime /// initialization fails. + #[expect( + deprecated, + reason = "fixture covers the legacy builder connection API" + )] pub fn start_app(&mut self, timeout_ms: u64) -> TestResult { let message_limit = NonZeroUsize::new(BUFFER_CAPACITY.saturating_mul(16)).unwrap_or(NonZeroUsize::MIN); diff --git a/tests/fixtures/unified_codec/mod.rs b/tests/fixtures/unified_codec/mod.rs index 38b2c589..dbc0cb46 100644 --- a/tests/fixtures/unified_codec/mod.rs +++ b/tests/fixtures/unified_codec/mod.rs @@ -70,6 +70,10 @@ impl UnifiedCodecWorld { /// /// # Errors /// Returns an error if app creation or spawning fails. + #[expect( + deprecated, + reason = "fixture covers the legacy builder connection API" + )] pub fn start_server( &mut self, runtime: &Runtime, diff --git a/tests/frame_codec.rs b/tests/frame_codec.rs index d12e0cce..99c1ced3 100644 --- a/tests/frame_codec.rs +++ b/tests/frame_codec.rs @@ -11,7 +11,7 @@ use std::{ use bytes::{Buf, BufMut, Bytes, BytesMut}; use futures::{SinkExt, StreamExt}; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::AsyncWriteExt; use tokio_util::codec::{Decoder, Encoder, Framed}; use wireframe::{ app::{Envelope, Packet, WireframeApp}, @@ -19,6 +19,7 @@ use wireframe::{ correlation::CorrelatableFrame, serializer::{BincodeSerializer, Serializer}, }; +use wireframe_testing::drive_prepared_with_frames; #[derive(Clone, Debug)] struct TaggedFrame { @@ -144,13 +145,6 @@ async fn custom_codec_round_trips_frames() { .route(1, Arc::new(|_: &Envelope| Box::pin(async {}))) .expect("route configured"); - let (mut client, server) = tokio::io::duplex(256); - let server_task = tokio::spawn(async move { - app.handle_connection_result(server) - .await - .expect("server should exit cleanly"); - }); - let request = Envelope::new(1, None, b"ping".to_vec()); let payload = BincodeSerializer .serialize(&request) @@ -162,16 +156,10 @@ async fn custom_codec_round_trips_frames() { .encode(TaggedFrame { tag: 7, payload }, &mut buf) .expect("encode request"); - client.write_all(&buf).await.expect("write request"); - client.shutdown().await.expect("shutdown client"); - - let mut output = Vec::new(); - client - .read_to_end(&mut output) + let prepared = app.prepare().await.expect("prepare app"); + let output = drive_prepared_with_frames(&prepared, vec![buf.to_vec()]) .await - .expect("read response"); - - server_task.await.expect("join server task"); + .expect("drive prepared app"); let mut decoder = TaggedAdapter::new(64); let mut response_buf = BytesMut::from(&output[..]); @@ -190,6 +178,7 @@ async fn custom_codec_round_trips_frames() { } #[tokio::test] +#[expect(deprecated, reason = "test covers the legacy builder connection API")] async fn stateful_codec_advances_tags_per_connection() { let app = WireframeApp::::new() .expect("build app") diff --git a/tests/middleware_order.rs b/tests/middleware_order.rs index 43b3ad78..cf3619bd 100644 --- a/tests/middleware_order.rs +++ b/tests/middleware_order.rs @@ -57,6 +57,7 @@ impl Transform> for TagMiddleware { clippy::panic_in_result_fn, reason = "asserts provide clearer diagnostics in tests" )] +#[expect(deprecated, reason = "test covers the legacy builder connection API")] async fn middleware_applied_in_reverse_order() -> TestResult<()> { let handler: Handler = std::sync::Arc::new(|_env: &Envelope| Box::pin(async {})); let app = TestApp::new() diff --git a/tests/prepared_app.rs b/tests/prepared_app.rs new file mode 100644 index 00000000..aa0dcb6c --- /dev/null +++ b/tests/prepared_app.rs @@ -0,0 +1,449 @@ +//! Integration coverage for one-time application preparation. + +use std::{ + convert::Infallible, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; + +use async_trait::async_trait; +use proptest::{ + prelude::*, + test_runner::{TestCaseError, TestCaseResult}, +}; +use tokio::{ + io::AsyncWriteExt, + net::TcpStream, + runtime::Builder, + sync::{Barrier, oneshot}, + time::{sleep, timeout}, +}; +use wireframe::{ + app::{Envelope, Handler, PreparedApp, WireframeApp}, + middleware::{HandlerService, Service, ServiceRequest, ServiceResponse, Transform}, + serializer::{BincodeSerializer, Serializer}, + server::WireframeServer, +}; +use wireframe_testing::{ + TestResult, + decode_frames, + drive_prepared_with_frames, + encode_frame, + unused_listener, + wait_for_listener_release, + wait_for_server_readiness, +}; + +type TestApp = WireframeApp; +type TestPreparedApp = PreparedApp; + +const ROUTES: usize = 2; +const MIDDLEWARE_LAYERS: usize = 2; +const CONNECTIONS: usize = 2; + +/// Counter snapshots for the application connection-startup baseline. +#[derive(Clone)] +struct ConnectionStartupInstrumentation { + factory_calls: Arc, + transforms: Arc, +} + +impl ConnectionStartupInstrumentation { + /// Creates counters for factory invocations and middleware transforms. + fn new() -> Self { + Self { + factory_calls: Arc::new(AtomicUsize::new(0)), + transforms: Arc::new(AtomicUsize::new(0)), + } + } + + /// Returns the current connection-startup counter values. + fn snapshot(&self) -> ConnectionStartupCounts { + ConnectionStartupCounts { + factory_calls: self.factory_calls.load(Ordering::SeqCst), + transforms: self.transforms.load(Ordering::SeqCst), + } + } +} + +#[derive(Debug, PartialEq, Eq)] +struct ConnectionStartupCounts { + factory_calls: usize, + transforms: usize, +} + +struct TransformCountingMiddleware { + tag: u8, + transforms: Arc, +} + +struct TagService { + inner: S, + tag: u8, +} + +#[async_trait] +impl Service for TagService +where + S: Service + Send + Sync + 'static, +{ + type Error = Infallible; + + /// Adds this service's tag around the delegated request and response. + async fn call(&self, mut request: ServiceRequest) -> Result { + request.frame_mut().push(self.tag); + let mut response = self.inner.call(request).await?; + response.frame_mut().push(self.tag); + Ok(response) + } +} + +#[async_trait] +impl Transform> for TransformCountingMiddleware { + type Output = HandlerService; + + /// Counts this transformation and wraps the route service with its tag. + async fn transform(&self, service: HandlerService) -> Self::Output { + self.transforms.fetch_add(1, Ordering::SeqCst); + let id = service.id(); + HandlerService::from_service( + id, + TagService { + inner: service, + tag: self.tag, + }, + ) + } +} + +/// Builds a handler that accepts an envelope without changing it. +fn handler() -> Handler { Arc::new(|_envelope: &Envelope| Box::pin(async {})) } + +/// Creates the test factory used to compare legacy and prepared startup work. +fn counted_app_factory( + instrumentation: ConnectionStartupInstrumentation, +) -> impl Fn() -> TestResult + Clone + Send + Sync + 'static { + move || { + instrumentation.factory_calls.fetch_add(1, Ordering::SeqCst); + Ok(TestApp::new()? + .route(1, handler())? + .route(2, handler())? + .wrap(TransformCountingMiddleware { + tag: b'A', + transforms: Arc::clone(&instrumentation.transforms), + })? + .wrap(TransformCountingMiddleware { + tag: b'B', + transforms: Arc::clone(&instrumentation.transforms), + })?) + } +} + +/// Waits until connection-startup counters reach the expected values. +async fn wait_for_counts( + instrumentation: &ConnectionStartupInstrumentation, + expected: &ConnectionStartupCounts, +) -> TestResult<()> { + timeout(Duration::from_secs(1), async { + while instrumentation.snapshot() != *expected { + sleep(Duration::from_millis(10)).await; + } + }) + .await + .map_err(|_| { + format!( + "connection startup counts did not reach {expected:?}; observed {:?}", + instrumentation.snapshot() + ) + })?; + Ok(()) +} + +/// Runs legacy server connections and waits for their startup instrumentation. +async fn run_legacy_server_connections( + app_factory: impl Fn() -> TestResult + Clone + Send + Sync + 'static, + instrumentation: &ConnectionStartupInstrumentation, + expected: &ConnectionStartupCounts, +) -> TestResult<()> { + let server = WireframeServer::new(app_factory) + .workers(1) + .bind_existing_listener(unused_listener()?)?; + let address = server + .local_addr() + .ok_or_else(|| "server did not report a bound address".to_string())?; + let (ready_tx, ready_rx) = oneshot::channel(); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let server_task = tokio::spawn(async move { + server + .ready_signal(ready_tx) + .run_with_shutdown(async { + let _ = shutdown_rx.await; + }) + .await + }); + + wait_for_server_readiness(ready_rx).await?; + let frame = build_frame(1, Vec::new())?; + let mut connections = Vec::with_capacity(CONNECTIONS); + for _ in 0..CONNECTIONS { + let mut connection = TcpStream::connect(address).await?; + connection.write_all(&frame).await?; + connections.push(connection); + } + let counts_result = wait_for_counts(instrumentation, expected).await; + drop(connections); + let shutdown_result = shutdown_tx + .send(()) + .map_err(|()| "server shutdown receiver was dropped"); + let server_result = server_task.await; + let listener_result = wait_for_listener_release(address).await; + + counts_result?; + shutdown_result?; + server_result??; + listener_result +} + +/// Encodes an envelope into a frame for an in-process connection. +fn build_frame(id: u32, payload: Vec) -> TestResult> { + let serializer = BincodeSerializer; + let envelope = Envelope::new(id, Some(7), payload); + let payload = serializer.serialize(&envelope)?; + let mut codec = TestApp::default().length_codec(); + Ok(encode_frame(&mut codec, payload)?) +} + +/// Decodes one response frame and returns its envelope payload. +fn response_payload(bytes: Vec) -> TestResult> { + let frames = decode_frames(bytes)?; + let [frame] = frames.as_slice() else { + return Err("expected one response frame".into()); + }; + let serializer = BincodeSerializer; + let (response, _) = serializer.deserialize::(frame)?; + Ok(wireframe::app::Packet::into_parts(response).into_payload()) +} + +#[tokio::test] +#[expect( + clippy::panic_in_result_fn, + reason = "assertions make transform counts and middleware order failures explicit" +)] +async fn connection_startup_records_counts_before_and_after_preparation() -> TestResult<()> { + let instrumentation = ConnectionStartupInstrumentation::new(); + let app_factory = counted_app_factory(instrumentation.clone()); + + assert_eq!( + instrumentation.snapshot(), + ConnectionStartupCounts { + factory_calls: 0, + transforms: 0, + } + ); + + let legacy_counts = ConnectionStartupCounts { + factory_calls: CONNECTIONS, + transforms: CONNECTIONS * ROUTES * MIDDLEWARE_LAYERS, + }; + run_legacy_server_connections(app_factory.clone(), &instrumentation, &legacy_counts).await?; + assert_eq!(instrumentation.snapshot(), legacy_counts); + + let prepared: TestPreparedApp = app_factory()? + .prepare() + .await + .map_err(|error| -> Box { Box::new(error) })?; + let prepared_counts = ConnectionStartupCounts { + factory_calls: CONNECTIONS + 1, + transforms: (CONNECTIONS + 1) * ROUTES * MIDDLEWARE_LAYERS, + }; + assert_eq!(instrumentation.snapshot(), prepared_counts); + + let first = drive_prepared_with_frames(&prepared, vec![build_frame(1, vec![b'X'])?]).await?; + let second = drive_prepared_with_frames(&prepared, vec![build_frame(2, vec![b'Y'])?]).await?; + + assert_eq!(instrumentation.snapshot(), prepared_counts); + assert_eq!(response_payload(first)?, [b'X', b'A', b'B', b'B', b'A']); + assert_eq!(response_payload(second)?, [b'Y', b'A', b'B', b'B', b'A']); + Ok(()) +} + +#[tokio::test] +#[expect( + clippy::panic_in_result_fn, + reason = "assertions make prepared-connection failure behaviour explicit" +)] +async fn prepared_app_runs_teardown_after_processing_error() -> TestResult<()> { + let teardown_calls = Arc::new(AtomicUsize::new(0)); + let teardown_counter = Arc::clone(&teardown_calls); + let prepared = TestApp::new()? + .on_connection_setup(|| async {})? + .on_connection_teardown(move |()| { + let teardown_counter = Arc::clone(&teardown_counter); + async move { + teardown_counter.fetch_add(1, Ordering::SeqCst); + } + })? + .prepare() + .await + .map_err(|error| -> Box { Box::new(error) })?; + + let error = drive_prepared_with_frames(&prepared, vec![vec![0, 0, 0, 2, 1]]) + .await + .expect_err("truncated frame should fail processing"); + assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof); + assert_eq!(teardown_calls.load(Ordering::SeqCst), 1); + + let (mut client, server) = tokio::io::duplex(64); + client.write_all(&[0, 0, 0, 2, 1]).await?; + client.shutdown().await?; + prepared.handle_connection(server).await; + assert_eq!(teardown_calls.load(Ordering::SeqCst), 2); + Ok(()) +} + +#[tokio::test] +#[expect( + clippy::panic_in_result_fn, + reason = "assertions make concurrent prepared-service reuse explicit" +)] +async fn prepared_app_reuses_services_across_overlapping_connections() -> TestResult<()> { + let transforms = Arc::new(AtomicUsize::new(0)); + let barrier = Arc::new(Barrier::new(CONNECTIONS)); + let handler_barrier = Arc::clone(&barrier); + let handler: Handler = Arc::new(move |_: &Envelope| { + let barrier = Arc::clone(&handler_barrier); + Box::pin(async move { + barrier.wait().await; + }) + }); + let prepared = TestApp::new()? + .route(1, handler)? + .wrap(TransformCountingMiddleware { + tag: b'A', + transforms: Arc::clone(&transforms), + })? + .prepare() + .await + .map_err(|error| -> Box { Box::new(error) })?; + assert_eq!(transforms.load(Ordering::SeqCst), 1); + + let first_frame = build_frame(1, vec![b'X'])?; + let second_frame = build_frame(1, vec![b'Y'])?; + let (first, second) = timeout(Duration::from_secs(1), async { + tokio::join!( + drive_prepared_with_frames(&prepared, vec![first_frame]), + drive_prepared_with_frames(&prepared, vec![second_frame]), + ) + }) + .await + .map_err(|_| "prepared connections did not overlap")?; + assert_eq!(response_payload(first?)?, [b'X', b'A', b'A']); + assert_eq!(response_payload(second?)?, [b'Y', b'A', b'A']); + assert_eq!(transforms.load(Ordering::SeqCst), 1); + Ok(()) +} + +// Generate bounded prepared-application cases and preserve one-time transforms. +proptest! { + #![proptest_config(ProptestConfig { + cases: 32, + .. ProptestConfig::default() + })] + + #[test] + fn prepared_app_transforms_once_and_reuses_services( + route_count in 1usize..=4, + middleware_layers in 0usize..=4, + connection_count in 1usize..=4, + ) { + run_prepared_app_property_case(route_count, middleware_layers, connection_count)?; + } +} + +/// Exercise a generated preparation case on a deterministic Tokio runtime. +fn run_prepared_app_property_case( + route_count: usize, + middleware_layers: usize, + connection_count: usize, +) -> TestCaseResult { + let runtime = Builder::new_current_thread() + .enable_all() + .build() + .map_err(|error| TestCaseError::fail(error.to_string()))?; + runtime + .block_on(exercise_prepared_app_property_case( + route_count, + middleware_layers, + connection_count, + )) + .map_err(|error| TestCaseError::fail(error.to_string())) +} + +/// Prepare a bounded generated application and verify every requested dispatch. +async fn exercise_prepared_app_property_case( + route_count: usize, + middleware_layers: usize, + connection_count: usize, +) -> TestResult<()> { + let transforms = Arc::new(AtomicUsize::new(0)); + let tags = middleware_tags(middleware_layers)?; + let mut app = TestApp::new()?; + for route_id in 1..=route_count { + app = app.route(u32::try_from(route_id)?, handler())?; + } + for tag in &tags { + app = app.wrap(TransformCountingMiddleware { + tag: *tag, + transforms: Arc::clone(&transforms), + })?; + } + + let prepared = app + .prepare() + .await + .map_err(|error| -> Box { Box::new(error) })?; + let expected_transforms = route_count * middleware_layers; + if transforms.load(Ordering::SeqCst) != expected_transforms { + return Err(format!( + "preparation transformed {} route services, expected {expected_transforms}", + transforms.load(Ordering::SeqCst) + ) + .into()); + } + + for route_id in (1..=route_count) + .cycle() + .take(route_count + connection_count) + { + let route_id = u32::try_from(route_id)?; + let payload = vec![u8::try_from(route_id)?]; + let response = + drive_prepared_with_frames(&prepared, vec![build_frame(route_id, payload)?]).await?; + let mut expected = vec![u8::try_from(route_id)?]; + expected.extend(tags.iter().copied()); + expected.extend(tags.iter().rev().copied()); + if response_payload(response)? != expected { + return Err( + format!("route {route_id} did not preserve generated middleware order").into(), + ); + } + } + if transforms.load(Ordering::SeqCst) != expected_transforms { + return Err(format!( + "prepared connections rebuilt middleware: observed {}, expected {expected_transforms}", + transforms.load(Ordering::SeqCst) + ) + .into()); + } + Ok(()) +} + +/// Build distinct middleware tags for a bounded generated layer count. +fn middleware_tags(layer_count: usize) -> TestResult> { + (0..layer_count) + .map(|layer| Ok(b'A' + u8::try_from(layer)?)) + .collect() +} diff --git a/tests/prepared_app_observability.rs b/tests/prepared_app_observability.rs new file mode 100644 index 00000000..ff19752c --- /dev/null +++ b/tests/prepared_app_observability.rs @@ -0,0 +1,49 @@ +//! Observability coverage for prepared application transitions. +#![cfg(feature = "metrics")] + +use metrics::with_local_recorder; +use tokio::runtime::Builder; +use wireframe::{ + app::{Envelope, WireframeApp}, + metrics::{ + APPLICATION_PREPARATION_DURATION, + APPLICATION_PREPARATIONS, + PREPARED_CONNECTION_USES, + }, + serializer::BincodeSerializer, +}; +use wireframe_testing::{ObservabilityHandle, TestResult, drive_prepared_with_frames}; + +/// Verify preparation and prepared-connection metrics use bounded series. +#[test] +fn prepared_application_metrics_record_outcome_duration_and_use() -> TestResult<()> { + let mut observability = ObservabilityHandle::new(); + let runtime = Builder::new_current_thread().enable_all().build()?; + with_local_recorder(observability.recorder(), || { + runtime.block_on(async { + let app: WireframeApp = WireframeApp::new()?; + let prepared = app + .prepare() + .await + .map_err(|error| Box::new(error) as Box)?; + drive_prepared_with_frames(&prepared, Vec::new()).await?; + drive_prepared_with_frames(&prepared, Vec::new()).await?; + Ok::<(), wireframe_testing::TestError>(()) + }) + })?; + + observability.snapshot(); + observability + .assert_counter(APPLICATION_PREPARATIONS, [("outcome", "success")], 1) + .map_err(|error| format!("preparation success metric missing: {error}"))?; + observability + .assert_counter(APPLICATION_PREPARATIONS, [("outcome", "failure")], 0) + .map_err(|error| format!("preparation failure series should remain absent: {error}"))?; + observability + .assert_counter(PREPARED_CONNECTION_USES, [], 2) + .map_err(|error| format!("prepared-connection use metric missing: {error}"))?; + observability + .assert_histogram_recorded(APPLICATION_PREPARATION_DURATION, [("outcome", "success")]) + .map_err(|error| format!("preparation duration metric missing: {error}"))?; + Ok(()) +} diff --git a/tests/prepared_app_tcp.rs b/tests/prepared_app_tcp.rs new file mode 100644 index 00000000..61b38d9e --- /dev/null +++ b/tests/prepared_app_tcp.rs @@ -0,0 +1,120 @@ +//! End-to-end TCP coverage for immutable prepared applications. + +use std::{ + convert::Infallible, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, +}; + +use async_trait::async_trait; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{TcpListener, TcpStream}, +}; +use wireframe::{ + app::{Envelope, Handler, WireframeApp}, + middleware::{HandlerService, Service, ServiceRequest, ServiceResponse, Transform}, + serializer::{BincodeSerializer, Serializer}, +}; +use wireframe_testing::{TestResult, decode_frames, encode_frame}; + +type TestApp = WireframeApp; + +/// Middleware that exposes transform and request-response execution counts. +struct TransformCountingMiddleware { + transforms: Arc, +} + +/// Service that tags requests and responses around its delegate. +struct TagService { + inner: S, +} + +#[async_trait] +impl Service for TagService +where + S: Service + Send + Sync + 'static, +{ + type Error = Infallible; + + /// Tag both sides of the delegated request-response exchange. + async fn call(&self, mut request: ServiceRequest) -> Result { + request.frame_mut().push(b'A'); + let mut response = self.inner.call(request).await?; + response.frame_mut().push(b'A'); + Ok(response) + } +} + +#[async_trait] +impl Transform> for TransformCountingMiddleware { + type Output = HandlerService; + + /// Count transformation and wrap the route service once. + async fn transform(&self, service: HandlerService) -> Self::Output { + self.transforms.fetch_add(1, Ordering::SeqCst); + HandlerService::from_service(service.id(), TagService { inner: service }) + } +} + +/// Build a handler that accepts an envelope without changing it. +fn handler() -> Handler { Arc::new(|_envelope| Box::pin(async {})) } + +/// Encode an envelope into the default transport frame. +fn build_frame(payload: Vec) -> TestResult> { + let serializer = BincodeSerializer; + let envelope = Envelope::new(1, Some(7), payload); + let payload = serializer.serialize(&envelope)?; + let mut codec = TestApp::default().length_codec(); + Ok(encode_frame(&mut codec, payload)?) +} + +/// Decode the response envelope and return its payload. +fn response_payload(bytes: Vec) -> TestResult> { + let frames = decode_frames(bytes)?; + let [frame] = frames.as_slice() else { + return Err("expected one response frame".into()); + }; + let serializer = BincodeSerializer; + let (response, _) = serializer.deserialize::(frame)?; + Ok(wireframe::app::Packet::into_parts(response).into_payload()) +} + +/// Prepared applications serve TCP connections without rebuilding middleware. +#[tokio::test] +#[expect( + clippy::panic_in_result_fn, + reason = "assertions make prepared TCP dispatch and transform reuse explicit" +)] +async fn prepared_app_serves_tcp_connection_without_retransforming() -> TestResult<()> { + let transforms = Arc::new(AtomicUsize::new(0)); + let prepared = TestApp::new()? + .route(1, handler())? + .wrap(TransformCountingMiddleware { + transforms: Arc::clone(&transforms), + })? + .prepare() + .await + .map_err(|error| -> Box { Box::new(error) })?; + assert_eq!(transforms.load(Ordering::SeqCst), 1); + + let listener = TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await?; + prepared.handle_connection_result(stream).await + }); + + let mut client = TcpStream::connect(address).await?; + client.write_all(&build_frame(vec![b'X'])?).await?; + client.shutdown().await?; + let mut response = Vec::new(); + client.read_to_end(&mut response).await?; + server.await??; + + assert_eq!(response_payload(response)?, [b'X', b'A', b'A']); + assert_eq!(transforms.load(Ordering::SeqCst), 1); + Ok(()) +} diff --git a/tests/ui/prepared_app_rejects_route.rs b/tests/ui/prepared_app_rejects_route.rs new file mode 100644 index 00000000..224b1b09 --- /dev/null +++ b/tests/ui/prepared_app_rejects_route.rs @@ -0,0 +1,16 @@ +//! Compile-fail coverage for the `PreparedApp` route-registration boundary. +use wireframe::{ + app::{Envelope, Handler, WireframeApp}, + serializer::BincodeSerializer, +}; + +#[tokio::main] +async fn main() { + let handler: Handler = std::sync::Arc::new(|_| Box::pin(async {})); + let prepared = WireframeApp::::new() + .expect("builder") + .prepare() + .await + .expect("prepared"); + let _ = prepared.route(1, handler); +} diff --git a/tests/ui/prepared_app_rejects_route.stderr b/tests/ui/prepared_app_rejects_route.stderr new file mode 100644 index 00000000..10759a78 --- /dev/null +++ b/tests/ui/prepared_app_rejects_route.stderr @@ -0,0 +1,5 @@ +error[E0599]: no method named `route` found for struct `PreparedApp` in the current scope + --> tests/ui/prepared_app_rejects_route.rs:15:22 + | +15 | let _ = prepared.route(1, handler); + | ^^^^^ method not found in `PreparedApp` diff --git a/tests/wireframe_protocol.rs b/tests/wireframe_protocol.rs index 5414baa5..97d88e3a 100644 --- a/tests/wireframe_protocol.rs +++ b/tests/wireframe_protocol.rs @@ -269,3 +269,36 @@ fn message_assembler_accessor_reflects_installation() { "installed assembler should be visible" ); } + +#[rstest] +#[tokio::test] +async fn prepared_app_retains_runtime_protocol_accessors(queues: QueueResult) -> TestResult<()> { + let counter = Arc::new(AtomicUsize::new(0)); + let prepared = TestApp::new()? + .with_protocol(TestProtocol { + counter: Arc::clone(&counter), + }) + .with_message_assembler(DemoAssembler) + .prepare() + .await + .map_err(|error| -> Box { Box::new(error) })?; + + assert!( + prepared.protocol().is_some(), + "prepared protocol should be visible" + ); + assert!( + prepared.message_assembler().is_some(), + "prepared assembler should be visible" + ); + + let (_queues, handle) = queues?; + let mut hooks = prepared.protocol_hooks(); + hooks.on_connection_setup(handle, &mut ConnectionContext); + assert_eq!( + counter.load(Ordering::SeqCst), + 1, + "prepared hooks should run" + ); + Ok(()) +} diff --git a/tests/workflow_contracts/coverage_main_workflow_test.py b/tests/workflow_contracts/coverage_main_workflow_test.py new file mode 100644 index 00000000..767d9c5b --- /dev/null +++ b/tests/workflow_contracts/coverage_main_workflow_test.py @@ -0,0 +1,97 @@ +"""Protect CodeScene's default-branch coverage baseline workflow. + +Run these workflow contract tests with ``make test-workflow-contracts``. +""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import cast + +import yaml + +WORKFLOW_PATH: Path = ( + Path(__file__).resolve().parents[2] / ".github" / "workflows" / "coverage-main.yml" +) +CODESCENE_USES_RE: re.Pattern[str] = re.compile( + r"^leynos/shared-actions/\.github/actions/upload-codescene-coverage@" + r"[0-9a-f]{40}$" +) + + +def _load_steps() -> list[dict[str, object]]: + """Parse and return the default-branch coverage-upload steps.""" + workflow = yaml.safe_load(WORKFLOW_PATH.read_text(encoding="utf-8")) + assert isinstance(workflow, dict), "the coverage workflow must be a mapping" + jobs = workflow.get("jobs") + assert isinstance(jobs, dict), "the coverage workflow must declare jobs" + coverage_upload = jobs.get("coverage-upload") + assert isinstance(coverage_upload, dict), ( + "the coverage workflow must declare coverage-upload" + ) + steps = coverage_upload.get("steps") + assert isinstance(steps, list), "the coverage-upload job must declare steps" + assert all(isinstance(step, dict) for step in steps), ( + "every coverage-upload step must be a mapping" + ) + return cast("list[dict[str, object]]", steps) + + +def _find_step(steps: list[dict[str, object]], name: str) -> dict[str, object]: + """Return the uniquely named default-branch coverage workflow step.""" + matches = [step for step in steps if step.get("name") == name] + assert len(matches) == 1, f"expected one {name!r} step, found {len(matches)}" + return matches[0] + + +def test_codescene_upload_follows_successful_coverage_generation() -> None: + """Upload the newly generated LCOV report before PR gates can use its baseline.""" + steps = _load_steps() + generation = _find_step(steps, "Test and Measure Coverage") + upload = _find_step(steps, "Upload coverage data to CodeScene") + assert steps.index(upload) == steps.index(generation) + 1, ( + "the CodeScene upload must immediately follow coverage generation" + ) + assert generation.get("with") == { + "output-path": "lcov.info", + "format": "lcov", + "with-ratchet": "true", + }, "main must generate the ratcheted LCOV report before uploading it" + + +def test_codescene_upload_uses_wireframe_project_and_repository() -> None: + """Upload main coverage to the project and repository used by PR checks.""" + steps = _load_steps() + checkout = steps[0] + assert checkout.get("uses") == ( + "actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1" + ), ( + "the coverage workflow must start from Wireframe's checkout" + ) + assert checkout.get("with") == { + "repository": "leynos/wireframe", + "persist-credentials": False, + }, ( + "the checkout origin must identify github.com/leynos/wireframe without " + "persisting credentials" + ) + + upload = _find_step(steps, "Upload coverage data to CodeScene") + assert upload.get("env") == {"CS_ACCESS_TOKEN": "${{ secrets.CS_ACCESS_TOKEN }}"}, ( + "the CodeScene token must remain scoped to the upload step" + ) + assert upload.get("if") == "env.CS_ACCESS_TOKEN != ''", ( + "the upload must remain safe for contexts without the CodeScene secret" + ) + uses = upload.get("uses") + assert isinstance(uses, str) and CODESCENE_USES_RE.fullmatch(uses), ( + "the upload must invoke upload-codescene-coverage at a full commit SHA" + ) + assert upload.get("with") == { + "format": "lcov", + "mode": "upload", + "project-url": "https://api.codescene.io/v2/projects/68308", + "access-token": "${{ env.CS_ACCESS_TOKEN }}", + "installer-checksum": "${{ vars.CODESCENE_CLI_SHA256 }}", + }, "the upload must target the project used by the pull-request gate" diff --git a/wireframe_testing/src/helpers.rs b/wireframe_testing/src/helpers.rs index eb80c056..52303d71 100644 --- a/wireframe_testing/src/helpers.rs +++ b/wireframe_testing/src/helpers.rs @@ -74,6 +74,7 @@ pub use codec_fixtures::{ valid_hotline_wire, }; pub use drive::{ + drive_prepared_with_frames, drive_with_frame, drive_with_frame_mut, drive_with_frame_with_capacity, @@ -82,6 +83,7 @@ pub use drive::{ drive_with_frames_mut, drive_with_frames_with_capacity, drive_with_frames_with_capacity_mut, + prepare_and_drive_with_frames, }; pub use payloads::{drive_with_bincode, drive_with_payloads, drive_with_payloads_mut}; pub use runtime::{run_app, run_with_duplex_server}; diff --git a/wireframe_testing/src/helpers/codec_drive.rs b/wireframe_testing/src/helpers/codec_drive.rs index ba1386eb..13a17304 100644 --- a/wireframe_testing/src/helpers/codec_drive.rs +++ b/wireframe_testing/src/helpers/codec_drive.rs @@ -38,7 +38,7 @@ async fn drive_codec_frames_internal( where F: FrameCodec, H: FnOnce(DuplexStream) -> Fut, - Fut: std::future::Future + Send, + Fut: std::future::Future> + Send, { let encoded = encode_payloads_with_codec(codec, payloads)?; let raw = drive_internal(handler, encoded, capacity).await?; @@ -176,6 +176,10 @@ where /// # Ok(()) /// # } /// ``` +#[expect( + deprecated, + reason = "compatibility helper drives the legacy builder API" +)] pub async fn drive_with_codec_payloads_with_capacity_mut( app: &mut WireframeApp, codec: &F, @@ -189,7 +193,7 @@ where F: FrameCodec, { let frames = drive_codec_frames_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, codec, payloads, capacity, @@ -257,6 +261,10 @@ where /// # Ok(()) /// # } /// ``` +#[expect( + deprecated, + reason = "compatibility helper drives the legacy builder API" +)] pub async fn drive_with_codec_frames_with_capacity( app: WireframeApp, codec: &F, @@ -270,7 +278,7 @@ where F: FrameCodec, { drive_codec_frames_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, codec, payloads, capacity, diff --git a/wireframe_testing/src/helpers/drive.rs b/wireframe_testing/src/helpers/drive.rs index fe58564a..b36c8d1a 100644 --- a/wireframe_testing/src/helpers/drive.rs +++ b/wireframe_testing/src/helpers/drive.rs @@ -3,7 +3,10 @@ use std::io; use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream, duplex}; -use wireframe::app::{Packet, WireframeApp}; +use wireframe::{ + app::{Packet, PreparedApp, WireframeApp}, + codec::FrameCodec, +}; use super::{DEFAULT_CAPACITY, TestSerializer}; @@ -13,14 +16,17 @@ use super::{DEFAULT_CAPACITY, TestSerializer}; /// The server function receives the server half of a `tokio::io::duplex` /// connection. Every provided frame is written to the client side in order and /// the collected output is returned once the server task completes. If the -/// server panics, the panic message is surfaced as an `io::Error` beginning -/// with `"server task failed"`. +/// Server I/O failures are propagated to the caller. If the server panics, the +/// panic message is surfaced as an `io::Error` beginning with +/// `"server task failed"`. /// /// ```rust /// use tokio::io::{AsyncWriteExt, DuplexStream}; /// use wireframe_testing::helpers::drive::drive_internal; /// -/// async fn echo(mut server: DuplexStream) { let _ = server.write_all(&[1, 2]).await; } +/// async fn echo(mut server: DuplexStream) -> std::io::Result<()> { +/// server.write_all(&[1, 2]).await +/// } /// /// # async fn demo() -> std::io::Result<()> { /// let bytes = drive_internal(echo, vec![vec![0]], 64).await?; @@ -35,17 +41,17 @@ pub(super) async fn drive_internal( ) -> io::Result> where F: FnOnce(DuplexStream) -> Fut, - Fut: std::future::Future + Send, + Fut: std::future::Future> + Send, { let (mut client, server) = duplex(capacity); let server_fut = async { use futures::FutureExt as _; - let result = std::panic::AssertUnwindSafe(server_fn(server)) + let result = std::panic::AssertUnwindSafe(async { server_fn(server).await }) .catch_unwind() .await; match result { - Ok(()) => Ok(()), + Ok(result) => result, Err(panic) => { let panic_msg = wireframe::panic::format_panic(&panic); Err(io::Error::other(format!("server task failed: {panic_msg}"))) @@ -211,6 +217,10 @@ forward_default! { /// # Ok(()) /// # } /// ``` +#[expect( + deprecated, + reason = "compatibility helper drives the legacy builder API" +)] pub async fn drive_with_frames_with_capacity( app: WireframeApp, frames: Vec>, @@ -222,13 +232,64 @@ where E: Packet, { drive_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, frames, capacity, ) .await } +/// Prepare `app`, drive it with multiple frames, and return the response bytes. +/// +/// This is the migration path for tests that own a builder but need to exercise +/// the prepared connection path. +/// +/// # Errors +/// +/// Returns an I/O error if preparation or duplex connection handling fails. +pub async fn prepare_and_drive_with_frames( + app: WireframeApp, + frames: Vec>, +) -> io::Result> +where + S: TestSerializer, + C: Send + 'static, + E: Packet, + F: FrameCodec, +{ + let prepared = app + .prepare() + .await + .map_err(|error| io::Error::other(error.to_string()))?; + drive_prepared_with_frames(&prepared, frames).await +} + +/// Drive one connection through an already prepared application. +/// +/// The borrowed prepared application can be driven repeatedly, allowing tests +/// to verify that route middleware transforms are not rebuilt per connection. +/// +/// # Errors +/// +/// Returns an I/O error from the duplex transport or prepared application. +pub async fn drive_prepared_with_frames( + app: &PreparedApp, + frames: Vec>, +) -> io::Result> +where + S: TestSerializer, + C: Send + 'static, + E: Packet, + F: FrameCodec, +{ + drive_internal( + |server| async move { app.handle_connection_result(server).await }, + frames, + DEFAULT_CAPACITY, + ) + .await +} + forward_default! { /// Feed a single frame into a mutable `app`, allowing the instance to be reused /// across calls. @@ -293,6 +354,10 @@ forward_default! { /// # Ok(()) /// # } /// ``` +#[expect( + deprecated, + reason = "compatibility helper drives the legacy builder API" +)] pub async fn drive_with_frames_with_capacity_mut( app: &mut WireframeApp, frames: Vec>, @@ -304,7 +369,7 @@ where E: Packet, { drive_internal( - |server| async { app.handle_connection(server).await }, + |server| async { app.handle_connection_result(server).await }, frames, capacity, ) diff --git a/wireframe_testing/src/helpers/fragment_drive.rs b/wireframe_testing/src/helpers/fragment_drive.rs index f8dd9d59..b49f861c 100644 --- a/wireframe_testing/src/helpers/fragment_drive.rs +++ b/wireframe_testing/src/helpers/fragment_drive.rs @@ -118,7 +118,7 @@ async fn drive_fragments_internal( where F: FrameCodec, H: FnOnce(DuplexStream) -> Fut, - Fut: std::future::Future + Send, + Fut: std::future::Future> + Send, { let serialized_envelopes = fragment_and_encode(request.fragmenter, request.payload, request.route_id)?; @@ -190,6 +190,7 @@ where /// # Ok(()) /// # } /// ``` +#[expect(deprecated, reason = "compatibility helper drives the legacy builder API")] pub async fn drive_with_fragments_with_capacity( app: WireframeApp, codec: &F, @@ -204,7 +205,7 @@ where F: FrameCodec, { let frames = drive_fragments_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, codec, FragmentRequest::new(fragmenter, payload).with_capacity(capacity), ) @@ -236,6 +237,7 @@ where /// # Ok(()) /// # } /// ``` +#[expect(deprecated, reason = "compatibility helper drives the legacy builder API")] pub async fn drive_with_fragments_mut( app: &mut WireframeApp, codec: &F, @@ -249,7 +251,7 @@ where F: FrameCodec, { let frames = drive_fragments_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, codec, FragmentRequest::new(fragmenter, payload), ) @@ -285,6 +287,7 @@ where /// # Ok(()) /// # } /// ``` +#[expect(deprecated, reason = "compatibility helper drives the legacy builder API")] pub async fn drive_with_fragment_frames( app: WireframeApp, codec: &F, @@ -298,7 +301,7 @@ where F: FrameCodec, { drive_fragments_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, codec, FragmentRequest::new(fragmenter, payload), ) @@ -333,6 +336,7 @@ where /// # Ok(()) /// # } /// ``` +#[expect(deprecated, reason = "compatibility helper drives the legacy builder API")] pub async fn drive_with_partial_fragments( app: WireframeApp, codec: &F, @@ -350,7 +354,7 @@ where let encoded = encode_payloads_with_codec(codec, serialized_envelopes)?; let wire_bytes: Vec = encoded.into_iter().flatten().collect(); let raw = drive_chunked_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, wire_bytes, chunk_size, DEFAULT_CAPACITY, diff --git a/wireframe_testing/src/helpers/partial_frame.rs b/wireframe_testing/src/helpers/partial_frame.rs index a3bca30a..66883e3a 100644 --- a/wireframe_testing/src/helpers/partial_frame.rs +++ b/wireframe_testing/src/helpers/partial_frame.rs @@ -61,7 +61,9 @@ impl ChunkConfig { /// of the public `drive_with_partial_*` wrappers instead. /// /// ```rust,ignore -/// async fn echo(mut s: DuplexStream) { let _ = s.write_all(&[1, 2]).await; } +/// async fn echo(mut s: DuplexStream) -> std::io::Result<()> { +/// s.write_all(&[1, 2]).await +/// } /// /// let out = drive_chunked_internal( /// echo, @@ -80,17 +82,17 @@ pub(super) async fn drive_chunked_internal( ) -> io::Result> where F: FnOnce(DuplexStream) -> Fut, - Fut: std::future::Future + Send, + Fut: std::future::Future> + Send, { let (mut client, server) = duplex(capacity); let server_fut = async { use futures::FutureExt as _; - let result = std::panic::AssertUnwindSafe(server_fn(server)) + let result = std::panic::AssertUnwindSafe(async { server_fn(server).await }) .catch_unwind() .await; match result { - Ok(()) => Ok(()), + Ok(result) => result, Err(panic) => { let panic_msg = wireframe::panic::format_panic(&panic); Err(io::Error::new( @@ -139,7 +141,7 @@ async fn drive_partial_frames_internal( where F: FrameCodec, H: FnOnce(DuplexStream) -> Fut, - Fut: std::future::Future + Send, + Fut: std::future::Future> + Send, { let encoded = encode_payloads_with_codec(codec, payloads)?; let wire_bytes: Vec = encoded.into_iter().flatten().collect(); @@ -215,6 +217,7 @@ where /// # Ok(()) /// # } /// ``` +#[expect(deprecated, reason = "compatibility helper drives the legacy builder API")] pub async fn drive_with_partial_frames_with_capacity( app: WireframeApp, codec: &F, @@ -229,7 +232,7 @@ where F: FrameCodec, { let frames = drive_partial_frames_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, codec, payloads, ChunkConfig::with_capacity(chunk_size, capacity), @@ -261,6 +264,7 @@ where /// # Ok(()) /// # } /// ``` +#[expect(deprecated, reason = "compatibility helper drives the legacy builder API")] pub async fn drive_with_partial_frames_mut( app: &mut WireframeApp, codec: &F, @@ -274,7 +278,7 @@ where F: FrameCodec, { let frames = drive_partial_frames_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, codec, payloads, ChunkConfig::new(chunk_size), @@ -311,6 +315,7 @@ where /// # Ok(()) /// # } /// ``` +#[expect(deprecated, reason = "compatibility helper drives the legacy builder API")] pub async fn drive_with_partial_codec_frames( app: WireframeApp, codec: &F, @@ -324,7 +329,7 @@ where F: FrameCodec, { drive_partial_frames_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, codec, payloads, ChunkConfig::new(chunk_size), diff --git a/wireframe_testing/src/helpers/runtime.rs b/wireframe_testing/src/helpers/runtime.rs index 6c53d279..926f9c70 100644 --- a/wireframe_testing/src/helpers/runtime.rs +++ b/wireframe_testing/src/helpers/runtime.rs @@ -29,6 +29,10 @@ use super::{EMPTY_SERVER_CAPACITY, MAX_CAPACITY, TestSerializer, drive::drive_in /// # Ok(()) /// # } /// ``` +#[expect( + deprecated, + reason = "compatibility helper drives the legacy builder API" +)] pub async fn run_app( app: WireframeApp, frames: Vec>, @@ -54,7 +58,7 @@ where } drive_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, frames, capacity, ) @@ -78,6 +82,10 @@ where /// run_with_duplex_server(app).await; /// # } /// ``` +#[expect( + deprecated, + reason = "compatibility helper drives the legacy builder API" +)] pub async fn run_with_duplex_server(app: WireframeApp) where S: TestSerializer, diff --git a/wireframe_testing/src/helpers/slow_io.rs b/wireframe_testing/src/helpers/slow_io.rs index ac55921e..bd8206ba 100644 --- a/wireframe_testing/src/helpers/slow_io.rs +++ b/wireframe_testing/src/helpers/slow_io.rs @@ -205,7 +205,7 @@ async fn drive_slow_internal( ) -> io::Result> where F: FnOnce(DuplexStream) -> Fut, - Fut: std::future::Future, + Fut: std::future::Future>, { let config = config.validate()?; let (client, server) = tokio::io::duplex(config.capacity); @@ -217,7 +217,7 @@ where .catch_unwind() .await; match result { - Ok(()) => Ok(()), + Ok(result) => result, Err(panic) => { let panic_msg = wireframe::panic::format_panic(&panic); Err(io::Error::other(format!("server task failed: {panic_msg}"))) @@ -266,6 +266,7 @@ fn encode_length_delimited_payloads(payloads: Vec>) -> io::Result( app: WireframeApp, frames: Vec>, @@ -278,7 +279,7 @@ where { let wire_bytes: Vec = frames.into_iter().flatten().collect(); drive_slow_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, wire_bytes, config, ) @@ -287,6 +288,7 @@ where /// Encode payloads with the default length-delimited codec and drive `app` /// using optional slow writer and reader pacing. +#[expect(deprecated, reason = "compatibility helper drives the legacy builder API")] pub async fn drive_with_slow_payloads( app: WireframeApp, payloads: Vec>, @@ -299,7 +301,7 @@ where { let wire_bytes = encode_length_delimited_payloads(payloads)?; drive_slow_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, wire_bytes, config, ) @@ -355,6 +357,7 @@ where /// Drive `app` with codec-encoded payloads using optional slow I/O pacing and /// return decoded response frames. +#[expect(deprecated, reason = "compatibility helper drives the legacy builder API")] pub async fn drive_with_slow_codec_frames( app: WireframeApp, codec: &F, @@ -370,7 +373,7 @@ where let encoded = encode_payloads_with_codec(codec, payloads)?; let wire_bytes: Vec = encoded.into_iter().flatten().collect(); let raw = drive_slow_internal( - |server| async move { app.handle_connection(server).await }, + |server| async move { app.handle_connection_result(server).await }, wire_bytes, config, ) diff --git a/wireframe_testing/src/helpers/tests/helper_tests.rs b/wireframe_testing/src/helpers/tests/helper_tests.rs index 71421916..4d411bc0 100644 --- a/wireframe_testing/src/helpers/tests/helper_tests.rs +++ b/wireframe_testing/src/helpers/tests/helper_tests.rs @@ -4,14 +4,36 @@ use std::{io, sync::Arc}; use futures::future::BoxFuture; +use tokio::io::DuplexStream; use wireframe::{ app::{Envelope, WireframeApp}, prelude::Serializer, serializer::BincodeSerializer, }; +use super::super::drive::drive_internal; use crate::helpers::{MAX_CAPACITY, decode_frames, drive_with_payloads, run_app}; +/// Convert synchronous server-factory panics into the documented I/O error. +#[tokio::test] +async fn drive_internal_converts_synchronous_server_panics_to_io_errors() { + let result = drive_internal( + |_: DuplexStream| -> std::future::Ready> { + panic!("synchronous server factory panic") + }, + Vec::new(), + 64, + ) + .await; + + let error = result.expect_err("synchronous server panic should become an I/O error"); + assert_eq!(error.kind(), io::ErrorKind::Other); + assert!( + error.to_string().starts_with("server task failed"), + "unexpected panic conversion: {error}" + ); +} + #[tokio::test] async fn run_app_rejects_zero_capacity() { let app: WireframeApp = diff --git a/wireframe_testing/src/lib.rs b/wireframe_testing/src/lib.rs index f58046b3..bc32882f 100644 --- a/wireframe_testing/src/lib.rs +++ b/wireframe_testing/src/lib.rs @@ -44,6 +44,7 @@ pub use helpers::{ decode_frames, decode_frames_with_codec, decode_frames_with_max, + drive_prepared_with_frames, drive_with_bincode, drive_with_codec_frames, drive_with_codec_frames_with_capacity, @@ -78,6 +79,7 @@ pub use helpers::{ mismatched_total_size_wire, new_test_codec, oversized_hotline_wire, + prepare_and_drive_with_frames, run_app, run_with_duplex_server, sequential_hotline_wire, diff --git a/wireframe_testing/src/observability/assertions.rs b/wireframe_testing/src/observability/assertions.rs index 97d3347f..c0077f52 100644 --- a/wireframe_testing/src/observability/assertions.rs +++ b/wireframe_testing/src/observability/assertions.rs @@ -23,7 +23,7 @@ impl ObservabilityHandle { /// /// # Examples /// - /// ```no_run + /// ``` /// use wireframe_testing::{ObservabilityHandle, observability::Labels}; /// /// let mut obs = ObservabilityHandle::new(); @@ -58,6 +58,54 @@ impl ObservabilityHandle { } } + /// Assert that a histogram contains at least one recorded value. + /// + /// # Errors + /// + /// Returns `Err` when no matching histogram contains a sample. + /// + /// # Examples + /// + /// ``` + /// use wireframe_testing::ObservabilityHandle; + /// + /// let mut obs = ObservabilityHandle::new(); + /// obs.snapshot(); + /// assert!( + /// obs.assert_histogram_recorded("wireframe_example_duration_seconds", []) + /// .is_err() + /// ); + /// ``` + pub fn assert_histogram_recorded( + &self, + name: &str, + labels: impl Into, + ) -> Result<(), String> { + let labels = labels.into(); + let recorded = self + .captured + .iter() + .filter(|(key, ..)| key.key().name() == name) + .filter(|(key, ..)| { + labels + .as_str_pairs() + .iter() + .all(|(label, value)| { + key.key() + .labels() + .any(|actual| actual.key() == *label && actual.value() == *value) + }) + }) + .any(|(.., value)| matches!(value, DebugValue::Histogram(samples) if !samples.is_empty())); + if recorded { + Ok(()) + } else { + Err(format!( + "histogram {name} with labels {labels:?} did not record a value" + )) + } + } + /// Assert no metric with the given name exists in the snapshot. /// /// # Errors