feat(a2a-nats): add client event stream consumer - #330
Merged
Conversation
yordis
commented
Jun 18, 2026
yordis
commented
Member
- message/stream and tasks/resubscribe share the same JetStream-backed event delivery path, so a typed stream that tracks last-delivered sequence belongs in one place rather than duplicated per operation.
Per-task JetStream subscriptions for message/stream and tasks/resubscribe need a shared decoded-event stream that tracks last delivered sequence so reconnect logic can resume without replaying already-acked events. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
PR SummaryMedium Risk Overview
The pull loop tracks a resumption cursor by parsing the stream sequence from Reviewed by Cursor Bugbot for commit d0bd3e1. Bugbot is set up for automated code reviews on this repo. Configure here. |
|
Warning Review limit reached
More reviews will be available in 36 minutes and 55 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the 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 credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan refill rate. 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, the refill rate gradually slows as usage increases. The highest same-day bursts are limited more strictly. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
WalkthroughA new ChangesTypedEventStream JetStream module
Sequence Diagram(s)sequenceDiagram
participant Caller
participant build_event_stream
participant TokioTask
participant JetStreamConsumer
participant TypedEventStream
Caller->>build_event_stream: consumer, last_seq_cell
build_event_stream->>TokioTask: tokio::spawn(async loop)
build_event_stream->>Caller: TypedEventStream { receiver, last_seq }
loop Per JetStream message
TokioTask->>JetStreamConsumer: messages().await → next message
TokioTask->>TokioTask: extract_sequence → update last_seq
TokioTask->>TokioTask: serde_json::from_slice → StreamResponse
TokioTask->>JetStreamConsumer: msg.ack().await
TokioTask->>TypedEventStream: channel.send(Ok(event) | Err(ClientError))
end
Caller->>TypedEventStream: poll_next() → Result<StreamResponse, ClientError>
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
|
Code Coverage SummaryDetailsDiff against mainResults for commit: d0bd3e1 Minimum allowed coverage is ♻️ This comment has been updated with latest results |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
rsworkspace/crates/a2a-nats/src/client/event_stream.rs (1)
146-291: ⚡ Quick winAdd regression tests for ACK failure and receiver-drop during active consumption.
Current tests don’t exercise: (1) ACK failure path, and (2) behavior when downstream receiver is closed while producer task is still consuming. Add both to lock in no-loss semantics around resume and delivery guarantees.
🤖 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 `@rsworkspace/crates/a2a-nats/src/client/event_stream.rs` around lines 146 - 291, Add two new regression test functions to the test module: one that verifies the event stream properly handles ACK failures when attempting to acknowledge consumed messages, and another that tests the behavior when the downstream receiver is dropped or closed while the producer task is actively consuming and sending events. Both tests should use the existing build_event_stream helper and MockJetStreamConsumer to set up the scenario, verify that no messages are lost, and confirm proper error handling or state management in these failure modes. Reference the pattern established by existing tests like stream_yields_error_on_consumer_stream_error and stream_closes_when_sender_dropped to maintain consistency.
🤖 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 `@rsworkspace/crates/a2a-nats/src/client/event_stream.rs`:
- Around line 46-55: The event_stream.rs file is converting typed errors into
strings using `.to_string()` when constructing ClientError::ConsumerSetup (line
46) and ClientError::JetStream (line 54) variants, which loses important error
context and type information. Update the ClientError enum definition to have
these variants accept the actual source error types instead of String fields,
then modify both error creation sites to pass the original error object directly
rather than stringifying it. This preserves the full typed error context for
better structured error handling downstream.
- Around line 61-76: The sequence tracking and acknowledgment handling have
incorrect ordering and error handling that can lose events. The
last_seq_for_task is being updated before the message is acknowledged and
delivered, ACK failures are ignored on the js_msg.ack().await call, and
unbounded_send failures are silently ignored. Reorder the operations so that
last_seq_for_task is only updated after confirming both that the message was
successfully sent downstream through the tx channel AND the ACK was successful.
Check the result of js_msg.ack().await and only proceed to update the sequence
if the ACK succeeds, and verify that both unbounded_send calls succeed before
advancing the resume position. Move the sequence update logic to occur after
both the send and ACK operations have completed successfully.
---
Nitpick comments:
In `@rsworkspace/crates/a2a-nats/src/client/event_stream.rs`:
- Around line 146-291: Add two new regression test functions to the test module:
one that verifies the event stream properly handles ACK failures when attempting
to acknowledge consumed messages, and another that tests the behavior when the
downstream receiver is dropped or closed while the producer task is actively
consuming and sending events. Both tests should use the existing
build_event_stream helper and MockJetStreamConsumer to set up the scenario,
verify that no messages are lost, and confirm proper error handling or state
management in these failure modes. Reference the pattern established by existing
tests like stream_yields_error_on_consumer_stream_error and
stream_closes_when_sender_dropped to maintain consistency.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3d4cf024-4a92-461f-bc3c-330a6df760c4
📒 Files selected for processing (2)
rsworkspace/crates/a2a-nats/src/client/event_stream.rsrsworkspace/crates/a2a-nats/src/client/mod.rs
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
…tream metadata Advancing last_seq and acking before the downstream send could mark events delivered while the receiver was dropped, letting resubscribe skip unprocessed messages. Sourcing the sequence from an optional Nats-Sequence header also missed the actual stream sequence because the EVENTS stream doesn't configure republish, so resume position was effectively never recorded. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 38860c3. Configure here.
Dropping TypedEventStream only closes the receiver but left the spawned JetStream pull loop awaiting the next message indefinitely, holding the consumer open as a zombie task. Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>
This was referenced Jun 19, 2026
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.