Implement outbound streaming send_streaming with tests (8.4.1–8.4.5)#485
Merged
Implement outbound streaming send_streaming with tests (8.4.1–8.4.5)#485
Conversation
Contributor
Reviewer's GuideImplements a new outbound streaming helper WireframeClient::send_streaming with configurable chunking and timeout, plus tests, BDD scaffolding, and docs/ADR/roadmap updates to integrate it into the client API and testing strategy. Sequence diagram for WireframeClient::send_streaming with timeout and error hooksequenceDiagram
actor Caller
participant WireframeClient
participant BodyReader
participant TokioTimeout
participant TransportFramed
participant ErrorHook
Caller->>WireframeClient: send_streaming(frame_header, body_reader, config)
WireframeClient->>WireframeClient: effective_chunk_size(header_len, max_frame_length, config)
alt invalid_header_or_config
WireframeClient-->>Caller: Err(ClientError)
else valid_config
alt config.timeout is Some
WireframeClient->>TokioTimeout: timeout(duration, send_streaming_inner(...))
note over TokioTimeout,WireframeClient: Tokio runs send_streaming_inner under timeout
else no_timeout
WireframeClient->>WireframeClient: send_streaming_inner(frame_header, body_reader, chunk_size)
end
loop for_each_frame
WireframeClient->>BodyReader: read(chunk_size)
alt read_ok_and_n_gt_0
WireframeClient->>TransportFramed: send(Bytes{header + chunk})
alt transport_error
TransportFramed-->>WireframeClient: Err(io_error)
WireframeClient->>ErrorHook: invoke_error_hook(ClientError)
WireframeClient-->>Caller: Err(ClientError)
else transport_ok
TransportFramed-->>WireframeClient: Ok(())
end
else read_eof
WireframeClient-->>Caller: Ok(SendStreamingOutcome{frames_sent})
else read_error
BodyReader-->>WireframeClient: Err(io_error)
WireframeClient->>ErrorHook: invoke_error_hook(ClientError)
WireframeClient-->>Caller: Err(ClientError)
end
end
opt timeout_configured
alt timeout_elapsed_before_completion
TokioTimeout-->>WireframeClient: Elapsed
WireframeClient->>ErrorHook: invoke_error_hook(ClientError{TimedOut})
WireframeClient-->>Caller: Err(ClientError{TimedOut})
else completed_before_timeout
TokioTimeout-->>WireframeClient: Ok(Ok(SendStreamingOutcome))
WireframeClient-->>Caller: Ok(SendStreamingOutcome)
end
end
end
Class diagram for SendStreamingConfig, SendStreamingOutcome, and WireframeClient::send_streamingclassDiagram
class WireframeClient {
+send_streaming(frame_header: slice_u8, body_reader: AsyncRead, config: SendStreamingConfig) Result~SendStreamingOutcome, ClientError~
-send_streaming_inner(frame_header: slice_u8, body_reader: AsyncRead, chunk_size: usize) Result~SendStreamingOutcome, ClientError~
-invoke_error_hook(error: ClientError) void
-codec_config: ClientCodecConfig
}
class SendStreamingConfig {
-chunk_size: Option~usize~
-timeout: Option~Duration~
+with_chunk_size(size: usize) SendStreamingConfig
+with_timeout(duration: Duration) SendStreamingConfig
+chunk_size() Option~usize~
+timeout() Option~Duration~
}
class SendStreamingOutcome {
-frames_sent: u64
+new(frames_sent: u64) SendStreamingOutcome
+frames_sent() u64
}
class ClientCodecConfig {
+max_frame_length_value() usize
}
class ClientError
class SendStreamingModule {
+effective_chunk_size(header_len: usize, max_frame_length: usize, config: SendStreamingConfig) Result~usize, ClientError~
}
WireframeClient --> SendStreamingConfig : uses
WireframeClient --> SendStreamingOutcome : returns
WireframeClient --> ClientCodecConfig : reads_max_frame_length
WireframeClient --> ClientError : error_type
SendStreamingModule --> SendStreamingConfig : reads
SendStreamingModule --> ClientError : returns
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Contributor
Summary by CodeRabbit
WalkthroughAdd an outbound transport streaming helper: public Changes
Sequence Diagram(s)sequenceDiagram
participant Client as WireframeClient
participant SendStream as send_streaming
participant Reader as AsyncRead\n(Body)
participant Framing as FrameConstructor
participant Transport as ClientStream\n(Network)
Client->>SendStream: send_streaming(header, reader, config)
Note over SendStream: Start overall timeout (if configured)
loop read until EOF or error
SendStream->>Reader: read(effective_chunk_size)
alt read timed out
Reader-->>SendStream: TimedOut
SendStream->>SendStream: Abort, do not emit partial frame
SendStream-->>Client: ClientError::Wireframe(WireframeError::Io(TimedOut))
else read ok
Reader-->>SendStream: chunk_bytes
SendStream->>Framing: prepend header -> framed_payload
Framing-->>SendStream: framed_payload
SendStream->>Transport: send(framed_payload)
alt transport error
Transport-->>SendStream: Io error
SendStream-->>Client: ClientError (frames_sent so far)
else success
Transport-->>SendStream: OK
SendStream->>SendStream: increment frames_sent
end
end
end
SendStream-->>Client: SendStreamingOutcome { frames_sent }
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![codescene-delta-analysis[bot]](https://avatars.githubusercontent.com/in/53921?s=60&v=4){kind=small}
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
Owner
Author
|
@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/steps/client_send_streaming_steps.rs Comment on file fn given_receiving_server(
client_send_streaming_world: &mut ClientSendStreamingWorld,
) -> TestResult {
client_send_streaming_world.block_on(|w| {
Box::pin(async {
w.start_receiving_server().await?;
w.connect_client().await
})
})?
}❌ New issue: Code Duplication |
Owner
Author
|
@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/client/tests/send_streaming_infra.rs Comment on file pub(super) async fn create_send_client(
addr: SocketAddr,
) -> Result<
WireframeClient<
crate::serializer::BincodeSerializer,
crate::rewind_stream::RewindStream<tokio::net::TcpStream>,
>,
Box<dyn std::error::Error + Send + Sync>,
> {
Ok(WireframeClient::builder().connect(addr).await?)
}❌ New issue: Code Duplication |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Owner
Author
|
@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/client/tests/send_streaming.rs Comment on file async fn exact_chunk_boundary_produces_single_frame(protocol_header: Vec<u8>) -> TestResult {
let (server, frames) = spawn_receiving_server().await?;
let mut client = create_send_client(server.addr).await?;
let body = test_body(100);
let config = SendStreamingConfig::default().with_chunk_size(100);
let outcome = client
.send_streaming(&protocol_header, &body[..], config)
.await?;
if outcome.frames_sent() != 1 {
return Err(format!(
"exactly chunk_size bytes should produce 1 frame, got {}",
outcome.frames_sent()
)
.into());
}
drop(client);
tokio::time::sleep(Duration::from_millis(50)).await;
let received = frames.lock().await;
if received.len() != 1 {
return Err(format!("expected 1 frame, got {}", received.len()).into());
}
Ok(())
}❌ New issue: Code Duplication |
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
leynos
changed the title
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 9248745288
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Contributor
There was a problem hiding this comment.
Actionable comments posted: 9
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/adr-002-streaming-requests-and-shared-message-assembly.md`:
- Around line 369-370: Update the doc text about SendStreamingOutcome to remove
the claim that it reports progress on failure; instead state that
SendStreamingOutcome reports frames_sent: u64 only on successful completion, and
that callers cannot rely on this type for failure-path progress visibility.
Replace the existing sentence referencing `SendStreamingOutcome` and
`frames_sent: u64` with this corrected wording.
In `@docs/execplans/8-4-1-send-streaming-helper.md`:
- Around line 389-392: Update the test instruction for
send_streaming_timeout_returns_timed_out to match the working approach: remove
the guidance to use tokio::time::pause() and instead instruct creating an
AsyncRead that blocks indefinitely (e.g., a tokio::io::duplex where no data is
ever written or the writer is dropped without writing), set a short timeout, and
assert the resulting error is io::ErrorKind::TimedOut; reference the test name
send_streaming_timeout_returns_timed_out and the types tokio::io::duplex,
AsyncRead, and io::ErrorKind::TimedOut so readers can find and implement the
proven pattern.
- Around line 119-224: Update the document to use the repository-mandated Oxford
en-GB spelling (en-GB-oxendict): replace all instances of "finalise" →
"finalize", "serialisation"/"serialised"/"serialisation" variants →
"serialization"/"serialized", and any other -ise suffixes to -ize forms (e.g.,
"realise"→"realize", "organisation"→"organization"); scan the file for these
tokens and update them consistently (headings, paragraphs, decision log entries,
ADR wording, and glossary-like phrases such as "serialisation layer") so all
occurrences in this document use the -ize spelling.
In `@src/client/send_streaming.rs`:
- Around line 186-190: The call to effective_chunk_size should handle Err
explicitly so the preflight error hook is invoked before returning: replace the
direct ?-propagation of effective_chunk_size(frame_header.len(),
self.codec_config.max_frame_length_value(), &config) with a match (or if let
Err) that on Err calls the configured preflight error hook (e.g.,
self.error_hook.preflight_failed(&err) or whatever the hook method is) and then
returns the error; on Ok continue and bind effective_chunk_size to the success
value.
- Around line 276-278: The match on config.chunk_size currently allows a
requested chunk size of 0 which leads to silent success; modify the logic around
the match on config.chunk_size (the branch that returns
requested.min(available)) to treat Some(0) as invalid: if requested == 0 return
an Err (or propagate an error) indicating an invalid chunk size instead of
returning 0, otherwise use requested.min(available); also add a validation in
the with_chunk_size setter or caller to reject zero up-front so callers cannot
set Some(0).
In `@src/client/tests/send_streaming_infra.rs`:
- Around line 78-85: The test currently drops the accepted socket (tcp) which
performs a graceful FIN close; instead call shutdown on the socket's write
direction to force immediate write failure: after accepting the socket in the
spawned task from listener.accept() call tcp.shutdown(std::net::Shutdown::Write)
(or the async equivalent) before dropping tcp so the client sees a deterministic
transport error; also remove the defensive
tokio::time::sleep(Duration::from_millis(50)) in
invokes_error_hook_on_transport_failure since shutdown makes the error occur on
the first write.
In `@tests/features/client_send_streaming.feature`:
- Line 18: Update the scenario title text from "Client send times out on a slow
body reader" to "Client send operation times out on a slow body reader" in the
feature file and make the matching change in the test definition in
client_send_streaming_scenarios.rs so the scenario name string/identifier in the
Rust test matches the new phrasing exactly.
In `@tests/fixtures/client_send_streaming.rs`:
- Around line 143-145: The Given step set_blocking_reader sets
use_blocking_reader but do_send_streaming_with_timeout doesn't currently honor
it; modify do_send_streaming_with_timeout to check the use_blocking_reader flag
at the start and fail fast (return an immediate error/result) if it's not true,
then proceed with the blocking-reader behavior only when true and reset
use_blocking_reader to false after consuming it; update the equivalent logic
referenced around the other Given step (lines 167-174) so both call sites read
and clear use_blocking_reader the same way.
In `@tests/steps/client_send_streaming_steps.rs`:
- Around line 47-55: Remove the #[expect(clippy::unnecessary_wraps,...)]
attribute and change the step function given_blocking_reader to be infallible:
update its signature to return () (i.e., no -> TestResult) and remove the Ok(())
return so the body simply calls
client_send_streaming_world.set_blocking_reader(); ensuring the function matches
other infallible step examples like then_end.
ℹ️ Review info
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (16)
docs/adr-002-streaming-requests-and-shared-message-assembly.mddocs/execplans/8-4-1-send-streaming-helper.mddocs/roadmap.mddocs/users-guide.mdsrc/client/mod.rssrc/client/send_streaming.rssrc/client/tests/mod.rssrc/client/tests/send_streaming.rssrc/client/tests/send_streaming_infra.rstests/features/client_send_streaming.featuretests/fixtures/client_send_streaming.rstests/fixtures/mod.rstests/scenarios/client_send_streaming_scenarios.rstests/scenarios/mod.rstests/steps/client_send_streaming_steps.rstests/steps/mod.rs
Introduce `send_streaming` method on `WireframeClient` to support sending large request bodies as multiple framed chunks. This method reads from an async source, prepends a protocol-defined header to each chunk, and sends framed packets over the transport. Chunk size can be configured or auto-derived from the max frame length minus header size. The entire operation can be wrapped with a configurable timeout that returns a `TimedOut` error without emitting partial frames. Integrate error hook invocation on all failure paths. Return `SendStreamingOutcome` reporting frames sent for partial success awareness. Add comprehensive unit and BDD tests covering chunking, timeout, error handling, and frame content correctness. Update documentation (ADR 0002, users-guide, roadmap) to describe new functionality and mark roadmap items 8.4.1-8.4.5 as complete. Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>
…server startup - Introduce `test_frames_sent_for_body` helper to verify frame counts sent and received. - Refactor existing tests to use this helper, reducing duplication. - Update test infrastructure to use common max frame size. - Abstract server start-and-connect logic in BDD steps to a helper fn. - Simplify BDD step implementations using the new helper. - Clean up documentation wording on async server task design and clippy lint usage. Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>
devboxerhub
Bot
force-pushed
the
implement-send-streaming-helper-c91skz
branch
from
9248745 to
6b0c2cb
Compare
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
…st coverage - Add proper error hook invocation on chunk size calculation failure. - Shutdown write side on timeout to prevent codec state leakage. - Refactor chunk read logic to robustly handle EOF and errors. - Add validation rejecting zero chunk sizes. - Create a blocking_reader helper to reliably test timeouts. - Update tests to use blocking_reader for deterministic behavior. - Strengthen tests for error conditions and chunk/frame correctness. - Improve documentation and clarify send_streaming outcome semantics. - Minor docs and test scenario wording improvements for clarity. Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (1)
docs/execplans/8-4-1-send-streaming-helper.md (1)
102-105:⚠️ Potential issue | 🟡 MinorUpdate the timeout-risk mitigation to match the implemented method.
Replace this mitigation text. Line 103 still prescribes
tokio::time::pause(), but the same document records that this approach failed and was replaced.🛠️ Proposed fix
-- Risk: unit tests for timeout behaviour can be flaky with real time. - Severity: medium Likelihood: medium Mitigation: use `tokio::time::pause()` - for deterministic virtual time in timeout tests. +- Risk: unit tests for timeout behaviour can be flaky with real time. + Severity: medium Likelihood: medium Mitigation: use an `mpsc`-backed + blocking `AsyncRead` (for example `tokio_util::io::StreamReader`) with a + short wall-clock timeout, and avoid `tokio::time::pause()` with + `tokio::io::duplex` for this path.As per coding guidelines, "Use markdown files within the
docs/directory as a knowledge base and source of truth for project requirements, dependency choices, and architectural decisions... Ensure documentation remains accurate and current."🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@docs/execplans/8-4-1-send-streaming-helper.md` around lines 102 - 105, Replace the incorrect mitigation that recommends tokio::time::pause() with the actual deterministic virtual-time approach used in our tests: remove the explicit tokio::time::pause() suggestion and instead describe the project's implemented helper/strategy for virtualizing time (name that helper or test utility used in the codebase) so the mitigation matches reality and points readers to the correct test helper/implementation.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/client/tests/send_streaming.rs`:
- Around line 54-55: The test uses fixed sleeps (drop(client);
tokio::time::sleep(Duration::from_millis(50)).await;) which causes timing races;
replace each fixed sleep with an explicit readiness check or awaiting the
server/test task instead — for example, await a server handle future
(server_handle.await) or poll a shared counter/AtomicUsize or oneshot receiver
in a loop with tokio::time::timeout until the expected frame count is reached,
then assert; update the occurrences around drop(client) and the other locations
(near lines 98–99, 157–158, 205–206, 288–289) to use the same condition-based
synchronization instead of sleep.
In `@tests/fixtures/client_send_streaming.rs`:
- Around line 156-158: The test duplicates the cyclic byte generation used by
the existing helper test_body
(src/client/tests/send_streaming_infra.rs::test_body); replace the inline body
creation (the local body: Vec<u8> computed from 0..body_size) with a call to the
shared helper or move the helper into a common test util and call that from
tests/fixtures/client_send_streaming.rs so both tests reuse the same function
(reference test_body) to remove duplication.
---
Duplicate comments:
In `@docs/execplans/8-4-1-send-streaming-helper.md`:
- Around line 102-105: Replace the incorrect mitigation that recommends
tokio::time::pause() with the actual deterministic virtual-time approach used in
our tests: remove the explicit tokio::time::pause() suggestion and instead
describe the project's implemented helper/strategy for virtualizing time (name
that helper or test utility used in the codebase) so the mitigation matches
reality and points readers to the correct test helper/implementation.
ℹ️ Review info
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (10)
docs/adr-002-streaming-requests-and-shared-message-assembly.mddocs/execplans/8-4-1-send-streaming-helper.mddocs/users-guide.mdsrc/client/send_streaming.rssrc/client/tests/send_streaming.rssrc/client/tests/send_streaming_infra.rstests/features/client_send_streaming.featuretests/fixtures/client_send_streaming.rstests/scenarios/client_send_streaming_scenarios.rstests/steps/client_send_streaming_steps.rs
Owner
Author
|
@coderabbitai Have the following now been resolved? |
Contributor
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: Length of output: 1840 🏁 Script executed: Length of output: 86 🏁 Script executed: Length of output: 548 🏁 Script executed: Length of output: 789 🏁 Script executed: Length of output: 1464 🏁 Script executed: Length of output: 586 🏁 Script executed: Length of output: 407 🏁 Script executed: Length of output: 1860 🏁 Script executed: Length of output: 829 🏁 Script executed: Length of output: 1512 🏁 Script executed: Length of output: 748 🏁 Script executed: Length of output: 3323 🏁 Script executed: Length of output: 5578 🏁 Script executed: Length of output: 2151
Overall Comments — ✅ All Resolved
Individual Comments — ✅ All Resolved
All issues have been addressed. The implementation is in good shape! 🎉 |
Owner
Author
|
@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/client/tests/send_streaming.rs Comment on file async fn rejects_oversized_header() -> TestResult {
let (server, _frames) = spawn_receiving_server().await?;
let mut client = create_send_client(server.addr).await?;
let header = vec![0u8; DEFAULT_MAX_FRAME];
let config = SendStreamingConfig::default();
assert_io_error(
client
.send_streaming(&header, b"hello" as &[u8], config)
.await,
io::ErrorKind::InvalidInput,
)
}❌ New issue: Code Duplication |
Owner
Author
|
@coderabbitai Have the following now been resolved? |
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
This comment was marked as resolved.
…for determinism - Refactor `spawn_receiving_server` to return a `ReceivingServer` struct with an async `collect_frames` method, eliminating mutex-based frame sharing and avoiding fixed sleeps in tests. - Update send streaming tests to await server frame collection instead of sleeping for fixed durations. - Update tests to use the new API and remove unnecessary sleeps. - Add a helper `test_body` function in the fixtures for deterministic body data. - Update docs to recommend deterministic blocking reader usage for timeout tests instead of unreliable tokio time pausing. These changes improve test reliability and deterministically synchronize client-server interactions in streaming tests. Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>
This comment was marked as outdated.
This comment was marked as outdated.
Sorry, something went wrong.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 6
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/execplans/8-4-1-send-streaming-helper.md`:
- Line 34: Update the first occurrence of "Behavioural tests (`rstest-bdd`
v0.5.0)" to expand the BDD acronym immediately after first use; e.g., change the
sentence to include "(Behaviour-Driven Development, BDD)" right after
"Behavioural" or "Behavioural tests" so the line reads like "Behavioural tests
(Behaviour-Driven Development, BDD) (`rstest-bdd` v0.5.0) validate end-to-end
streaming send scenarios." Ensure the `rstest-bdd` reference and version remain
unchanged and keep the phrasing natural.
- Around line 711-717: Update the "Timeout semantics" section to explicitly
state that timeouts are transport-level write failures and the connection must
be terminated and not reused after a timeout: after the sentence that ends with
"partially successful." add a short sentence referencing
`std::io::ErrorKind::TimedOut` (and ADR 0002 if desired) that says the
connection actor SHOULD terminate the connection and callers must not reuse the
connection once a `TimedOut` error is returned; keep the wording concise and
place it before the "# Errors" subsection.
- Around line 50-52: Consolidate the three repetitive MUST bullets about
send_streaming into fewer, varied sentences: update the three lines referencing
send_streaming so they either combine related constraints (e.g.,
"`send_streaming` MUST NOT emit partial frames on timeout or perform automatic
retries.") and a separate line for timeout behavior ("Timeouts MUST return
`std::io::ErrorKind::TimedOut`."), or rephrase them to vary structure (e.g.,
keep "`send_streaming` MUST NOT emit a partial frame on timeout.", change the
second to "Automatic retries are forbidden.", and the third to "Timeouts MUST
surface as `std::io::ErrorKind::TimedOut`."). Ensure the normative MUST/NOT
language and the symbols `send_streaming` and `std::io::ErrorKind::TimedOut` are
preserved.
In `@src/client/tests/send_streaming_infra.rs`:
- Around line 197-202: The test_body function uses
u8::try_from(...).unwrap_or(0) unnecessarily because i.wrapping_rem(256) is
always 0–255; update the closure in test_body to directly cast the iterator
index to u8 (e.g., |i| i as u8) and remove the try_from/unwrap_or machinery so
the code is simpler and idiomatic.
In `@src/client/tests/send_streaming.rs`:
- Around line 318-343: The test invokes_error_hook_on_transport_failure uses a
fragile tokio::time::sleep(Duration::from_millis(50)) to wait for the server
write-side shutdown; replace this nondeterministic wait with a deterministic
synchronization mechanism: change spawn_dropping_server (or
create_send_client_with_error_hook) to return or expose a readiness/closed
signal (e.g., a oneshot/Notify or a future) that is completed when the server
has accepted the connection and shut down the write side, then await that signal
before calling client.send_streaming(b"\x01", &body[..], config).await so the
test deterministically observes the transport failure instead of sleeping.
In `@tests/fixtures/client_send_streaming.rs`:
- Around line 52-55: The closure in test_body is overcomplicating conversion by
using u8::try_from(...).unwrap_or(0); replace the map body inside test_body (the
closure passed to .map(|i| ...)) with a direct cast to u8 (e.g., map(|i| i as
u8) or map(|i| i.wrapping_rem(256) as u8)) since the value always fits in a u8.
ℹ️ Review info
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (4)
docs/execplans/8-4-1-send-streaming-helper.mdsrc/client/tests/send_streaming.rssrc/client/tests/send_streaming_infra.rstests/fixtures/client_send_streaming.rs
- Refactor send_streaming tests by adding a helper function to assert expected I/O errors, reducing duplication. - Change spawn_dropping_server to return a notification handle allowing deterministic synchronization instead of sleeps. - Update tests to await this notification to improve reliability. - Enhance code comments and documentation for timeout and error behavior, emphasizing that TimedOut errors indicate transport failure and require connection termination. - Minor improvements to BDD test documentation wording and formatting. These changes clarify the error handling contract and improve the stability and maintainability of send_streaming tests. Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>
Owner
Author
|
@coderabbitai Have the following now been resolved? ❌ Remaining Issue (1/11)send_streaming doc comment missing connection termination warning — The doc comment (lines 141-147) mentions timeout semantics and partial success, but does NOT explicitly state that the connection must not be reused after a timeout. The issue requested: "update the send_streaming/docs comment to state that a timeout terminates the connection and it must not be reused." The current doc says:
It should add a sentence warning that the connection is in an indeterminate state and must not be reused. |
Contributor
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: Length of output: 3229
The doc comment (lines 146-148) now explicitly states:
This directly addresses the requirement. The warning uses RFC 2119 keywords (SHOULD, MUST NOT) to make the semantics unambiguous and references the ADR for design rationale. The connection termination requirement is now clearly documented. All 11 issues are now resolved! 🎉 |
…y expects Replaced usages of `u8::try_from(i.wrapping_rem(256)).unwrap_or(0)` with `(i % 256) as u8` in test helpers to generate deterministic test byte patterns. Added `#[expect]` attributes to suppress Clippy lints for integer division remainder and possible truncation, with explanations. This change improves code clarity and explicitly documents intentional use of modulo operations in tests. Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>
Owner
Author
|
|
Owner
Author
|
@coderabbitai review |
Contributor
✅ Actions performedReview triggered.
|
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/execplans/8-4-1-send-streaming-helper.md`:
- Line 260: The phrase "sub-modules" in the document entry for
`src/client/tests/mod.rs` should be changed to the single-word form
"submodules"; update the text that currently reads "`src/client/tests/mod.rs` —
unit test module with sub-modules." to use "submodules" instead to match
standard compound usage.
ℹ️ Review info
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
📒 Files selected for processing (5)
docs/execplans/8-4-1-send-streaming-helper.mdsrc/client/send_streaming.rssrc/client/tests/send_streaming.rssrc/client/tests/send_streaming_infra.rstests/fixtures/client_send_streaming.rs
Co-authored-by: devboxerhub[bot] <devboxerhub[bot]@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/execplans/8-4-1-send-streaming-helper.md`:
- Around line 499-509: The documentation's Stage E quality-gate command list is
missing the required "make nixie" invocation; update the command block(s) that
currently run "make markdownlint" (and the other gates) to also run "make nixie"
so the documented validation flow matches repository checks — specifically add
"make nixie" alongside "make markdownlint" in the Stage E blocks referenced
around the existing commands (the lines invoking make markdownlint /
MDLINT=/root/.bun/bin/markdownlint-cli2) so Mermaid diagram validation is
included.
- Around line 485-493: Update the users guide and Stage D docs to explicitly
document the tuple-based multi-packet streaming API: add a subsection titled
"Outbound streaming sends" that explains when to use send_streaming, shows the
SendStreamingConfig builder API, and details timeout semantics and
partial-success assumptions; additionally, describe the helper constructors that
prepare the channel pair for multi-packet streaming (mention their names and the
tuple shape they produce), include a brief example demonstrating typical usage
with send_streaming + SendStreamingConfig and the helper constructors, and
ensure the Stage D checklist requires this tuple-based coverage so published
docs match repository standards.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Changes
Why this is needed
Roadmap item 8.4.1–8.4.5 introduces outbound streaming for large request bodies. The new helper provides a transport-level, header-aware streaming primitive that reads from an AsyncRead source, chunks data according to protocol budgets, prefixes frames with protocol headers, handles timeouts, and integrates with the existing error hooks. This avoids boilerplate in protocol implementations and aligns outbound streaming with inbound streaming design.
How to test
Notes for reviewers
Migration & compatibility
Examples
◳ Generated by DevBoxer ◰
ℹ️ Tag @devboxerhub to ask questions and address PR feedback
📎 Task: https://www.devboxer.com/task/cf7e7719-c829-4cff-a1e1-38df5af2237f
Summary by Sourcery
Add outbound streaming send support to the client with configuration, tests, and documentation updates.
New Features:
Enhancements:
Documentation:
Tests: