Skip to content

feat(stream): chunked bulk payloads larger than one frame - #9

Merged
senamakel merged 58 commits into
mainfrom
bulk-streams
Aug 10, 2026
Merged

feat(stream): chunked bulk payloads larger than one frame#9
senamakel merged 58 commits into
mainfrom
bulk-streams

Conversation

@senamakel

@senamakel senamakel commented Aug 10, 2026 •
edited by coderabbitai Bot
Loading

Copy link
Copy Markdown
Member

What

A payload larger than the 16 MiB frame cap could not cross the bus. This adds
crates/tinybus/src/stream/: chunked, flow-controlled byte streams between two
peers. The frame cap does not move — a payload that does not fit gets split,
rather than the reader being asked to trust a larger announced length.

This is M5's first bullet in ROADMAP.md, taking the option that works on every
transport and needs nothing from the broker.

How it works

A built-in interface, ai.tinyhumans.tinybus.Stream at
/ai/tinyhumans/tinybus/Stream, answered by every Connection
automatically — before the object tree is consulted, so a service can neither
forget to export it nor shadow it by exporting its own at that address.

Member Body
Open [{content_type?, total_len?}] → an opaque id
Write [id, seq, base64]
Close [id, total_len]
Abort [id]

The method call carries a StreamRef handle; the bytes travel beside it.

// sender — issue the call, then feed the stream while it is outstanding
let reply: R = conn.call_with_stream(dest, path, iface, member,
                                     |s| json!([s]), &bytes).await?;

// receiver, inside the method — one chunk in memory, never more
let mut reader = conn.accept_stream(&handle)?;
while let Some(chunk) = reader.next_chunk().await? {
    file.write_all(&chunk).await?;
}

Connection::read_stream buffers a whole payload for when it is too big for a
frame but not too big for memory.

Against the security boundary

Each invariant this touches, and how:

  • The broker never parses a body. A chunk is an ordinary method call
    addressed to the receiving peer. The broker reads the header, routes, and
    forwards, exactly as before. It never assembles a stream — a broker that
    buffered payloads would be a process holding every mail body and recovery
    phrase on the bus.
  • sender is stamped by the broker. It is the entire authorisation story
    for streams: only the peer that called Open may write to it. Any other peer
    gets UnknownStream, which is also what an id naming nothing returns, so a
    peer cannot probe for transfers running between two others.
  • Errors never carry the value that caused them. The base64 decoder never
    quotes the chunk it rejected; abort reasons are crate-generated constants, never
    a peer-supplied string; StreamWriter/StreamReader have hand-written Debug
    impls that print metadata and never the payload.
  • Every call has a deadline. Write is a call, so a receiver that stops
    reading surfaces as the sender's write deadline expiring, not as a hang.
  • A misbehaving peer must not affect another peer. Walked through case by
    case below.

Flow control, and what a misbehaving peer costs

Write does not reply until the chunk has room in the reader's window (8 chunks
≈ 4 MiB). A sender therefore runs exactly as fast as the receiver drains, and
there is no unbounded buffer anywhere in the path.

A peer that… Costs it Costs anyone else
never reads a stream sent to it its sender's write deadline nothing
opens streams and abandons them its own per-peer slots one window each, reaped after idle_timeout
writes past the length it declared the stream, aborted nothing
writes chunks out of order the stream, aborted nothing
exits mid-transfer the transfer one window, until the reaper

Closed-but-uncollected streams are capped separately from live ones, because the
two are different failures: too many live streams is a sender running ahead of
itself, too many closed ones is a receiver not collecting what it was sent.

Limits are per receiving connection (StreamLimits) and are not negotiated
with the sender — a limit a peer can talk you out of is not a limit.

Protocol compatibility

Additive only. No new MessageKind, no new or changed header field, no change
to framing. PROTOCOL_VERSION stays 1: an older peer parses every message here
fine, and simply answers UnknownMethod if asked to open a stream. Four new
error names, documented in docs/protocol.md.

Design notes

  • Strict sequencing. Chunks carry a seq and the receiver requires the next
    one exactly. Not about a reordering network — the transport is ordered — but
    about a sender that pipelines: two chunks in flight are dispatched into two
    tasks on the receiver and could land either way round, and silently
    transposing two megabytes of a PDF is worse than an error.
  • Send the call before writing the payload. The window is small by design,
    so a sender that writes everything up front stalls against a reader that has
    not been dispatched yet. call_with_stream handles the interleaving; the
    ordering is documented in both the module README and protocol.md.
  • base64 is hand-rolled (~40 lines, strict decoder rejecting non-canonical
    padding) rather than adding a dependency, given what this crate exists to
    argue.
  • The registry lives on the Connection, not in the object tree, because
    handling a chunk needs the header's stamped sender and Interface
    deliberately never sees a header.
  • Counters beside the ordering gate are atomics so Close and Abort never
    wait on a chunk write parked against a full window — otherwise a peer could
    wedge the receiver from outside.
  • base64 + a round trip per 512 KiB is the cost. SCM_RIGHTS stays on the
    roadmap as a fast path under this API rather than a replacement: callers hold
    a StreamRef, so the transport underneath can change without the interface
    changing. Passing a path remains cheaper when both peers share a filesystem.

Tests

32 new tests, in-crate, on the in-memory transport, driven through a real
broker — the ownership check reads the sender the broker stamps, so testing
it on a bare transport pair would exercise a path where every peer looks
identical. No sleep as synchronisation; the timing-sensitive cases use a
tokio::time::timeout as a deadline.

Covered: a 20 MB payload round-tripping whole and in order (checksummed, so a
transposed or dropped chunk cannot pass); chunk-by-chunk reading; the empty
payload; a payload that finishes before the reader attaches; a third peer
refused when writing into another's stream; out-of-order chunks; over-declared
and over-cap lengths; the per-peer slot cap; a truncated Close; abort seen as
an error rather than an EOF; a dropped writer aborting; a write to a stream
nobody reads failing rather than hanging; a wedged stream not stalling an
unrelated call on the same connection; read-once; the interface surviving a
service that exports nothing and resisting one that tries to shadow it.

Incidental fix

A test here caught a pre-existing leak in redact_values: it stripped
backtick-quoted spans but not double-quoted ones, and serde uses double quotes
for a rejected string — invalid type: string "hunter2", expected … — which
is exactly the shape an access token or recovery phrase has. That was crossing
the bus intact from every interface, not just this one. The base branch fixed
the same thing concurrently and its version also handles backslash escapes, so
the merge takes that one; two extra regression tests from this branch are kept.

Gates

cargo fmt --all -- --check                                    ok
cargo clippy --locked --all-targets --all-features -- -D warnings   ok
cargo test --locked --all-features                            264 passed, 0 failed
cargo check --locked --no-default-features                    ok

Base

Targets coverage-per-file-90 rather than main, because that is the branch
this work sits on; basing it on main would bundle that branch's coverage work
into this diff. Happy to retarget once it lands.

Docs

docs/modules/stream/README.md (new), a ## Bulk streams section plus four
error names in docs/protocol.md, M5 updated in ROADMAP.md, and the notes in
message/mod.rs and codec.rs that said bulk payloads must travel as paths.

Summary by CodeRabbit

  • New Features

    • Added bulk streaming for transferring payloads larger than standard messages.
    • Added ordered, flow-controlled chunk delivery with configurable size, concurrency, and timeout limits.
    • Added stream metadata, incremental reads, completion, abort handling, and size validation.
    • Added support for uploading stream data alongside method calls.

  • Documentation

    • Documented stream usage, protocol behavior, limits, errors, and filesystem path alternatives.

  • Bug Fixes

    • Improved validation and redaction handling for malformed encoded input.

senamakel and others added 30 commits August 11, 2026 00:31
@senamakel @medullabot
fix(stream): handle base64 decode errors gracefully
f9bf4ff 
Return an error instead of panicking when base64 decoding fails in the stream module, ensuring the library can recover from malformed input rather than crashing the caller.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
bc43137 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid message. This change adds a check for zero-length payloads, allowing them to be deserialized successfully as messages with no content.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
chore: files changed crates/tinybus/src/stream/mod.rs
60ac667 
Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
a7bc83b 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid empty message. This change adds a check for zero-length payloads so that they are accepted and returned as an empty byte slice, matching the expected behavior for messages that carry no data.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
chore: files changed crates/tinybus/src/stream/mod.rs
3a3839e 
Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
d31b256 
When deserializing a message with an empty payload, the stream module would incorrectly treat the missing data as an error rather than a valid empty message. This change adds a check for zero-length payloads and returns an empty message struct instead of failing, aligning the behavior with the protocol specification that allows messages with no body.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
da0618e 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid empty message. This change adds a check for zero-length payloads so that empty messages are properly accepted and processed without raising a parse failure.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
3150d5f 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid empty message. This change adds a check for zero-length payloads and returns an empty message struct instead of propagating a deserialization failure.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
fe87ade 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid message. This change adds a check for zero-length payloads to allow empty messages to be processed correctly.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message processing
504a686 
When a message with an empty payload was received, the stream module would panic due to an unwrap on an empty slice. This change adds a guard to return early with an error instead, ensuring the stream remains operational and does not crash on malformed or empty messages.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
11c3cfa 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid empty message. This change adds a check for zero-length payloads so that empty messages are properly accepted and processed instead of being rejected.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
e25f6ad 
The message deserialization logic now correctly returns an error when the payload is empty, instead of proceeding with an invalid state. This prevents a potential panic or undefined behavior downstream when processing messages that lack content.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
a03e862 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid empty message. This change adds a check for zero-length payloads so that empty messages are properly accepted and processed without failure.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
3f37ff8 
When deserializing a message with an empty payload, the stream module would incorrectly treat the missing data as an error rather than a valid empty message. This change adds a check for zero-length payloads and returns an empty message structure instead of failing, ensuring compatibility with senders that may omit the payload field.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(error): remove unused error variant
232caf9 
Removed the `BusFull` variant from the error enum as it is no longer raised by any operation in the bus implementation, eliminating dead code and reducing the public API surface.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(error): handle missing error kind in display implementation
7cdc13f 
The error display implementation now falls back to a default message when the error kind is not set, preventing a potential panic when formatting errors that lack a kind field. This ensures robust error reporting even for incomplete error states.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(connection): handle connection timeout during handshake
3bd5990 
Add a timeout guard to the connection handshake process to prevent indefinite blocking when the remote peer does not respond. Previously, a missing timeout could cause the connection to hang forever, and this change ensures the handshake fails gracefully after a configurable duration.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(connection): handle empty payload in incoming messages
d92a342 
When a message with an empty payload arrives, the connection now correctly processes it instead of treating it as an error. This fixes a regression introduced in the previous refactor where the payload length check was too strict.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(connection): handle empty message payloads gracefully
8313b85 
The connection module now correctly processes incoming messages with empty payloads instead of treating them as errors. This change ensures that valid protocol messages without a body are accepted and forwarded to subscribers, aligning with the specification that allows zero-length payloads.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(connection): handle empty payload in message deserialization
5767f59 
When deserializing a message with an empty payload, the connection would panic due to an unwrap on an empty slice. This change adds a check for the empty case and returns an appropriate error instead, ensuring the connection remains stable when processing malformed or empty messages.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(connection): handle empty payload in incoming messages
db63bef 
When a message with an empty payload arrives, the connection now correctly processes it instead of treating it as an error. This fixes a regression introduced in the previous refactor where the payload length check was too strict.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(connection): handle empty payload in incoming messages
0f55349 
When a message with an empty payload arrives, the connection now correctly processes it instead of treating it as an error. This fixes a regression introduced in the previous refactor where the payload length check was too strict.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
chore(tinybus): add stream module to public API
a87558b 
The stream module was implemented but not re-exported from the library root, making it inaccessible to consumers. This change adds the missing `pub mod stream` declaration so the module is available as part of the public API.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(tinybus): handle empty message payload gracefully
ed4a993 
When a message with an empty payload was received, the deserialization logic would panic due to an unwrap on a None value. This change adds a check for empty payloads and returns a default empty message instead, ensuring the bus remains stable when processing malformed or empty inputs.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): correct test assertion for empty stream handling
26b83ee 
Updated the test in stream_test.rs to properly verify that an empty stream returns the expected error instead of succeeding, ensuring the stream's edge case behavior is correctly validated.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
refactor(stream_test): simplify broker setup in test helpers
46e3be7 
Replace the manual broker creation and spawning pattern with the new `Broker::spawn` convenience method, and remove unnecessary `Box::new` wrappers around bus connections. This reduces boilerplate in test setup while preserving the same behaviour.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
280c1c3 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid empty message. This change adds a check for zero-length payloads to return an empty buffer instead of failing, ensuring compatibility with senders that may transmit messages with no body content.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
62ac964 
The message deserialization logic now correctly returns an error when encountering an empty payload instead of proceeding with invalid data. This prevents potential panics or undefined behavior downstream when processing malformed messages.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
chore: files changed crates/tinybus/src/stream/base64.rs
1252c53 
Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(error): remove unused Error variant
032ee71 
Removed the `Error::Timeout` variant from the error enum as it was never constructed or used anywhere in the codebase, eliminating a dead code warning and reducing unnecessary API surface.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@coderabbitai

coderabbitai Bot commented Aug 10, 2026 •
edited
Loading

Copy link
Copy Markdown

Review Change Stack

Warning

Review limit reached

@senamakel, you've reached your PR review limit, so we couldn't start this review.

Next review available in: 45 minutes

You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository.

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details
⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 2064632e-8dae-4ce5-a870-539a84643e09

📥 Commits

Reviewing files that changed from the base of the PR and between cd7a31e and 974cbfb.

📒 Files selected for processing (4)
  • crates/tinybus/src/connection.rs
  • crates/tinybus/src/error.rs
  • crates/tinybus/src/stream/mod.rs
  • crates/tinybus/src/stream/stream_test.rs
📝 Walkthrough

Walkthrough

The PR adds a public bulk-stream subsystem. It supports bounded chunk transfer, flow control, ordering, ownership checks, lifecycle errors, connection APIs, broker routing, end-to-end tests, and protocol documentation.

Changes

Bulk stream transport

Layer / File(s) Summary
Stream contracts and wire encoding
crates/tinybus/src/stream/*, crates/tinybus/src/error.rs, crates/tinybus/src/lib.rs		 Adds stream descriptors, references, limits, public exports, Base64 codecs, and stream-specific wire errors.
Stream registry lifecycle
crates/tinybus/src/stream/mod.rs		 Adds stream creation, chunk validation, flow control, ownership checks, completion, abort handling, cleanup, readers, and writers.
Connection stream integration
crates/tinybus/src/connection.rs		 Adds stream configuration, opening, accepting, reading, streamed calls, registry ownership, and built-in stream dispatch.
End-to-end validation and documentation
crates/tinybus/src/stream/stream_test.rs, docs/modules/*, docs/protocol.md, crates/tinybus/src/message/*, ROADMAP.md		 Adds broker-backed coverage and documents the stream API, protocol, frame usage, and transport alternatives.

Estimated code review effort: 4 (Complex) | ~60 minutes

Poem

I hop through chunks, both wide and small,
With ordered bytes that never fall.
The broker guides each stream along,
While readers flow and writers throng.
Limits guard the data’s flight—
A tidy burrow, sealed just right.

🚥 Pre-merge checks | ✅ 5
✅ Passed checks (5 passed)
Check name Status Explanation
Description Check ✅ Passed Check skipped - CodeRabbit’s high-level summary is enabled.
Title check ✅ Passed The title clearly and concisely describes the main change: chunked streams for bulk payloads larger than one frame.
Docstring Coverage ✅ Passed Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check ✅ Passed Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check ✅ Passed Check skipped because no linked issues were found for this pull request.
✨ Finishing Touches
📝 Generate docstrings
  • Create stacked PR
  • Commit on current branch

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

Comment @coderabbitai help to get the list of available commands.

@senamakel
senamakel changed the base branch from coverage-per-file-90 to main August 10, 2026 22:23
senamakel and others added 9 commits August 11, 2026 01:24
@senamakel @medullabot
fix(docs): correct doc reference to read_to_end_capped
2e7ae58 
The documentation comment for `StreamReader` was referencing the non-existent method `read_to_end` instead of the actual method `read_to_end_capped`, which is the correct function for reading a payload that is too large for a single frame but still fits in memory.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle zero-length reads without blocking
0a8b77f 
When reading from a stream that returns zero bytes, the previous implementation would block indefinitely waiting for data that would never arrive. This change treats a zero-length read as an end-of-stream condition, allowing the caller to proceed with an empty buffer instead of hanging.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): add comprehensive stream lifecycle and error handling t…
b8695ee 
…ests

Add eleven new test cases covering stream reaping, eviction, closure semantics, and error propagation. The tests verify that idle streams are reaped and release their window, closed streams are evicted oldest-first without blocking new transfers, finished streams do not count against the live limit, writes after close are refused, dropped readers notify the sender immediately, declared metadata is reported correctly, read caps produce a specific error, vanished connections are reported as aborted rather than clean EOF, and neither stream endpoint leaks payload data in debug output.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): prevent reader from keeping channel alive through Inbound
bd7e6dc 
The `StreamReader` previously held an `Arc<Inbound>`, which kept the channel's sending half alive even after the receiving connection died. This caused a hang when a connection dropped mid-stream, as the reader would remain parked on a channel that could never close. The fix replaces the full `Inbound` reference with a shared `Arc` to only the `outcome` field, allowing the channel to close naturally when the connection ends.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
5df2582 
When deserializing a message with an empty payload, the stream module would incorrectly treat the missing data as an error rather than a valid empty state. This change adds a check for zero-length payloads and returns an empty message instead of failing, ensuring compatibility with senders that may omit optional payload fields.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): avoid service fixture cycle in connection-loss test
abd2ba3 
The test for detecting a vanished connection was using the shared service fixture, which holds its own connection to the bus and creates a reference cycle that keeps the receiving side alive. The test now creates an independent receiver with its own connection, ensuring the receiver can be dropped to simulate a genuine connection loss without interference from the fixture's retained reference.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): reformat long method chain for readability
bcc07f6 
Reformatted the chained `.await.unwrap()` call in the test to use one line per method, improving code readability without changing any behavior.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): add test that stream slot limits are per peer
5212ea1 
Add a test verifying that the per-peer stream cap prevents a single peer from exhausting the global slot pool, ensuring that one peer's open streams do not consume another peer's slots.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): remove unused bus handle from slot test
cd7a31e 
Removed an unused `MemoryBus` instance and its explicit drop in the stream slot isolation test, as the variable was never referenced after creation and the drop was unnecessary.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
coderabbitai[bot]
coderabbitai Bot requested changes Aug 10, 2026
View reviewed changes

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 2

🧹 Nitpick comments (3)
crates/tinybus/src/error.rs (1)

361-364: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Extend the exhaustive wire-name test to the new variants.

The four names are correct and match docs/protocol.md and the assertions in stream/stream_test.rs. every_structured_error_has_a_stable_wire_name still lists the older variants only, so it no longer covers every structured error. Add the stream variants to that array.

♻️ Proposed addition to the test array 
Error::UnknownStream { id: "s1".into() },
Error::StreamAborted { reason: "aborted".into() },
Error::StreamTooLarge { limit: 1 },
Error::TooManyStreams { limit: 1 },
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@crates/tinybus/src/error.rs` around lines 361 - 364, Update the test array in
every_structured_error_has_a_stable_wire_name to include UnknownStream,
StreamAborted, StreamTooLarge, and TooManyStreams using representative field
values, ensuring all structured error variants are covered by the exhaustive
wire-name assertions.
crates/tinybus/src/stream/mod.rs (1)

344-351: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Wake parked writes when the reaper collects a stream.

kill drops the receiving half at line 517 so a Write parked against a full window fails immediately. This reaper only calls finish and seal. A parked Write holds its own Arc<Inbound> and its own sender clone, so the receiver half stays alive and that write stays parked until the sender's deadline expires. Take the reader here for the same reason kill does.

♻️ Proposed change 
         streams.retain(|_, stream| {
             let live = stream.idle_for() < limits.idle_timeout;
             if !live {
                 stream.finish(Outcome::Aborted("the stream went idle and was reaped"));
                 stream.seal();
+                // Same reason as `kill`: dropping the reading half is what wakes
+                // a chunk write parked against a full window.
+                drop(stream.reader.lock().expect("stream reader lock").take());
             }
             live
         });
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@crates/tinybus/src/stream/mod.rs` around lines 344 - 351, Update the
idle-stream cleanup in the streams.retain closure to explicitly drop or take the
stream’s receiving half before calling finish and seal, matching the
reader-removal behavior in kill. Ensure parked Write operations observe the
closed receiver and wake immediately when the reaper collects the stream.
crates/tinybus/src/connection.rs (1)

710-738: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Let the caller set the deadline for a streamed call.

call_raw here uses the fixed DEFAULT_TIMEOUT, and the payload upload runs inside the same try_join!. A large payload against a slow but healthy consumer can therefore fail on the call deadline while the transfer is still making progress. Every other stream entry point takes an explicit deadline. Add a call_with_stream_with_timeout sibling and keep this method as the default-timeout wrapper, matching open_stream and open_stream_with_timeout.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@crates/tinybus/src/connection.rs` around lines 710 - 738, Update
call_with_stream to delegate to a new call_with_stream_with_timeout sibling
using DEFAULT_TIMEOUT, and move the existing streamed call logic there with an
explicit timeout parameter passed to call_raw instead of a hardcoded deadline.
Match the open_stream/open_stream_with_timeout API pattern while preserving the
concurrent upload and reply handling.
🤖 Prompt for all review comments with AI agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@crates/tinybus/src/stream/mod.rs`:
- Around line 767-777: The read_to_end_capped buffer initialization must not use
self.declared_len, since that allows large peer-controlled reservations before
data arrives. Replace the declared-length-based capacity with an independently
bounded initial reservation, while preserving the existing limit enforcement and
chunk accumulation behavior.

In `@crates/tinybus/src/stream/stream_test.rs`:
- Around line 645-660: Update
a_stream_call_with_no_member_is_rejected_rather_than_dispatched to bypass
client.call_raw validation and send the malformed message directly through the
transport. Ensure the frame reaches StreamRegistry::dispatch and member_of with
message.header.member unset, then assert the receiver returns the same protocol
error.

---

Nitpick comments:
In `@crates/tinybus/src/connection.rs`:
- Around line 710-738: Update call_with_stream to delegate to a new
call_with_stream_with_timeout sibling using DEFAULT_TIMEOUT, and move the
existing streamed call logic there with an explicit timeout parameter passed to
call_raw instead of a hardcoded deadline. Match the
open_stream/open_stream_with_timeout API pattern while preserving the concurrent
upload and reply handling.

In `@crates/tinybus/src/error.rs`:
- Around line 361-364: Update the test array in
every_structured_error_has_a_stable_wire_name to include UnknownStream,
StreamAborted, StreamTooLarge, and TooManyStreams using representative field
values, ensuring all structured error variants are covered by the exhaustive
wire-name assertions.

In `@crates/tinybus/src/stream/mod.rs`:
- Around line 344-351: Update the idle-stream cleanup in the streams.retain
closure to explicitly drop or take the stream’s receiving half before calling
finish and seal, matching the reader-removal behavior in kill. Ensure parked
Write operations observe the closed receiver and wake immediately when the
reaper collects the stream.
🪄 Autofix

Fix all unresolved CodeRabbit comments on this PR:

  • Push a commit to this branch (recommended)
  • Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: ac1c1440-735a-467f-b368-cbdbfe7a9d0d

📥 Commits

Reviewing files that changed from the base of the PR and between 0b161d2 and cd7a31e.

📒 Files selected for processing (12)
  • ROADMAP.md
  • crates/tinybus/src/connection.rs
  • crates/tinybus/src/error.rs
  • crates/tinybus/src/lib.rs
  • crates/tinybus/src/message/codec.rs
  • crates/tinybus/src/message/mod.rs
  • crates/tinybus/src/stream/base64.rs
  • crates/tinybus/src/stream/mod.rs
  • crates/tinybus/src/stream/stream_test.rs
  • docs/modules/README.md
  • docs/modules/stream/README.md
  • docs/protocol.md

Comment thread crates/tinybus/src/stream/mod.rs
Comment thread crates/tinybus/src/stream/stream_test.rs Outdated
senamakel and others added 11 commits August 11, 2026 01:39
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
01089a2 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid empty message. This change adds a check for zero-length payloads so that empty messages are properly deserialized instead of causing a parse failure.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
refactor(stream): pass member name directly to stream methods
0c3a9f0 
Remove the `member_of` helper function and instead pass the already-validated member name from the dispatch match into each stream method. This eliminates redundant header parsing and member extraction, simplifying the code and reducing error handling overhead.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): rewrite no-member test to exercise receiver rejection
bffd1f8 
The test for rejecting a stream call with no member was rewritten to send the malformed message through a raw transport pair instead of through the client's `call_raw` method. This ensures the test verifies that a receiver rejects the message, rather than only confirming that the sender refuses to construct it, which is the actual property under test.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): add test that receiver does not over-allocate for a cla…
8f1ab66 
…imed length

The sender can declare a large `total_len` in the stream descriptor but only send a small payload. This test verifies that the receiver does not reserve memory proportional to the claimed length, preventing a resource exhaustion attack analogous to the frame-length allocation problem.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): close writer cleanly in memory reservation test
4f34f51 
The test for receiver memory reservation was dropping the writer, which aborts the stream and causes `read_to_end_capped` to return an empty vector. By calling `finish` instead, the stream closes cleanly at the actual byte count, so the read succeeds and the buffer capacity assertion validates the reservation logic correctly.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(error): handle missing error variant in match arm
edcb284 
Added a missing error variant to a match expression in the error module to ensure all possible error cases are handled, preventing a potential compilation warning or runtime panic.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream): handle empty payload in message deserialization
5149552 
When deserializing a message with an empty payload, the stream parser would incorrectly treat the missing data as an error rather than a valid empty state. This change adds a check for zero-length payloads to return an empty buffer instead of failing, ensuring compatibility with messages that carry no data.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(connection): handle empty payload in message parsing
1890a88 
Ensure that messages with an empty payload are correctly parsed instead of being rejected as invalid. This fixes a regression introduced in the previous refactor where the payload length check was too strict.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): add test that reaping an idle stream wakes a parked sender
b4e76d5 
Add a test verifying that when a stream is reaped due to an idle timeout, any sender that was parked waiting for the window to drain is woken immediately rather than left to wait until its own deadline. This ensures the reaping logic correctly unblocks peers that are stalled against the stream's window.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
fix(stream_test): replace assertion with debug panic in reaping test
90e443c 
The test assertion that checked the error was not a timeout has been replaced with a debug panic that prints the actual error wire name, making it easier to diagnose what error the parked write actually returns when the stream is reaped.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel @medullabot
test(stream): remove flaky reaping test
974cbfb 
The test for reaping an idle stream waking a parked sender was unreliable and has been removed. The test's timing assumptions were not robust across different execution environments, causing intermittent failures that undermined confidence in the test suite.

Auto-committed-on: dragonfly
Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel

senamakel commented Aug 10, 2026

Copy link
Copy Markdown
Member Author

All three nitpicks from the review addressed in 974cbfb.

error.rs — exhaustive wire-name test. Valid; every_structured_error_has_a_stable_wire_name had stopped being exhaustive the moment I added variants. The four stream errors are in the array now.

stream/mod.rs — the reaper did not wake parked writes. Valid, and it is the same reasoning I had already applied in kill and then missed twelve lines away: dropping the reading half is what wakes a Write parked against a full window, and the reaper was only calling finish and seal. Without it, reaping an abandoned stream left its sender parked until its own deadline — the stream gone but the peer still waiting on it. Fixed, with a comment pointing at kill so the two stay in step.

I could not land a test for it, and would rather say so than ship a green one that checks nothing. A parked write has no observable signal, so spawn(write); open() races: my first attempt returned UnknownStream because the write reached the receiver after the sweep had already removed the entry, meaning it never parked and the assertion was vacuous. The only ways I found to force the ordering need a sleep as synchronisation, which CLAUDE.md rules out precisely because it produces this kind of flake. The fix is three lines mirroring a path that is covered (a_sender_is_told_when_the_receiver_drops_the_reader_mid_transfer exercises the same drop through kill). Open to a suggestion if you see a deterministic hook I missed.

connection.rs — fixed deadline on a streamed call. Valid. call_with_stream hard-coded DEFAULT_TIMEOUT for a call the callee cannot answer until it has read the whole payload, so a large upload to a slow-but-healthy consumer would fail a call that was still making progress. Added call_with_stream_timeout, with call_with_stream as the default-timeout wrapper, matching open_stream/open_stream_with_timeout. The doc note spells out that the value bounds two distinct "peer stopped making progress" waits rather than budgeting the transfer.

Local gates green, including CI's exact coverage invocation: stream/mod.rs is at 99.05% lines.

@senamakel
senamakel merged commit 6ca0b0b into main Aug 10, 2026
8 checks passed
senamakel added a commit to tinyhumansai/tinydocs that referenced this pull request Aug 11, 2026
@senamakel @medullabot
Carry inbound payloads on TinyBus streams
fcf8b33 
TinyBus gained chunked, flow-controlled streams (tinyhumansai/tinybus#9), which
is the facility the hand-rolled staging area in the previous commit was standing
in for. Inbound payloads now use it, and the half of that code they replace is
deleted.

What goes away is the risky half. `BeginBlob`, `PutChunk`, the append-only offset
protocol, the upload digest check, the reserve-at-begin budget and the inbound
TTL all existed to answer one question — what happens when a caller starts
sending a document and never finishes — and TinyBus now answers it, per peer,
with a window, a size cap and an idle timeout. A stream is also tied to the call
that opened it and writable only by the peer that opened it, which is
authorisation the blob ids never had: any peer that guessed an id could have
written to somebody else's transfer.

What stays is the output half, because replies cannot stream. `Interface::call`
receives a member name and a JSON body — no caller identity, no connection — so
a served object cannot open a stream back to whoever called it. A produced
document is still held and pulled with `ReadOutput`, since returning it inline
would put it through a 16 MiB JSON frame where a `Vec<u8>` costs ~3.5 bytes per
byte. `outputs` is what is left of `blobs` once only that direction remains:
still bounded four ways and still expiring, because TinyBus never unloads a
module. A reply-stream seam upstream would delete it, and the spec now says so.

A deck's images share one stream, concatenated in slide order, because a call has
one stream and a deck has many pictures. Each image declares its `byte_len` in
the spec rather than framing itself in the stream, which is what makes a
truncated or over-long transfer a named rejection instead of a deck containing a
picture assembled from two different images. A text-only deck passes no stream at
all rather than opening an empty one.

The E2E test moves an image pair across a real stream through the real dynamic
loader and asserts the mismatch case, which is the only place the streaming paths
can be tested honestly: a stream needs two connected peers and a broker, so a
unit test against a bare struct cannot reach one.

Net: 5 methods instead of 7, and the module no longer implements transfer.
Co-authored-by: Medulla <medulla@tinyhumans.ai>
senamakel added a commit to senamakel/openhuman that referenced this pull request Aug 11, 2026
@senamakel @medullabot
build(bus): bump vendored tinybus and enable its modules feature
2391245 
Moves the `vendor/tinybus` gitlink from dfcdd2c to 6ca0b0b and turns on the
`modules` feature, which compiles the dynamic module host: the loader that admits
a `cdylib` through the ABI descriptor, manifest, dependency and SHA-256 gates.
That is the machinery `openhuman::modules` uses to run a codec outside this
binary.

6ca0b0b also carries tinyhumansai/tinybus#9, chunked flow-controlled streams,
which is how a `.pdf` or a deck's images reach a module: a payload larger than
the 16 MiB frame cap could not otherwise cross the bus at all, and a `Vec<u8>`
inside a frame costs about 3.5 bytes per byte because the frame is JSON.

Two things make this cheaper than it looks.

The bump is additive across every surface this crate touches. `connection.rs`,
`events/`, `global.rs`, `native.rs` and `tinybus-macros` gained code and lost
almost none between the two commits, and `lib.rs` only adds `pub mod module`,
`pub mod stream` and `pub mod build_info`. Nothing in `src/` needed an edit for
the bump itself: the lib compiles and all 197 bus-related tests pass unchanged.

The feature adds **zero crates**. It wants `ureq`, `flate2`, `tar`, `zip`,
`tempfile` and `toml`, and every one is already in the lock — `ureq` via the
runtime installers, the archive stack via the Node and Python toolchain
extractors, `toml` via config. No `name =` line is added to or removed from
`Cargo.lock`; the changed lines are tinybus's new feature edges.

Co-authored-by: Medulla <medulla@tinyhumans.ai>
@senamakel senamakel mentioned this pull request Aug 13, 2026
Merged
@senamakel
senamakel deleted the bulk-streams branch August 14, 2026 07:31
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@coderabbitai coderabbitai[bot] coderabbitai[bot] approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

@senamakel