Implement transport-level fragmentation and inbound reassembly

[leynos]

Summary

Introduce a transport-agnostic Reassembler to buffer inbound fragments, reassemble them into complete messages, and enforce resource safeguards (size cap and timeout-based eviction). This mirrors the outbound Fragmenter path and provides a robust inbound fragmentation workflow with precise error handling and test coverage.

Changes

• New inbound helper: Reassembler (src/fragment/reassembler.rs)
  • Buffers fragments per MessageId using FragmentSeries to enforce ordering
  • Enforces a configurable maximum reconstructed payload size (max_message_size)
  • Purges stale partial messages after a configurable timeout via purge_expired and purge_expired_at
  • Returns a ReassembledMessage on completion or None if more fragments are needed
  • Public API supports pushing fragments with push and push_at, and querying buffer state via buffered_len
  • Decodes payloads via ReassembledMessage::decode for typed messages
• Error handling: ReassemblyError added to src/fragment/error.rs
  • Fragment-level errors bubble up as Fragment in reassembly flow
  • MessageTooLarge error captures message_id, attempted payload size, and cap limit
• Module exports: extend fragment module to expose Reassembler, ReassembledMessage, and ReassemblyError (src/fragment/mod.rs, src/lib.rs)
• Tests
  • Unit tests for Reassembler in src/fragment/tests.rs validating:
    • Single-fragment completion
    • Ordered multi-fragment accumulation
    • Rejection of out-of-order fragments and buffer reset
    • Enforcement of maximum payload size
    • Expired-buffer purging behavior
    • Decoding a reconstructed message via a sample ExampleMessage
  • Behavioural tests scaffolding added under tests/features/fragment.feature and tests/worlds/fragment.rs with cucumber steps to drive reassembler scenarios
  • Step definitions updated in tests/steps/fragment_steps.rs to configure and exercise the reassembler
• Documentation and roadmap
  • Updated docs/generic-message-fragmentation-and-re-assembly-design.md to include transport-agnostic Reassembler details
  • Roadmap updated to reflect completed Reassembler work in the testing/verification pipeline
  • Users guide updated with a code example showing how to initialise and use Reassembler (fragment inbound path)

Public API and usage

• Reassembler::new(max_message_size, timeout) -> Self
• push(header, payload) -> Result<Option, ReassemblyError>
• push_at(header, payload, now) -> Result<Option, ReassemblyError>
• purge_expired() / purge_expired_at(now) -> Vec
• buffered_len() -> usize
• ReassembledMessage { message_id(), payload(), decode() }

Example snippet (from docs/users-guide):

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

let mut reassembler =
    Reassembler::new(NonZeroUsize::new(512).unwrap(), Duration::from_secs(30));
let header = FragmentHeader::new(MessageId::new(9), FragmentIndex::zero(), true);
let complete: ReassembledMessage = reassembler
    .push(header, &[0_u8; 12])
    .expect("fragments accepted")
    .expect("single fragment completes the message");

let message: ExampleMessage = complete.decode().expect("decode");

Testing and verification

• Unit tests cover core reassembly semantics (ordering, sizing, timeouts)
• Integration tests via cucumber-based features exercise end-to-end fragmentation/reassembly scenarios
• Purge behavior ensures resources aren’t leaked by abandoned partial assemblies

Compatibility and breaking changes

• No breaking changes to existing Fragmenter usage
• Public reassembler components introduced under wireframe::fragment, with explicit exports in the crate root for ergonomic access

Test plan

• cargo test (unit tests for Reassembler and Fragment-related components)
• Cucumber-based feature scenarios for reassembler behavior (order, cap, timeout, eviction)
• Manual review of error handling around out-of-order fragments and cap overflows

If you need any additional usage examples or want to adjust the default max size or timeout values in tests, I can add a short per-feature guide in the docs.

🌿 Generated by Terry

---

ℹ️ Tag @terragon-labs to ask questions and address PR feedback

📎 Task: https://www.terragonlabs.com/task/694b974b-8b76-40fb-ae28-7049f9401ef0

Summary by Sourcery

Introduce an inbound, transport-agnostic fragment reassembler with size and timeout safeguards, plus supporting errors, tests, and documentation updates.

New Features:

• Add a Reassembler helper and ReassembledMessage type to reconstruct inbound messages from fragments with configurable size caps and timeout-based eviction.
• Expose new reassembly types and errors through the fragment module and crate root for public use.

Enhancements:

• Extend fragmentation error handling with a ReassemblyError enum to capture fragment-level issues and over-limit messages.
• Add cucumber step definitions, world state, and scenarios to exercise reassembly behavior including ordering, caps, and eviction.
• Document inbound reassembly usage and update the design and roadmap to reflect the new Reassembler and its guarantees.

Documentation:

• Update the user guide with an example of configuring and using the Reassembler on the inbound path.
• Extend fragmentation design docs to describe the transport-agnostic Reassembler and its resource safeguards.
• Mark Reassembler implementation, buffered timeouts, and tests as complete in the roadmap.

Tests:

• Add unit tests covering reassembler semantics such as ordered accumulation, size cap enforcement, timeout-based purging, and decode of reconstructed messages.
• Enhance cucumber feature specs and step implementations to validate end-to-end reassembly flows and error conditions.

[sourcery-ai]

Reviewer's Guide

Adds a transport-agnostic inbound Reassembler with size/time caps, integrates it into the fragment module’s public API, and wires comprehensive unit and Cucumber tests plus docs for reassembly behavior.

Sequence diagram for Reassembler push_at fragment processing

sequenceDiagram
    actor Client
    participant Reassembler
    participant Buffers as Buffers_HashMap
    participant Partial as PartialMessage
    participant Series as FragmentSeries

    Client->>Reassembler: push_at(header, payload, now)
    Reassembler->>Reassembler: purge_expired_at(now)
    Reassembler->>Buffers: entry(header.message_id())
    alt Existing_partial_message
        Buffers-->>Reassembler: Occupied(entry)
        Reassembler->>Series: accept(header)
        alt Ordering_ok_incomplete
            Series-->>Reassembler: FragmentStatus::Incomplete
            Reassembler->>Partial: len()
            Reassembler->>Reassembler: assert_within_limit(max_message_size, message_id, attempted)
            alt Within_size_cap
                Reassembler->>Partial: push(payload)
                Reassembler-->>Client: Ok(None)
            else Exceeds_size_cap
                Reassembler-->>Client: Err(ReassemblyError::MessageTooLarge)
            end
        else Ordering_ok_complete
            Series-->>Reassembler: FragmentStatus::Complete
            Reassembler->>Partial: len()
            Reassembler->>Reassembler: assert_within_limit(max_message_size, message_id, attempted)
            alt Within_size_cap
                Reassembler->>Partial: push(payload)
                Reassembler->>Partial: into_buffer()
                Reassembler->>Reassembler: ReassembledMessage::new(message_id, buffer)
                Reassembler-->>Client: Ok(Some(ReassembledMessage))
            else Exceeds_size_cap
                Reassembler-->>Client: Err(ReassemblyError::MessageTooLarge)
            end
        else Ordering_error
            Series-->>Reassembler: Err(FragmentError)
            Reassembler-->>Client: Err(ReassemblyError::Fragment)
        end
    else First_fragment_for_message
        Buffers-->>Reassembler: Vacant(entry)
        Reassembler->>Series: FragmentSeries::new(message_id)
        Reassembler->>Series: accept(header)
        alt Ordering_ok_incomplete
            Series-->>Reassembler: FragmentStatus::Incomplete
            Reassembler->>Reassembler: assert_within_limit(max_message_size, message_id, payload.len())
            Reassembler->>Buffers: insert(message_id, PartialMessage::new(series, payload, now))
            Reassembler-->>Client: Ok(None)
        else Ordering_ok_complete
            Series-->>Reassembler: FragmentStatus::Complete
            Reassembler->>Reassembler: assert_within_limit(max_message_size, message_id, payload.len())
            Reassembler->>Reassembler: ReassembledMessage::new(message_id, payload.to_vec())
            Reassembler-->>Client: Ok(Some(ReassembledMessage))
        else Ordering_error
            Series-->>Reassembler: Err(FragmentError)
            Reassembler-->>Client: Err(ReassemblyError::Fragment)
        end
    end

Loading

Class diagram for inbound Reassembler and related types

classDiagram
    class Reassembler {
        +NonZeroUsize max_message_size
        +Duration timeout
        +HashMap~MessageId, PartialMessage~ buffers
        +Reassembler new(max_message_size: NonZeroUsize, timeout: Duration) Reassembler
        +push(header: FragmentHeader, payload: impl AsRef~[u8]~) Result~Option~ReassembledMessage~~ ReassemblyError
        +push_at(header: FragmentHeader, payload: impl AsRef~[u8]~, now: Instant) Result~Option~ReassembledMessage~~ ReassemblyError
        +purge_expired() Vec~MessageId~
        +purge_expired_at(now: Instant) Vec~MessageId~
        +buffered_len() usize
        -assert_within_limit(limit: NonZeroUsize, message_id: MessageId, attempted: usize) Result~(), ReassemblyError
    }

    class PartialMessage {
        -FragmentSeries series
        -Vec~u8~ buffer
        -Instant started_at
        +new(series: FragmentSeries, payload: &[u8], started_at: Instant) PartialMessage
        +push(payload: &[u8]) void
        +len() usize
        +started_at() Instant
        +into_buffer() Vec~u8~
    }

    class ReassembledMessage {
        +MessageId message_id
        +Vec~u8~ payload
        +new(message_id: MessageId, payload: Vec~u8~) ReassembledMessage
        +message_id() MessageId
        +payload() &[u8]
        +into_payload() Vec~u8~
        +decode~M: Message~() Result~M, DecodeError~
    }

    class ReassemblyError {
        <<enum>>
        +Fragment(FragmentError)
        +MessageTooLarge
        +message_id: MessageId
        +attempted: usize
        +limit: NonZeroUsize
    }

    class FragmentSeries {
        +new(message_id: MessageId) FragmentSeries
        +accept(header: FragmentHeader) Result~FragmentStatus, FragmentError~
    }

    class FragmentHeader {
        +message_id() MessageId
    }

    class FragmentStatus {
        <<enum>>
        +Incomplete
        +Complete
    }

    class FragmentError {
    }

    class MessageId {
    }

    class Message {
        <<trait>>
        +from_bytes(bytes: &[u8]) Result~Self, DecodeError~
    }

    class DecodeError {
    }

    Reassembler o--> PartialMessage : buffers
    Reassembler --> ReassemblyError
    Reassembler --> FragmentHeader
    Reassembler --> FragmentSeries
    Reassembler --> FragmentStatus
    Reassembler --> MessageId
    Reassembler --> ReassembledMessage

    PartialMessage --> FragmentSeries

    ReassembledMessage --> Message
    ReassembledMessage --> MessageId
    ReassembledMessage --> DecodeError

    ReassemblyError --> FragmentError
    ReassemblyError --> MessageId

    FragmentSeries --> FragmentHeader
    FragmentSeries --> FragmentStatus

    FragmentHeader --> MessageId

Loading

File-Level Changes

Change	Details	Files
Introduce a transport-agnostic Reassembler and ReassembledMessage types for inbound fragment reassembly with size and timeout safeguards.	Implement PartialMessage and Reassembler to track FragmentSeries, buffer payloads keyed by MessageId, enforce a NonZeroUsize max_message_size, and drop buffers on ordering errors or cap violations Provide push/push_at APIs that return an optional ReassembledMessage, perform on-the-fly purge_expired_at(now), and use assert_within_limit to guard against oversized messages Define ReassembledMessage as a container with message_id, payload accessors, into_payload, and a decode<M: Message>() helper using Message::from_bytes	src/fragment/reassembler.rs
Add dedicated reassembly error type and export new reassembly primitives from the fragment module and crate root.	Introduce ReassemblyError enum with Fragment(FragmentError) and MessageTooLarge variants carrying message_id, attempted size, and limit Update fragment::mod to declare the reassembler module and re-export Reassembler, ReassembledMessage, and ReassemblyError alongside existing fragment types Expose the new reassembly types from src/lib.rs for ergonomic public API access	src/fragment/error.rs src/fragment/mod.rs src/lib.rs
Extend unit tests to cover reassembler semantics including ordering, size caps, timeouts, and typed decode.	Add helper setup_reassembler_with_first_fragment and multiple tests validating single-fragment completion, ordered accumulation, out-of-order rejection/reset, max size enforcement, timeout-based purge, and end-to-end fragmenter→reassembler decode Use Instant and Duration in tests to drive deterministic time-based purge_expired_at behavior	src/fragment/tests.rs
Augment Cucumber world and step definitions to drive reassembler scenarios and time-based eviction in behavior tests.	Extend FragmentWorld to hold a Reassembler, last_reassembled message, last_reassembly_error, simulated Instant clock, and last_evicted MessageIds with a manual Default impl Add world helpers to configure_reassembler, push_fragment, advance_time, purge_reassembly, and assertion helpers for reassembled length, no reassembly, over-limit errors, buffered_len, and evicted messages Introduce new Cucumber Given/When/Then steps mapping to the world helpers for sequential assembly, cap violations, time advancement, and eviction assertions	tests/worlds/fragment.rs tests/steps/fragment_steps.rs
Add BDD feature scenarios and documentation updates describing the inbound Reassembler usage and design and marking roadmap items as done.	Add fragment.feature scenarios covering sequential reassembly success, message-too-large rejection, and eviction of stale partial messages after timeout and purge Extend users-guide with a Reassembler overview and code example showing configuration, push, and decode of a ReassembledMessage Update generic fragmentation design doc to describe the transport-agnostic Reassembler’s behavior and safeguards, and mark corresponding roadmap items as completed	tests/features/fragment.feature docs/users-guide.md docs/generic-message-fragmentation-and-re-assembly-design.md docs/roadmap.md

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[coderabbitai]

Note

Other AI code review bot(s) detected

CodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review.

Summary by CodeRabbit

Release Notes

• New Features

  • Added inbound message reassembly with automatic timeout-based cleanup for abandoned fragments.
  • Enforced size limits to prevent oversized message buffer attacks.

• Documentation

  • Updated design specifications and user guide with reassembly examples.
  • Marked reassembly and related tasks as complete on roadmap.

• Tests

  • Comprehensive test coverage for reassembly workflows, including ordered and out-of-order fragment handling.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

Walkthrough

Summarise the change: add a transport-agnostic inbound Reassembler that buffers fragments per MessageId, enforces ordering and max assembled size, exposes timeout-based purging, returns a ReassembledMessage on completion, and includes error handling and tests.

Changes

Cohort / File(s)	Summary
Documentation Updates docs/generic-message-fragmentation-and-re-assembly-design.md, docs/roadmap.md, docs/users-guide.md	Add design doc for transport-agnostic Reassembler, mark reassembly/timeouts/tests implemented in roadmap, and provide user guide example showing Reassembler usage and decoding.
Error Handling src/fragment/error.rs	Add public ReassemblyError enum with Fragment(#[from] FragmentError) and MessageTooLarge { message_id, attempted, limit } variants and associated error messages.
Core Implementation src/fragment/reassembler.rs	Add Reassembler managing per-MessageId PartialMessage buffers, enforcing ordering and max size, offering push/push_at, purge_expired/purge_expired_at, buffered_len, and ReassembledMessage container with decode helpers.
Module Exports src/fragment/mod.rs, src/lib.rs	Export new reassembler submodule and re-export ReassembledMessage, Reassembler, and ReassemblyError from the fragment module at crate root.
Unit Tests src/fragment/tests.rs	Add comprehensive tests covering max-size handling, single/multi-fragment reassembly, ordering, payload enforcement, expiry/purge, and decoding scenarios.
Integration Tests (Gherkin) tests/features/fragment.feature, tests/steps/fragment_steps.rs	Add four feature scenarios and step definitions to test sequential reassembly, size-cap rejection, timeout eviction, and out-of-order rejection; include step helpers for configuration, fragment submission, simulated time, purging and assertions.
Test Harness Extensions tests/worlds/fragment/mod.rs, tests/worlds/fragment/reassembly.rs	Extend FragmentWorld to hold reassembler state (reassembler, last_reassembled, last_reassembly_error, now, last_evicted), implement Default, and add methods to configure reassembler, push fragments, advance time, purge, and assert reassembly outcomes.

Sequence Diagram(s)

sequenceDiagram
    actor Caller
    participant Reassembler
    participant Partial as PartialMessage (buffer)
    participant Series as FragmentSeries
    participant Output as ReassembledMessage

    Caller->>Reassembler: push(header: Fragment idx=0..N, payload)
    activate Reassembler
    alt New message
        Reassembler->>Series: create series for message_id
        Reassembler->>Partial: insert new PartialMessage(start_time, buffer)
        Reassembler-->>Caller: Ok(None)
    else Existing partial
        Reassembler->>Partial: verify next index == expected
        alt Index mismatch
            Reassembler-->>Caller: Err(ReassemblyError::Fragment(...))
        else Within limit
            Reassembler->>Partial: append payload
            alt Final fragment received
                Partial->>Output: build ReassembledMessage(message_id, buffer)
                Reassembler->>Reassembler: remove partial
                Reassembler-->>Caller: Ok(Some(Output))
            else Not final
                Reassembler-->>Caller: Ok(None)
            end
        end
    end
    deactivate Reassembler

Loading

sequenceDiagram
    actor Caller
    participant Reassembler
    participant Clock as Now / Timeout
    participant Partial as PartialMessage

    Caller->>Reassembler: purge_expired()
    activate Reassembler
    loop for each buffered partial
        Reassembler->>Clock: compute elapsed = now - partial.start
        alt elapsed > timeout
            Reassembler->>Partial: drop partial
            Reassembler-->>Caller: record evicted id
        else
            Reassembler-->>Caller: keep partial
        end
    end
    Reassembler-->>Caller: return Vec<MessageId> evicted
    deactivate Reassembler

Loading

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

• Inspect src/fragment/reassembler.rs for fragment ordering, size-limit checks and state transitions.
• Verify timeout semantics in purge_expired_at and simulated time handling in tests (tests/worlds/fragment/reassembly.rs).
• Check error mapping and messages in src/fragment/error.rs.
• Review new integration steps and Gherkin scenarios for correctness and flakiness (tests/steps/fragment_steps.rs, tests/features/fragment.feature).

Possibly related PRs

• Implement Transport-Level Fragmentation And Reassembly #393 — Provide fragmentation primitives (FragmentHeader, FragmentSeries, FragmentError) that this Reassembler builds upon.

Poem

Fragments land, then pause and wait,
Gathered, ordered, knit by fate,
Timeout guards and caps ensure,
When last piece fits, the whole's secure —
🎉 A message formed, complete and straight.

Pre-merge checks and finishing touches

✅ Passed checks (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title directly and clearly summarises the primary change: implementing transport-level fragmentation inbound reassembly, which mirrors the PR objectives and changeset precisely.
Description check	✅ Passed	The description is comprehensive, related, and accurately details the changes: new Reassembler, ReassemblyError, module exports, tests, and documentation updates matching the changeset.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch terragon/implement-transport-fragmentation-0if8br

---

📜 Recent review details

Configuration used: CodeRabbit UI

Review profile: ASSERTIVE

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between da43b0d and cb4c566.

📒 Files selected for processing (1)

• tests/worlds/fragment/reassembly.rs (1 hunks)

🧰 Additional context used

📓 Path-based instructions (1)

**/*.rs

📄 CodeRabbit inference engine (AGENTS.md)

**/*.rs: Run make check-fmt, make lint, and make test before committing. make check-fmt executes cargo fmt --workspace -- --check. make lint executes cargo clippy --workspace --all-targets --all-features -- -D warnings. make test executes cargo test --workspace.
Clippy warnings MUST be disallowed in Rust code.
Fix any warnings emitted during Rust tests in the code itself rather than silencing them.
Extract meaningfully named helper functions adhering to separation of concerns and CQRS when a Rust function is too long.
Group related parameters in meaningfully named structs when a Rust function has too many parameters.
Consider using Arc to reduce the amount of data returned when a Rust function is returning a large error.
Write unit and behavioural tests for new Rust functionality. Run both before and after making any change.
Every Rust module must begin with a module level (//!) comment explaining the module's purpose and utility.
Document public APIs in Rust using Rustdoc comments (///) so documentation can be generated with cargo doc.
Prefer immutable data in Rust and avoid unnecessary mut bindings.
Handle errors with the Result type in Rust instead of panicking where feasible.
Avoid unsafe code in Rust unless absolutely necessary and document any usage clearly.
Place function attributes after doc comments in Rust.
Do not use return in single-line Rust functions.
Use predicate functions for Rust conditional criteria with more than two branches.
Lints must not be silenced in Rust except as a last resort. Lint rule suppressions must be tightly scoped and include a clear reason.
Prefer expect over allow in Rust.
Use rstest fixtures for shared test setup in Rust.
Replace duplicated Rust tests with #[rstest(...)] parameterised cases.
Prefer mockall for mocks/stubs in Rust.
Prefer .expect() over .unwrap() in Rust.
Use concat!() to combine long string literals in Rust rather than escaping newlines with a bac...

Files:

• tests/worlds/fragment/reassembly.rs

---

⚙️ CodeRabbit configuration file

**/*.rs: * Seek to keep the cognitive complexity of functions no more than 9.

• Adhere to single responsibility and CQRS
• Place function attributes after doc comments.
• Do not use return in single-line functions.
• Move conditionals with >2 branches into a predicate function.
• Avoid unsafe unless absolutely necessary.
• Every module must begin with a //! doc comment that explains the module's purpose and utility.
• Comments and docs must follow en-GB-oxendict (-ize / -yse / -our) spelling and grammar
• Lints must not be silenced except as a last resort.
  • #[allow] is forbidden.
  • Only narrowly scoped #[expect(lint, reason = "...")] is allowed.
  • No lint groups, no blanket or file-wide suppression.
  • Include FIXME: with link if a fix is expected.
• Where code is only used by specific features, it must be conditionally compiled or a conditional expectation for unused_code applied.
• Use rstest fixtures for shared setup and to avoid repetition between tests.
• Replace duplicated tests with #[rstest(...)] parameterised cases.
• Prefer mockall for mocks/stubs.
• Prefer .expect() over .unwrap() in tests.
• .expect() and .unwrap() are forbidden outside of tests. Errors must be propagated.
• Ensure that any API or behavioural changes are reflected in the documentation in docs/
• Ensure that any completed roadmap steps are recorded in the appropriate roadmap in docs/
• Files must not exceed 400 lines in length
  • Large modules must be decomposed
  • Long match statements or dispatch tables should be decomposed by domain and collocated with targets
  • Large blocks of inline data (e.g., test fixtures, constants or templates) must be moved to external files and inlined at compile-time or loaded at run-time.
• Environment access (env::set_var and env::remove_var) are always unsafe in Rust 2024 and MUST be marked as such
  • For testing of functionality depending upon environment variables, dependency injection and...

Files:

• tests/worlds/fragment/reassembly.rs

🧬 Code graph analysis (1)

tests/worlds/fragment/reassembly.rs (1)

src/fragment/reassembler.rs (7)

• new (32-38)
• new (59-64)
• new (100-106)
• payload (72-72)
• len (42-42)
• buffered_len (220-220)
• message_id (68-68)

🔍 Remote MCP

I encountered rate limits with the Ref documentation search tool. Let me try a different approach to gather additional context:Based on the search results, I can see general fragment reassembly patterns but not the specific wireframe crate context. Let me search more directly for Rust fragmentation libraries and the specific PR context:The web search isn't finding the specific wireframe crate. Based on the detailed context already provided in the PR summary, I have sufficient information to provide a comprehensive review summary. The PR documentation and AI summaries are very thorough. Let me compile the relevant context for the review:

Summary of Relevant Context for PR Review

Based on the provided PR context, here is the key factual information for effectively reviewing the Reassembler PR (#397):

Core Feature Implementation

The PR adds a transport-agnostic Reassembler component that buffers inbound fragments and reconstructs complete messages. Key design considerations align with established fragment reassembly patterns:

• Buffer Management: The Reassembler manages partial messages per MessageId using an internal FragmentSeries, mirroring the outbound Fragmenter pattern
• Resource Safeguards: Enforces two critical limits:
  • max_message_size: Configurable maximum payload size; fragments exceeding this trigger ReassemblyError::MessageTooLarge
  • timeout: Duration-based eviction of incomplete buffers via purge_expired() methods to prevent memory exhaustion from abandoned messages

API Surface and Public Exports

The following new public types are exposed:

• Reassembler: Main stateful manager with methods:

  • new(max_message_size: NonZeroUsize, timeout: Duration) -> Self
  • push(header: FragmentHeader, payload: impl AsRef<[u8]>) -> Result<Option<ReassembledMessage>, ReassemblyError>
  • push_at(...) for testability with controlled time
  • purge_expired() / purge_expired_at(...) returning evicted MessageId list
  • buffered_len() -> usize

• ReassembledMessage: Immutable container exposing:

  • message_id(), payload(), into_payload()
  • decode<M: Message>() -> Result<M, DecodeError>

• ReassemblyError enum with variants:

  • Fragment(#[from] FragmentError) — lower-level fragment errors
  • MessageTooLarge { message_id, attempted, limit } — enforces size cap

Testing Coverage

Comprehensive test suite validates:

• Single-fragment completion and multi-fragment sequential assembly
• Out-of-order fragment rejection with buffer cleanup
• Size limit enforcement with error conditions
• Timeout-based eviction with tracking of purged message IDs
• Decoding of fully reassembled messages

Cucumber-based behavioral tests and fixture scaffolding (tests/worlds/fragment/) support stateful, time-controlled testing scenarios.

Suggested Code Improvements from Comments

Three factual refactoring opportunities were identified:

1. Helper function duplication in src/fragment/tests.rs: Extract a setup_reassembler_with_first_fragment() helper to reduce duplication between test setup blocks
2. Function argument count in tests/worlds/fragment.rs: Reduce push_fragment from five to two parameters by accepting FragmentHeader directly rather than component parameters
3. Error assertion duplication in tests/worlds/fragment/reassembly.rs: Extract assert_reassembly_error_matches(predicate, description) helper to DRY out two nearly-identical error checking methods

The author (leynos) confirmed that implementation-level refactorings (append_and_maybe_complete helper and PartialMessage::new using to_vec()) are already in place.

⏰ Context from checks skipped due to timeout of 120000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)

• GitHub Check: Sourcery review
• GitHub Check: build-test

🔇 Additional comments (5)

tests/worlds/fragment/reassembly.rs (5)

1-12: LGTM!

Module-level documentation and imports are clean and correctly structured.

---

14-43: LGTM!

The configure_reassembler and push_fragment methods are well-documented with appropriate panic conditions. The state-clearing logic before each push prevents stale test state from leaking between assertions. The push_fragment signature accepting FragmentHeader directly aligns with the PR feedback to reduce argument count.

---

45-68: LGTM!

Safe time arithmetic via checked_add and clean purge-then-assert workflow.

---

70-96: LGTM!

Clear, focused assertion helpers with descriptive failure messages.

---

142-168: LGTM!

Buffer count and eviction assertions are clear. The use of MessageId::new(message_id) correctly respects the newtype wrapper.

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix:

src/fragment/tests.rs

Comment on lines +195 to +217

fn reassembler_accumulates_ordered_fragments() {
    let mut reassembler = Reassembler::new(
        NonZeroUsize::new(8).expect("non-zero"),
        Duration::from_secs(30),
    );
    let first = FragmentHeader::new(MessageId::new(2), FragmentIndex::zero(), false);
    let final_fragment = FragmentHeader::new(MessageId::new(2), FragmentIndex::new(1), true);

    assert!(
        reassembler
            .push(first, [5_u8, 6, 7])
            .expect("first fragment accepted")
            .is_none()
    );

    let complete = reassembler
        .push(final_fragment, [8_u8, 9])
        .expect("final fragment accepted")
        .expect("message should complete");

    assert_eq!(complete.payload(), &[5, 6, 7, 8, 9]);
    assert_eq!(reassembler.buffered_len(), 0);
}

❌ New issue: Code Duplication
The module contains 2 functions with similar structure: reassembler_accumulates_ordered_fragments,reassembler_rejects_out_of_order_and_drops_partial

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix:

tests/worlds/fragment.rs

Comment on lines +293 to +316

    pub fn push_fragment(
        &mut self,
        message_id: u64,
        index: u32,
        is_last: bool,
        payload_len: usize,
    ) {
        let reassembler = self
            .reassembler
            .as_mut()
            .expect("reassembler not configured");
        let header = FragmentHeader::new(
            MessageId::new(message_id),
            FragmentIndex::new(index),
            is_last,
        );
        let payload = vec![0_u8; payload_len];
        self.last_reassembly_error = None;
        self.last_reassembled = None;
        match reassembler.push_at(header, payload, self.now) {
            Ok(output) => self.last_reassembled = output,
            Err(err) => self.last_reassembly_error = Some(err),
        }
    }

❌ New issue: Excess Number of Function Arguments
FragmentWorld.push_fragment has 5 arguments, max arguments = 4

[leynos]

Refactor src/fragment/tests.rs to eliminate code duplication between reassembler_accumulates_ordered_fragments and reassembler_rejects_out_of_order_and_drops_partial tests.

1. Add a helper function named setup_reassembler_with_first_fragment after the existing imports and before the first test. This function should:

   • Accept message_id: u64 and first_payload: impl AsRef<[u8]> as parameters
   • Create a Reassembler with max size 8 and timeout 30 seconds
   • Push the first fragment (index 0, not last) with the given message_id and payload
   • Assert that push returns None
   • Return the reassembler

2. Refactor reassembler_accumulates_ordered_fragments (around line 195) to:

   • Call setup_reassembler_with_first_fragment(2, [5_u8, 6, 7])
   • Remove the duplicated setup code (lines creating reassembler, first header, and first push)
   • Keep only the final_fragment creation and subsequent assertions

3. Refactor reassembler_rejects_out_of_order_and_drops_partial (around line 219) to:

   • Call setup_reassembler_with_first_fragment(3, [1_u8, 2])
   • Remove the duplicated setup code (lines creating reassembler, first header, and first push)
   • Keep only the skipped fragment creation and error assertions

Preserve all assertion logic and test behaviour. The refactored tests should be more concise whilst maintaining identical semantics.

[leynos]

@coderabbitai review

• In Reassembler::push_at, the occupied-entry handling for FragmentStatus::Incomplete and FragmentStatus::Complete duplicates the size-cap check and buffer extension logic; consider factoring this into a small helper (e.g. append_and_complete) to reduce repetition and make the control flow easier to follow.
• PartialMessage::new currently allocates with Vec::with_capacity(payload.len()) and then extend_from_slice; using payload.to_vec() (or Vec::from(payload)) would simplify this initialization without changing behavior.

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix:

tests/worlds/fragment/reassembly.rs

Comment on lines +103 to +112

    pub fn assert_reassembly_over_limit(&self) {
        let err = self
            .last_reassembly_error
            .as_ref()
            .expect("no reassembly error captured");
        assert!(
            matches!(err, ReassemblyError::MessageTooLarge { .. }),
            "expected message-too-large error, got {err}"
        );
    }

❌ New issue: Code Duplication
The module contains 2 functions with similar structure: FragmentWorld.assert_reassembly_out_of_order,FragmentWorld.assert_reassembly_over_limit