Packets, payloads, and serialization

Packets drive routing. Implement the Packet trait (or use the bundled Envelope) to expose a message identifier, optional correlation id, and raw payload. PacketParts rebuilds packets after middleware has finished editing frames, and inherit_correlation patches mismatched response identifiers while logging the discrepancy.[^8] Serializers plug into the Serializer trait; WireframeApp defaults to BincodeSerializer but can run any implementation that meets the trait bounds.[^4]

When FrameMetadata::parse succeeds, the framework extracts identifiers from metadata without deserializing the payload. If parsing fails, it falls back to full deserialization.[^9][^6] Serializer integration now uses adapter traits: EncodeWith<S> for encoding and DecodeWith<S> for decoding. Existing bincode message types remain compatible through Message, which still provides to_bytes and from_bytes helpers.[^10]

If existing code previously relied on M: Message for client/server API calls, no change is required for bincode-compatible types.

• Existing #[derive(bincode::Encode, bincode::BorrowDecode)] payloads still work in send, receive, call, and send_response.
• Serializer implementations should update method signatures to use EncodeWith<Self> and DecodeWith<Self> bounds.
• Custom serializers that rely on legacy Message-derived payload encoding should also implement wireframe::serializer::MessageCompatibilitySerializer.
• Metadata-aware serializers can implement Serializer::deserialize_with_context to inspect DeserializeContext.
• Serializer is not object-safe (Self: Sized on serializer entry points), so using dyn Serializer directly is unsupported unless concrete wrappers are provided. For migration, keep concrete serializer types in EncodeWith / DecodeWith bounds, retain wireframe::serializer::MessageCompatibilitySerializer for legacy Message-based payloads, and use SerdeMessage with SerdeSerializerBridge for Serde-only payloads. When runtime-polymorphic serializer behaviour is required, wrap concrete serializer implementations in adapter wrappers that expose object-safe APIs and delegate to Serializer::deserialize_with_context as needed.

Optional Serde bridge support is available behind the feature serializer-serde. Wrap values with SerdeMessage<T> (or into_serde_message()) and implement SerdeSerializerBridge on the serializer to reduce per-type adapter boilerplate. This explicit wrapper is required because blanket Serde adapters would overlap with the default T: Message adapter implementations.

wireframe::fragment exposes small, copy-friendly structs to describe the transport-level fragmentation state.[^41]

• MessageId uniquely identifies the logical message a fragment belongs to.
• FragmentIndex records the fragment's zero-based order.
• FragmentHeader bundles the identifier, index, and a boolean that signals whether the fragment is the last one in the sequence.
• FragmentSeries tracks the next expected fragment index and reports mismatches via structured errors.

Protocol implementers can emit a FragmentHeader for every physical frame and feed the header back into FragmentSeries to guarantee ordering before a fully reassembled message is surfaced to handlers. Behavioural tests can reuse the same types to assert that new codecs obey the transport invariants without spinning up a full server.[^42][^43]

The standalone Fragmenter helper now slices oversized payloads into capped fragments while stamping the shared MessageId and sequential FragmentIndex. Each call returns a FragmentBatch that reports whether the message required fragmentation and yields individual FragmentFrame values for serialization or logging. This keeps transport experiments lightweight while the full adapter layer evolves. The helper is fallible—FragmentationError surfaces encoding failures or index overflows—so production code should bubble the error up or log it rather than unwrapping.

use std::num::NonZeroUsize;
use wireframe::fragment::Fragmenter;

let fragmenter = Fragmenter::new(NonZeroUsize::new(512).unwrap());
let payload = [0_u8; 1400];
let batch = fragmenter.fragment_bytes(&payload).expect("fragment");
assert_eq!(batch.len(), 3);

for fragment in batch.fragments() {
    tracing::info!(
        msg_id = fragment.header().message_id().get(),
        index = fragment.header().fragment_index().get(),
        final = fragment.header().is_last_fragment(),
        payload_len = fragment.payload().len(),
    );
}

A companion Reassembler mirrors the helper on the inbound path. It buffers fragments per MessageId, rejects out-of-order fragments, and enforces a maximum assembled size while exposing purge_expired to clear stale partial messages after a configurable timeout. When the final fragment arrives, it returns a ReassembledMessage that can be decoded into the original type.

use std::{num::NonZeroUsize, time::Duration};
use wireframe::fragment::{
    FragmentHeader,
    FragmentIndex,
    MessageId,
    Reassembler,
};

let mut reassembler =
    Reassembler::new(NonZeroUsize::new(512).expect("non-zero capacity"), Duration::from_secs(30));

let header = FragmentHeader::new(MessageId::new(9), FragmentIndex::zero(), true);
let complete = reassembler
    .push(header, [0_u8; 12])
    .expect("fragments accepted")
    .expect("single fragment completes the message");

// Decode when ready:
// let message: MyType = complete.decode().expect("decode");

wireframe::request exposes RequestParts to separate routing metadata from streaming request payloads.[^45]

• RequestParts::id() returns the message identifier for routing.
• RequestParts::correlation_id() returns the optional correlation identifier.
• RequestParts::metadata() returns protocol-defined header bytes.

This type pairs with RequestBodyStream for incremental consumption of large request payloads. Handlers can choose between buffered (existing) and streaming consumption; the streaming path is opt-in.

use wireframe::request::RequestParts;

let parts = RequestParts::new(42, Some(123), vec![0x01, 0x02]);
assert_eq!(parts.id(), 42);
assert_eq!(parts.correlation_id(), Some(123));
assert_eq!(parts.metadata(), &[0x01, 0x02]);

Unlike PacketParts (which carries the raw payload for envelope reconstruction), RequestParts carries only protocol-defined metadata required to interpret the streaming body. The body itself is consumed through a separate stream, enabling back-pressure and incremental processing.[^46] When MessageAssembler is configured, this split happens after any transport fragment reassembly, so handlers and protocol extractors see protocol-level metadata rather than lower-level fragment state.

Wireframe exposes a protocol-facing MessageAssembler hook that parses per-frame headers into FrameHeader::First and FrameHeader::Continuation values. It returns a ParsedFrameHeader that includes the header length so the remaining bytes can be treated as the body chunk.

Register an assembler with WireframeApp::with_message_assembler:

use wireframe::{
    app::WireframeApp,
    message_assembler::{
        FrameHeader,
        FirstFrameHeader,
        MessageAssembler,
        MessageKey,
        ParsedFrameHeader,
    },
};

struct DemoAssembler;

impl MessageAssembler for DemoAssembler {
    fn parse_frame_header(
        &self,
        _payload: &[u8],
    ) -> Result<ParsedFrameHeader, std::io::Error> {
        Ok(ParsedFrameHeader::new(
            FrameHeader::First(FirstFrameHeader {
                message_key: MessageKey(1),
                metadata_len: 0,
                body_len: 0,
                total_body_len: None,
                is_last: true,
            }),
            0,
        ))
    }
}

let _app = WireframeApp::new()
    .expect("builder")
    .with_message_assembler(DemoAssembler);

When configured, this hook now runs on the inbound connection path after transport fragmentation reassembly and before handler dispatch. Incomplete assemblies remain buffered per message key until completion or timeout eviction.

The assembler's output drives the handler-facing request shape:

• fully assembled messages continue through the buffered request path; and
• streaming-capable messages surface as RequestParts plus RequestBodyStream / StreamingBody.

Message-assembly parsing and continuity failures are treated as inbound deserialization failures and follow the existing failure threshold policy.

WireframeApp::message_assembler returns the configured hook as an Option<&Arc<dyn MessageAssembler>> if direct access is required.

MemoryBudgets configures explicit per-connection byte caps for inbound buffering and assembly:

• bytes buffered per message;
• bytes buffered per connection; and
• bytes buffered across in-flight assemblies.

Configure budgets through the app builder:

use std::num::NonZeroUsize;
use wireframe::app::{BudgetBytes, MemoryBudgets, WireframeApp};

let budgets = MemoryBudgets::new(
    BudgetBytes::new(NonZeroUsize::new(16 * 1024).expect("non-zero")),
    BudgetBytes::new(NonZeroUsize::new(64 * 1024).expect("non-zero")),
    BudgetBytes::new(NonZeroUsize::new(48 * 1024).expect("non-zero")),
);

let _app = WireframeApp::new()
    .expect("builder creation should not fail")
    .memory_budgets(budgets)
    .read_timeout_ms(250);

When budgets are configured, the message assembly subsystem enforces them at frame acceptance time. Frames that would cause the total buffered bytes to exceed the per-connection or in-flight budget are rejected, the offending partial assembly is freed, and the failure is surfaced through the existing deserialization-failure policy (InvalidData). The effective per-message limit is the minimum of the fragmentation max_message_size and the configured bytes_per_message. Single-frame messages that complete immediately are never counted against aggregate budgets, since they do not buffer.

If memory_budgets(...) is not configured explicitly, Wireframe derives the same three fields from buffer_capacity, so the protection model remains active even on the default path.

Wireframe provides a three-tier protection model for inbound memory budgets:

1. Per-frame enforcement — frames that would cause total buffered bytes to exceed the per-connection or in-flight budget are rejected, the offending partial assembly is freed, and the failure is surfaced through the existing deserialization-failure policy (InvalidData).
2. Soft-limit read pacing — when buffered assembly bytes reach the soft-pressure threshold (80% of the smaller aggregate cap from bytes_per_connection and bytes_in_flight), the inbound connection loop briefly pauses socket reads before polling the next frame. This propagates back-pressure to senders while allowing progress on in-flight assemblies.
3. Hard-cap connection abort — if total buffered bytes strictly exceed the aggregate cap (100% of the smaller of bytes_per_connection and bytes_in_flight), the connection is terminated immediately with InvalidData. This is a defence-in-depth safety net; under normal operation, per-frame enforcement prevents this state from being reached.

Derived budget defaults

When memory_budgets(...) is not called on the builder, Wireframe derives sensible defaults automatically from buffer_capacity (the codec's max_frame_length()). The derived values use the same multiplier pattern as fragmentation defaults:

All three protection tiers (per-frame enforcement, soft-limit read pacing, and hard-cap connection abort) are active with derived defaults. Changing buffer_capacity adjusts derived budgets proportionally. Calling .memory_budgets(...) overrides the derived defaults entirely.

Message key multiplexing (8.2.3)

The MessageAssemblyState type manages multiple concurrent message assemblies keyed by MessageKey. This enables interleaved frame streams where frames from different logical messages arrive on the same connection:

use std::{num::NonZeroUsize, time::Duration};
use wireframe::message_assembler::{
    ContinuationFrameHeader,
    EnvelopeId,
    EnvelopeRouting,
    FirstFrameHeader,
    FirstFrameInput,
    FrameSequence,
    MessageAssemblyState,
    MessageKey,
};

let mut state = MessageAssemblyState::new(
    NonZeroUsize::new(1_048_576).expect("non-zero size"), // 1 mebibyte (MiB) max message
    Duration::from_secs(30),                               // 30s timeout for partial assemblies
);

// First frame for message key=1
let first1 = FirstFrameHeader {
    message_key: MessageKey(1),
    metadata_len: 0,
    body_len: 5,
    total_body_len: Some(15),
    is_last: false,
};
let routing1 = EnvelopeRouting { envelope_id: EnvelopeId(1), correlation_id: None };
let input1 = FirstFrameInput::new(&first1, routing1, vec![], b"hello")
    .expect("header lengths match");
state.accept_first_frame(input1)?;

// First frame for message key=2 (interleaved)
let first2 = FirstFrameHeader {
    message_key: MessageKey(2),
    metadata_len: 0,
    body_len: 5,
    total_body_len: None,
    is_last: false,
};
let routing2 = EnvelopeRouting { envelope_id: EnvelopeId(2), correlation_id: None };
let input2 = FirstFrameInput::new(&first2, routing2, vec![], b"world")
    .expect("header lengths match");
state.accept_first_frame(input2)?;

// Continuation for key=1 completes its message
let cont1 = ContinuationFrameHeader {
    message_key: MessageKey(1),
    sequence: Some(FrameSequence(1)),
    body_len: 10,
    is_last: true,
};
let msg1 = state.accept_continuation_frame(&cont1, b" completed")?
    .expect("message 1 should complete");
assert_eq!(msg1.body(), b"hello completed");

Continuity validation (8.2.4)

The MessageSeries type validates frame ordering when protocols supply sequence numbers via ContinuationFrameHeader::sequence. It detects:

• Out-of-order frames: sequence gaps produce MessageSeriesError::SequenceMismatch
• Duplicate frames: already-processed sequences produce MessageSeriesError::DuplicateFrame
• Frames after completion: produce MessageSeriesError::SeriesComplete

For protocols that do not supply sequence numbers, the series accepts frames in any order (ordering validation is skipped).

use wireframe::message_assembler::{
    ContinuationFrameHeader,
    FirstFrameHeader,
    FrameSequence,
    MessageKey,
    MessageSeries,
    MessageSeriesError,
    MessageSeriesStatus,
};

let first = FirstFrameHeader {
    message_key: MessageKey(1),
    metadata_len: 0,
    body_len: 10,
    total_body_len: None,
    is_last: false,
};
let mut series = MessageSeries::from_first_frame(&first);

// Accept continuation with sequence 1
let cont1 = ContinuationFrameHeader {
    message_key: MessageKey(1),
    sequence: Some(FrameSequence(1)),
    body_len: 5,
    is_last: false,
};
assert_eq!(series.accept_continuation(&cont1), Ok(MessageSeriesStatus::Incomplete));

// Attempting sequence 3 (gap) fails
let cont3 = ContinuationFrameHeader {
    message_key: MessageKey(1),
    sequence: Some(FrameSequence(3)), // Expected 2
    body_len: 5,
    is_last: false,
};
assert!(matches!(
    series.accept_continuation(&cont3),
    Err(MessageSeriesError::SequenceMismatch { .. })
));

// Accept sequence 2, then try sequence 1 again (duplicate)
let cont2 = ContinuationFrameHeader {
    message_key: MessageKey(1),
    sequence: Some(FrameSequence(2)),
    body_len: 5,
    is_last: false,
};
assert_eq!(series.accept_continuation(&cont2), Ok(MessageSeriesStatus::Incomplete));

let dup = ContinuationFrameHeader {
    message_key: MessageKey(1),
    sequence: Some(FrameSequence(1)), // Already seen
    body_len: 5,
    is_last: false,
};
assert!(matches!(
    series.accept_continuation(&dup),
    Err(MessageSeriesError::DuplicateFrame { .. })
));

// Complete the series, then try to add more (SeriesComplete)
let final_cont = ContinuationFrameHeader {
    message_key: MessageKey(1),
    sequence: Some(FrameSequence(3)),
    body_len: 5,
    is_last: true,
};
assert_eq!(series.accept_continuation(&final_cont), Ok(MessageSeriesStatus::Complete));

let extra = ContinuationFrameHeader {
    message_key: MessageKey(1),
    sequence: Some(FrameSequence(4)),
    body_len: 5,
    is_last: false,
};
assert!(matches!(
    series.accept_continuation(&extra),
    Err(MessageSeriesError::SeriesComplete)
));

Handlers can opt into streaming request bodies using the StreamingBody extractor or by accepting a RequestBodyStream directly. The framework creates a bounded channel and forwards body chunks as they arrive; back-pressure propagates automatically when the handler consumes slower than the network delivers. Routing and correlation metadata stay available immediately through RequestParts, so handlers can begin work before the full body is assembled.

use tokio::io::AsyncReadExt;
use wireframe::request::{RequestBodyReader, RequestBodyStream, RequestParts};

async fn handle_upload(parts: RequestParts, body: RequestBodyStream) {
    let mut reader = RequestBodyReader::new(body);
    let mut buf = Vec::new();
    reader.read_to_end(&mut buf).await.expect("read body");

    log::info!(
        "received {} bytes for request {}",
        buf.len(),
        parts.id()
    );
}

The RequestBodyReader adapter implements AsyncRead, allowing protocol crates to reuse existing parsers. For raw stream access, use the RequestBodyStream directly with StreamExt methods:

use bytes::Bytes;
use futures::StreamExt;
use wireframe::request::{RequestBodyStream, RequestParts};

async fn handle_stream(parts: RequestParts, mut body: RequestBodyStream) {
    while let Some(result) = body.next().await {
        match result {
            Ok(chunk) => log::debug!("received {} bytes", chunk.len()),
            Err(e) => log::error!("stream error: {e}"),
        }
    }
}

The StreamingBody extractor wraps the stream with convenience methods:

use wireframe::{
    extractor::StreamingBody,
    request::RequestParts,
};

async fn with_extractor(parts: RequestParts, body: StreamingBody) {
    // Convert to AsyncRead
    let reader = body.into_reader();

    // Or access the raw stream
    // let stream = body.into_stream();
}

Back-pressure is enforced via bounded channels: when the internal buffer fills, the framework pauses reading from the socket until the handler drains pending chunks. This prevents memory exhaustion under slow consumer conditions. When the request arrived through MessageAssembler, the same per-connection memory budgets continue to govern partial buffered state before chunks reach the handler. The body_channel helper creates channels with configurable capacity:

use wireframe::request::body_channel;

// Create a channel with capacity for 8 chunks
let (tx, rx) = body_channel(8);
// tx: connection sends chunks
// rx: handler consumes via RequestBodyStream

See [ADR 0002][adr-0002-ref] for the complete design rationale.