fix: preserve valid SSE events before parse errors#504
Conversation
Signed-off-by: Bryan Bednarski <bbednarski@nvidia.com>
github-actions
Bot
added
size:S
WalkthroughSSE decoding now returns per-frame results, allowing valid events to be yielded before a later JSON parsing error. The gateway propagates that error and terminates, with unit and integration tests covering the behavior. ChangesSSE partial decode propagation
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant TcpServer
participant sse_json_stream
participant SseEventDecoder
TcpServer->>sse_json_stream: SSE response with valid and invalid frames
sse_json_stream->>SseEventDecoder: Decode buffered frames
SseEventDecoder-->>sse_json_stream: Valid event
sse_json_stream-->>TcpServer: Yield event data
SseEventDecoder-->>sse_json_stream: JSON decoding error
sse_json_stream-->>TcpServer: Yield error and terminate
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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/core/src/codec/streaming.rs`:
- Around line 139-144: Update push_bytes_results to normalize CRLF sequences
that span successive bytes chunks, preserving a trailing carriage return until
the next append can determine whether it forms CRLF; ensure frame detection
still recognizes a terminator split across calls. Add a regression test that
splits the final CRLF of a CRLFCRLF terminator between two push_bytes_results
calls.
🪄 Autofix (Beta)
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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Enterprise
Run ID: 181e35f1-a989-430f-b80c-379a1f537a33
📒 Files selected for processing (4)
crates/cli/src/gateway/mod.rscrates/cli/tests/coverage/shared/gateway_tests.rscrates/core/src/codec/streaming.rscrates/core/tests/unit/codec/streaming_tests.rs
📜 Review details
⏰ Context from checks skipped due to timeout. (4)
- GitHub Check: Node.js / Test (windows-arm64)
- GitHub Check: Python / Test (windows-arm64)
- GitHub Check: Rust / Test (windows-arm64)
- GitHub Check: Rust / Test (windows-amd64)
🧰 Additional context used
📓 Path-based instructions (15)
**/*.rs
📄 CodeRabbit inference engine (.agents/skills/prepare-pr/SKILL.md)
**/*.rs: Any Rust change must runjust test-rust
Any Rust change must runcargo fmt --all
Any Rust change must runcargo clippy --workspace --all-targets -- -D warnings
**/*.rs: Runcargo fmt --allfor all FFI work since it is Rust work
Runjust test-rustto validate FFI changes
Runcargo clippy --workspace --all-targets -- -D warningsto enforce strict linting on FFI workWhen Rust files changed as part of Go work, also run
cargo fmt --all,just test-rust, andcargo clippy --workspace --all-targets -- -D warnings
**/*.rs: Runcargo fmt --allwhen Rust files are changed as part of Node work
Runcargo clippy --workspace --all-targets -- -D warningswhen Rust files are changed as part of Node work
Runjust test-rustwhen Rust files are changed as part of Node workWhen changing the core Rust runtime or Rust-facing API surface, format Rust code with
cargo fmt(rustfmt defaults), keepcargo clippy -- -D warningsclean, and satisfycargo deny checkperdeny.toml.
**/*.rs: If any Rust code changed, always runjust test-rust.
If any Rust code changed, also runcargo fmt --all.
If any Rust code changed, also runcargo clippy --workspace --all-targets -- -D warnings.
For Rust changes headed for review, runcargo fmt --allandcargo clippy --workspace --all-targets -- -D warningseven if relying on pre-commit.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/cli/tests/coverage/shared/gateway_tests.rscrates/cli/src/gateway/mod.rscrates/core/src/codec/streaming.rs
{crates/core,crates/adaptive}/**/*
📄 CodeRabbit inference engine (.agents/skills/prepare-pr/SKILL.md)
Changes to
crates/coreorcrates/adaptivemust run the full language matrix
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/core/src/codec/streaming.rs
crates/core/**/*.rs
📄 CodeRabbit inference engine (.agents/skills/test-go-binding/SKILL.md)
If the change touched
crates/coreor shared runtime semantics, also usevalidate-changefor broader validation
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/core/src/codec/streaming.rs
**/*.{rs,py}
📄 CodeRabbit inference engine (AGENTS.md)
Follow binding naming conventions in Rust and Python: use
snake_case.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/cli/tests/coverage/shared/gateway_tests.rscrates/cli/src/gateway/mod.rscrates/core/src/codec/streaming.rs
**/*.{rs,py,js,mjs,cjs,ts,tsx}
📄 CodeRabbit inference engine (AGENTS.md)
**/*.{rs,py,js,mjs,cjs,ts,tsx}: UseJson = serde_json::Valuein Rust-facing runtime APIs where the existing code expects JSON payloads.
UseResult<T>withFlowErrorin core runtime paths, and keep errors explicit and binding-appropriate at the wrapper layer.
Keep async behavior on the existing tokio-based model; bindings should preserve callback and future lifetimes rather than blocking or hiding async work unexpectedly.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/cli/tests/coverage/shared/gateway_tests.rscrates/cli/src/gateway/mod.rscrates/core/src/codec/streaming.rs
**/*.{rs,py,go,js,ts,c,h}
📄 CodeRabbit inference engine (CONTRIBUTING.md)
Use language-appropriate naming conventions: Rust
snake_case, C FFI exports prefixednemo_relay_, GoPascalCase, Node.jscamelCase, and Pythonsnake_case.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/cli/tests/coverage/shared/gateway_tests.rscrates/cli/src/gateway/mod.rscrates/core/src/codec/streaming.rs
**/*.{rs,go,js,ts}
📄 CodeRabbit inference engine (CONTRIBUTING.md)
Add the SPDX license header to all Rust, Go, JavaScript, and TypeScript source files using the corresponding
//comment form.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/cli/tests/coverage/shared/gateway_tests.rscrates/cli/src/gateway/mod.rscrates/core/src/codec/streaming.rs
**/*
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
**/*: Format changed files with the language-native formatter before the final lint/test pass.
If dynamic plugin behavior changed, usemaintain-dynamic-pluginsand include the native SDK, worker protocol, Python SDK, docs, packaging, and Codecov surfaces in the validation plan.
If code changes alter APIs, bindings, commands, paths, packaging behavior, observability/adaptive semantics, or documented best practices, update any dependent maintainer or consumer skills in the same branch.
During iteration, preferuv run pre-commit run --files <changed files...>.
Before review or handoff, runuv run pre-commit run --all-files.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/cli/tests/coverage/shared/gateway_tests.rscrates/cli/src/gateway/mod.rscrates/core/src/codec/streaming.rs
crates/{core,adaptive}/**/*
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
If
crates/coreorcrates/adaptivechanged, run the full validation matrix across Rust, Python, Go, and Node.js.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/core/src/codec/streaming.rs
**/*.{rs,py,go,js,ts}
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
If a language surface changed, always run that language's test target even when Rust core did not change.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/cli/tests/coverage/shared/gateway_tests.rscrates/cli/src/gateway/mod.rscrates/core/src/codec/streaming.rs
**/*.{rs,py,js,ts,tsx,go,java,kt,swift}
📄 CodeRabbit inference engine (.agents/skills/add-middleware/SKILL.md)
Add tests covering registration and duplicate names, deregistration and missing names, priority ordering, callback failure policy, scope-local inheritance and cleanup, event payload semantics, immutable mark and scope fields, and parity across affected bindings.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/cli/tests/coverage/shared/gateway_tests.rscrates/cli/src/gateway/mod.rscrates/core/src/codec/streaming.rs
crates/{core,adaptive}/**/*.rs
⚙️ CodeRabbit configuration file
crates/{core,adaptive}/**/*.rs: Review the Rust runtime for async correctness, scope isolation, middleware ordering, and event lifecycle regressions.
Pay close attention to task-local/thread-local scope propagation, callback lifetimes, stream finalization, and root_uuid isolation.
Public API changes should preserve existing behavior unless tests and docs show the intended migration path.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/core/src/codec/streaming.rs
{crates/**/tests/**,python/tests/**,go/nemo_relay/**/*_test.go}
⚙️ CodeRabbit configuration file
{crates/**/tests/**,python/tests/**,go/nemo_relay/**/*_test.go}: Tests should cover the behavior promised by the changed API surface, including error paths and cross-request isolation where relevant.
Prefer assertions on lifecycle events, scope stacks, middleware ordering, and binding parity over shallow smoke tests.
Files:
crates/core/tests/unit/codec/streaming_tests.rscrates/cli/tests/coverage/shared/gateway_tests.rs
{crates/**/src/**/*.rs,python/**/*.py}
📄 CodeRabbit inference engine (.agents/skills/maintain-dynamic-plugins/SKILL.md)
Do not add tests under
src; Rust tests belong in cratetests/trees, and Python SDK tests belong underpython/tests.
Files:
crates/cli/src/gateway/mod.rscrates/core/src/codec/streaming.rs
crates/core/src/{api/**/*.rs,api/runtime/**/*.rs,codec/**/*.rs,json.rs}
📄 CodeRabbit inference engine (.agents/skills/add-binding-feature/SKILL.md)
Implement the new or changed public runtime behavior first in the Rust core, especially under
crates/core/src/api/and related core modules such ascrates/core/src/api/runtime/,crates/core/src/codec/, andcrates/core/src/json.rs.
Files:
crates/core/src/codec/streaming.rs
🔇 Additional comments (5)
crates/core/src/codec/streaming.rs (2)
130-132: LGTM!
131-159: 📐 Maintainability & Code QualityConfirm the required core-runtime validation ran.
Please include successful validation evidence for
cargo fmt --all,cargo clippy --workspace --all-targets -- -D warnings,just test-rust,cargo deny check,validate-change, the full Rust/Python/Go/Node matrix, and final pre-commit.As per coding guidelines, “If the change touched
crates/coreor shared runtime semantics, also usevalidate-changefor broader validation” and changes tocrates/core“must run the full validation matrix across Rust, Python, Go, and Node.js.”Source: Coding guidelines
crates/core/tests/unit/codec/streaming_tests.rs (1)
80-97: LGTM!crates/cli/src/gateway/mod.rs (1)
522-529: LGTM!crates/cli/tests/coverage/shared/gateway_tests.rs (1)
877-914: LGTM!
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## release/0.6 #504 +/- ##
===============================================
+ Coverage 93.31% 93.34% +0.03%
===============================================
Files 313 313
Lines 81604 81623 +19
Branches 100 100
===============================================
+ Hits 76142 76183 +41
+ Misses 5462 5440 -22
... and 8 files with indirect coverage changes Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
Signed-off-by: Bryan Bednarski <bbednarski@nvidia.com>
|
/ok to test ef5913b |
|
/merge |
2 participants
Overview
Preserve valid SSE events parsed before a malformed frame in the same upstream read. This lets Switchyard observe the first successful stream item, commit the provider selection, and avoid fallback redispatch when the later frame fails to parse.
Details
SseEventDecoder::push_bytes_resultsAPI to return successful events and the first later parse error in wire order.push_bytescompatible by collecting the ordered results into its existing return type.Validation:
streaming_never_retries_after_first_item: passed.cargo fmt --all: passed.No breaking changes.
Where should the reviewer start?
Start with
SseEventDecoder::push_bytes_resultsincrates/core/src/codec/streaming.rs, then follow its ordered results intosse_json_streamincrates/cli/src/gateway/mod.rs.Related Issues: (use one of the action keywords Closes / Fixes / Resolves / Relates to)
Summary by CodeRabbit