feat: add internal event bus, stream events via HTTP endpoint - #384
feat: add internal event bus, stream events via HTTP endpoint#384nanderstabel wants to merge 16 commits into
Conversation
|
Warning Review limit reached
Next review available in: 8 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?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 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 configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (4)
📝 WalkthroughWalkthroughChangesThe pull request adds a shared CloudEvents event bus with in-memory fanout, history, CQRS integration, and MongoDB change-stream support. It wires the bus through application and store state, exposes filtered SSE events with catch-up, updates OpenAPI, and migrates test fixtures. CloudEvents event bus
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant EventsHandler
participant EventBus
participant MongoEventSource
participant MongoDB
Client->>EventsHandler: GET /v0/events
EventsHandler->>EventBus: request history and subscribe
MongoEventSource->>MongoDB: watch events collection
MongoDB-->>MongoEventSource: change-stream event
MongoEventSource->>EventBus: publish CloudEvent
EventBus-->>EventsHandler: historical and live events
EventsHandler-->>Client: SSE CloudEvent
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 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 |
…unctions and remove redundant SSE filtering logic
…d remove redundant EventBusPublisher
344bf7e to
3ba4bc4
Compare
…ventPublisher trait and registry pattern
3988e62 to
ae79d88
Compare
nanderstabel
changed the title
/v0/events endpoint
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
agent_application/src/lib.rs (1)
221-282: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winAvoid publishing MongoDB events twice.
The MongoDB path registers
event_bus.query()as every CQRS aggregate event publisher, so each persisted event is dispatched onevent_bus. It then also callsevent_bus.attach_source(MongoEventSource::new(...)), which reads the sameeventscollection via change streams and republishes those inserts onto the same bus. Sincebuild_cloud_eventderives the CloudEvent id fromaggregate_type:aggregate_id:sequence, the SSE/history consumers can receive duplicate events. Switch to one source only: either don’t passevent_bus.query()to the MongoDB builder and rely on the change-stream source, or keep direct dispatch and omitattach_source.🤖 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 `@agent_application/src/lib.rs` around lines 221 - 282, Prevent duplicate MongoDB event publication in the EventStoreType::MongoDb initialization by choosing a single dispatch path: retain either event_bus.attach_source(MongoEventSource::new(...)) or the event_bus.query() publishers passed into the aggregate builders, but not both. Update the MongoDB builder setup and related publisher arguments consistently while preserving event delivery through the selected source.
🧹 Nitpick comments (5)
agent_api_http/src/v0/issuance/credentials.rs (1)
721-722: 🗄️ Data Integrity & Integration | 🔵 Trivial | 🏗️ Heavy liftReuse the caller’s shared event bus in
setup_library_state.This helper creates a new
EventBusHandle, so tests that already share a bus across issuance and authorization still isolate library events on a different channel. Accept&EventBusHandlein the helper and pass that same handle tolibrary_state; update callers to create one bus per application fixture.Suggested fix
- pub async fn setup_library_state(issuance_state: &Arc<IssuanceState>) -> Arc<LibraryState> { + pub async fn setup_library_state( + issuance_state: &Arc<IssuanceState>, + event_bus: &shared_kernel::event_bus::EventBusHandle, + ) -> Arc<LibraryState> { let (projection, view_handle) = CredentialConfigurationProjection::new(issuance_state.clone()); - let event_bus = shared_kernel::event_bus::EventBusHandle::default(); - let lib = Arc::new(library_state(&InMemory, &event_bus, Default::default(), vec![Box::new(projection)]).await); + let lib = Arc::new(library_state(&InMemory, event_bus, Default::default(), vec![Box::new(projection)]).await);🤖 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 `@agent_api_http/src/v0/issuance/credentials.rs` around lines 721 - 722, Update setup_library_state to accept a shared &EventBusHandle parameter and pass it to library_state instead of creating a new EventBusHandle internally. Adjust every caller to create one EventBusHandle per application fixture and reuse it across issuance and authorization setup.agent_application/src/lib.rs (1)
112-113: 🚀 Performance & Scalability | 🔵 TrivialSingle shared broadcast channel (capacity 1024) for the entire application.
All aggregates across all services share one
EventBusHandle. Under sustained load, a slow/disconnected SSE subscriber lagging behind more than 1024 buffered events across every aggregate combined will start losing events (EventBusError::Lagged). Worth monitoring/alerting on lag counts once this ships, and revisiting the capacity if it's tuned only against a single-aggregate mental model.🤖 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 `@agent_application/src/lib.rs` around lines 112 - 113, Update the EventBusHandle initialization in the application setup so the shared broadcast channel’s capacity is explicitly sized for events across all aggregates and services, rather than assuming a single-aggregate workload. Preserve one shared EventBusHandle for the entire application, and add monitoring or alerting for EventBusError::Lagged counts if supported by the existing event-bus integration.shared-kernel/src/event_bus.rs (2)
299-321: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoidable
EventFilterclone per broadcast item per subscriber.
filter.clone()runs for every received broadcast item on every subscriber even thoughEventFilter::matchesonly needs&EventFilter. The match can be computed synchronously in the outer closure (which already ownsfilterviamove) without cloning it into the async block.♻️ Proposed refactor
fn subscribe(&self, filter: EventFilter) -> BusEventStream { let receiver = self.sender.subscribe(); let stream = tokio_stream::wrappers::BroadcastStream::new(receiver).filter_map(move |result| { - let filter = filter.clone(); - async move { - match result { - Ok(event) => { - if filter.matches(&event) { - Some(Ok(event.as_ref().clone())) - } else { - None - } - } - Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(n)) => { - Some(Err(EventBusError::Lagged(n))) - } - } - } + let mapped = match result { + Ok(event) => filter.matches(&event).then(|| Ok(event.as_ref().clone())), + Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(n)) => { + Some(Err(EventBusError::Lagged(n))) + } + }; + async move { mapped } }); Box::pin(stream) }🤖 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 `@shared-kernel/src/event_bus.rs` around lines 299 - 321, Update EventBusHandle::subscribe so the outer filter_map closure computes filter.matches(&event) synchronously while it owns the original filter, instead of cloning EventFilter for each broadcast item and moving a clone into the async block. Preserve the existing event filtering and Lagged error behavior, and keep the stream type compatible with BusEventStream.
164-167: 🔒 Security & Privacy | 🔵 TrivialGlobal fan-out with no built-in scoping.
EventBus/EventBusHandlebroadcast fullCloudEventpayloads (raw domain-event data) to every subscriber, andQuery<A>::dispatchserializes the entire domain event payload onto the bus. Combined with the SSE handler (context snippet) applying only type/source/subject/time filters with no visible authorization check, any subscriber to/v0/eventscan observe every domain event's full payload across every aggregate/tenant. If this data can include sensitive fields (tokens, PII, credential contents), consider adding a scoping/authorization boundary (either in this bus or enforced at the HTTP layer) before this ships broadly.Also applies to: 299-351
🤖 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 `@shared-kernel/src/event_bus.rs` around lines 164 - 167, Add an authorization/scoping boundary to the EventBus/EventBusHandle flow before exposing events through SSE. Ensure Query<A>::dispatch and subscribe enforce tenant/aggregate visibility so subscribers cannot receive unrelated CloudEvent payloads, and have the SSE handler reject or filter unauthorized subscriptions rather than relying only on type/source/subject/time filters.agent_api_http/src/v0/events/mod.rs (1)
127-210: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider adding tests for live push delivery and the "lagged"/"error" SSE branches.
Current tests only assert HTTP status for the route and catch-up cases; the live-subscription push path (event arriving after connection open) and the
lagged/errorevent kinds (lines 114-119) are untested, matching the coverage gap Codecov flagged for this file.🤖 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 `@agent_api_http/src/v0/events/mod.rs` around lines 127 - 210, The tests in test_events_sse_route, test_events_sse_catchup_route, and test_events_sse_timestamp_filter only verify response status; extend the test module to consume SSE response bodies and assert live delivery when an event is published after the connection opens. Add coverage for the stream’s lagged and error branches around the SSE event handling logic, asserting each produces the expected SSE output or termination behavior.
🤖 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 `@agent_api_http/src/v0/events/mod.rs`:
- Around line 86-103: The events_sse_handler flow currently reads history before
subscribing, allowing published events to be missed; subscribe via
event_bus.subscribe before or concurrently with history_ascending, then merge
catch-up and live streams while deduplicating by event ID so overlapping
reconnect events are emitted only once.
In `@agent_store/src/event_source.rs`:
- Around line 21-37: Update MongoDB event streaming in open to support
SubscribePosition::From(_) by applying the supplied resume token to the
change-stream options via the appropriate resume/start-after mechanism, instead
of returning UnsupportedPosition. Ensure the stream exposes and persists each
event’s Position so EventBusHandle::attach_source can pass the last consumed
token when reconnecting, while preserving Live behavior for initial
subscriptions.
- Around line 39-67: Update the change-stream mapping in the `filter_map`
closure to emit a `tracing` warning whenever required fields or payload
deserialization fail before returning `None`, including the affected field and
relevant error details where available. Preserve dropping malformed documents
while keeping the stream open, and extend `occurred_at` parsing in the same
closure to retain timestamps stored as native BSON dates in addition to RFC3339
strings.
In `@shared-kernel/src/event_bus.rs`:
- Around line 232-257: The history_ascending API silently hides gaps when
last_event_id is absent from the retained history. Update history_ascending and
its callers to return or otherwise propagate an explicit gap indicator for
stale/evicted IDs, and have the SSE handling path emit the existing lagged-style
signal while preserving normal event delivery.
- Around line 18-20: Update the schema example on the event_type field in the
event definition to match the format produced by build_cloud_event: use the
io.impierce.unicore prefix followed by the kebab-case event name, such as
offer-created, instead of the current dotted org.unicore value.
- Around line 207-220: Update EventBusHandle::publish to preserve publish order
by making the history-buffer write part of the awaited publish flow instead of
spawning an independent tokio task. Acquire the existing history write lock,
enforce history_capacity, and append the event inline; update all publish call
sites, including dispatch, attach_source, and tests, to await the async method.
---
Outside diff comments:
In `@agent_application/src/lib.rs`:
- Around line 221-282: Prevent duplicate MongoDB event publication in the
EventStoreType::MongoDb initialization by choosing a single dispatch path:
retain either event_bus.attach_source(MongoEventSource::new(...)) or the
event_bus.query() publishers passed into the aggregate builders, but not both.
Update the MongoDB builder setup and related publisher arguments consistently
while preserving event delivery through the selected source.
---
Nitpick comments:
In `@agent_api_http/src/v0/events/mod.rs`:
- Around line 127-210: The tests in test_events_sse_route,
test_events_sse_catchup_route, and test_events_sse_timestamp_filter only verify
response status; extend the test module to consume SSE response bodies and
assert live delivery when an event is published after the connection opens. Add
coverage for the stream’s lagged and error branches around the SSE event
handling logic, asserting each produces the expected SSE output or termination
behavior.
In `@agent_api_http/src/v0/issuance/credentials.rs`:
- Around line 721-722: Update setup_library_state to accept a shared
&EventBusHandle parameter and pass it to library_state instead of creating a new
EventBusHandle internally. Adjust every caller to create one EventBusHandle per
application fixture and reuse it across issuance and authorization setup.
In `@agent_application/src/lib.rs`:
- Around line 112-113: Update the EventBusHandle initialization in the
application setup so the shared broadcast channel’s capacity is explicitly sized
for events across all aggregates and services, rather than assuming a
single-aggregate workload. Preserve one shared EventBusHandle for the entire
application, and add monitoring or alerting for EventBusError::Lagged counts if
supported by the existing event-bus integration.
In `@shared-kernel/src/event_bus.rs`:
- Around line 299-321: Update EventBusHandle::subscribe so the outer filter_map
closure computes filter.matches(&event) synchronously while it owns the original
filter, instead of cloning EventFilter for each broadcast item and moving a
clone into the async block. Preserve the existing event filtering and Lagged
error behavior, and keep the stream type compatible with BusEventStream.
- Around line 164-167: Add an authorization/scoping boundary to the
EventBus/EventBusHandle flow before exposing events through SSE. Ensure
Query<A>::dispatch and subscribe enforce tenant/aggregate visibility so
subscribers cannot receive unrelated CloudEvent payloads, and have the SSE
handler reject or filter unauthorized subscriptions rather than relying only on
type/source/subject/time filters.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 77a8c5e9-f670-4a43-a568-0dcdd70a3097
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (34)
Cargo.tomlagent_api_http/Cargo.tomlagent_api_http/openapi-generated.yamlagent_api_http/src/lib.rsagent_api_http/src/v0/authorization/authorization_server/authorize.rsagent_api_http/src/v0/authorization/authorization_server/par.rsagent_api_http/src/v0/authorization/authorization_server/token.rsagent_api_http/src/v0/events/mod.rsagent_api_http/src/v0/events/openapi.rsagent_api_http/src/v0/issuance/credential_issuer/credential.rsagent_api_http/src/v0/issuance/credential_issuer/notification.rsagent_api_http/src/v0/issuance/credential_issuer/token_status_list.rsagent_api_http/src/v0/issuance/credential_issuer/well_known/oauth_authorization_server.rsagent_api_http/src/v0/issuance/credential_issuer/well_known/openid_credential_issuer.rsagent_api_http/src/v0/issuance/credentials.rsagent_api_http/src/v0/issuance/nonce/mod.rsagent_api_http/src/v0/issuance/offers/mod.rsagent_api_http/src/v0/issuance/public_offers.rsagent_api_http/src/v0/mod.rsagent_api_http/src/v0/openapi.rsagent_api_http/src/v0/templates/mod.rsagent_api_http/src/v0/verification/authorization_requests.rsagent_api_http/src/v0/verification/relying_party/redirect.rsagent_api_http/src/v0/verification/relying_party/request.rsagent_application/src/lib.rsagent_holder/src/offer/aggregate.rsagent_issuance/src/application/nonce_validation_service.rsagent_issuance/tests/credential_configuration_projection.rsagent_store/Cargo.tomlagent_store/src/event_source.rsagent_store/src/lib.rsshared-kernel/Cargo.tomlshared-kernel/src/event_bus.rsshared-kernel/src/lib.rs
There was a problem hiding this comment.
🧹 Nitpick comments (1)
agent_api_http/src/v0/events/mod.rs (1)
140-146: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert emitted SSE contents, not only HTTP status.
These tests never consume the response body, so catch-up delivery, filtering, serialization, and stream wiring can regress while all tests still pass. Assert the expected event IDs and that excluded events are absent.
Also applies to: 175-181, 203-214
🤖 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 `@agent_api_http/src/v0/events/mod.rs` around lines 140 - 146, Update the SSE tests around the response assertions in the events module to consume the response body and verify the emitted event contents, including the expected event IDs and absence of events excluded by source filtering. Apply the same assertions to the additional test cases noted in the comment, while retaining the existing status and content-type checks.
🤖 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.
Nitpick comments:
In `@agent_api_http/src/v0/events/mod.rs`:
- Around line 140-146: Update the SSE tests around the response assertions in
the events module to consume the response body and verify the emitted event
contents, including the expected event IDs and absence of events excluded by
source filtering. Apply the same assertions to the additional test cases noted
in the comment, while retaining the existing status and content-type checks.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: c164e8b8-b0be-4e0b-a786-e00eadac24d7
📒 Files selected for processing (1)
agent_api_http/src/v0/events/mod.rs
daniel-mader
changed the title
/v0/events endpoint
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
shared-kernel/src/event_bus.rs (1)
246-285: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winFix inconsistent
limithandling inhistory_ascending.Two distinct bugs exist in this method:
- When
last_event_idis found (Line 266-271), the result isn't capped bylimitat all. Onlyfilteris applied. A caller can receive up to the full history buffer size regardless of the requestedlimit, unlike the documentedlimit-bounded contract used byevents_sse_handler.- In the gap-detected branch (Line 273-277) and the no-id branch (Line 278-282),
skipis computed from the raw buffer length before filtering. A selective filter can then return fewer thanlimitmatching events even though more matches exist earlier in the buffer.history()avoids this by filtering during reverse iteration before applyingtake(limit).Apply the same reverse-filter-then-take approach used in
history()to all three branches ofhistory_ascending.🐛 Proposed fix
let events: Vec<CloudEvent> = if let Some(last_id) = last_event_id { if let Some(pos) = lock.iter().position(|e| e.id == last_id) { - lock.iter() - .skip(pos + 1) - .filter(|e| filter.matches(e)) - .cloned() - .collect() + let mut matched: Vec<CloudEvent> = lock + .iter() + .skip(pos + 1) + .filter(|e| filter.matches(e)) + .cloned() + .collect(); + if matched.len() > limit { + matched.drain(0..matched.len() - limit); + } + matched } else { gap_detected = true; - let count = lock.len(); - let skip = count.saturating_sub(limit); - lock.iter().skip(skip).filter(|e| filter.matches(e)).cloned().collect() + let mut matched: Vec<CloudEvent> = + lock.iter().rev().filter(|e| filter.matches(e)).take(limit).cloned().collect(); + matched.reverse(); + matched } } else { - let count = lock.len(); - let skip = count.saturating_sub(limit); - lock.iter().skip(skip).filter(|e| filter.matches(e)).cloned().collect() + let mut matched: Vec<CloudEvent> = + lock.iter().rev().filter(|e| filter.matches(e)).take(limit).cloned().collect(); + matched.reverse(); + matched };🤖 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 `@shared-kernel/src/event_bus.rs` around lines 246 - 285, Update history_ascending so every branch applies filter.matches before enforcing the limit: for a found last_event_id, return at most limit events after that ID; for gap-detected and no-ID branches, select the latest limit matching events using the same reverse-filter-then-take approach as history(), then restore chronological order. Preserve gap_detected semantics.
🧹 Nitpick comments (1)
agent_api_http/src/v0/events/mod.rs (1)
106-137: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate CloudEvent-to-SSE serialization logic.
The serialization match (
serde_json::to_string(&cloud_event)→sse::Eventon success,"error"event on failure) is duplicated between the catch-up loop (Lines 106-115) and the live stream.map()(Lines 127-137). Extract a shared helper, e.g.fn to_sse_event(cloud_event: CloudEvent) -> Result<sse::Event, axum::Error>, and call it from both places to avoid the two copies diverging over time.♻️ Proposed helper extraction
+fn cloud_event_to_sse(cloud_event: CloudEvent) -> Result<sse::Event, axum::Error> { + let event_type = cloud_event.event_type.clone(); + let event_id = cloud_event.id.clone(); + Ok(match serde_json::to_string(&cloud_event) { + Ok(json_data) => sse::Event::default().id(event_id).event(event_type).data(json_data), + Err(err) => sse::Event::default() + .event("error") + .data(format!("Serialization error: {}", err)), + }) +} + for cloud_event in catchup_events { - let event_type = cloud_event.event_type.clone(); - let event_id = cloud_event.id.clone(); - catchup_items.push(match serde_json::to_string(&cloud_event) { - Ok(json_data) => Ok(sse::Event::default().id(event_id).event(event_type).data(json_data)), - Err(err) => Ok(sse::Event::default() - .event("error") - .data(format!("Serialization error: {}", err))), - }); + catchup_items.push(cloud_event_to_sse(cloud_event)); } ... .map(move |result| match result { - Ok(cloud_event) => { - let event_type = cloud_event.event_type.clone(); - let event_id = cloud_event.id.clone(); - match serde_json::to_string(&cloud_event) { - Ok(json_data) => Ok(sse::Event::default().id(event_id).event(event_type).data(json_data)), - Err(err) => Ok(sse::Event::default() - .event("error") - .data(format!("Serialization error: {}", err))), - } - } + Ok(cloud_event) => cloud_event_to_sse(cloud_event), Err(EventBusError::Lagged(n)) => Ok(sse::Event::default() .event("lagged") .data(json!({ "dropped": n }).to_string())), Err(err) => Ok(sse::Event::default() .event("error") .data(format!("Event bus error: {}", err))), });🤖 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 `@agent_api_http/src/v0/events/mod.rs` around lines 106 - 137, Extract the duplicated CloudEvent-to-SSE conversion into a shared helper such as to_sse_event, preserving the existing success serialization and error-event behavior. Replace the serialization match in both the catchup_events loop and the live_stream map with calls to this helper, retaining the surrounding ID filtering and stream construction.
🤖 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 `@agent_api_http/src/v0/events/mod.rs`:
- Around line 96-126: Update the live-stream deduplication closure around
seen_ids and live_subscription.filter: check whether each successful event ID is
in the catch-up set by removing it, and treat removal failure as a non-duplicate
event. Preserve passing errors through unchanged, while ensuring seen_ids only
shrinks from its initial catch-up contents and never grows with live events.
In `@agent_store/src/event_source.rs`:
- Around line 33-39: Update the resume-token handling in the
SubscribePosition::From branch to add an else failure path for
bson::from_slice::<ResumeToken>(&pos.0). Log the deserialization error with
tracing::warn!, while preserving the existing successful resume_after assignment
and fallback behavior.
- Around line 20-24: Update the EventSource::open flow and related
BusEventStream/CloudEvent conversion so each MongoDB ChangeStreamEvent extracts
and propagates its resume token or position alongside the emitted event; ensure
attach_source can persist the last consumed position and pass
SubscribePosition::From on reconnect instead of always using
SubscribePosition::Live.
In `@shared-kernel/src/event_bus.rs`:
- Around line 218-229: Update EventBusHandle::publish to recover the history
write guard from a poisoned lock using into_inner(), matching the recovery
behavior in history and history_ascending, while preserving the existing
ring-buffer append logic. Log the poisoning event when recovery occurs so the
condition is observable.
---
Outside diff comments:
In `@shared-kernel/src/event_bus.rs`:
- Around line 246-285: Update history_ascending so every branch applies
filter.matches before enforcing the limit: for a found last_event_id, return at
most limit events after that ID; for gap-detected and no-ID branches, select the
latest limit matching events using the same reverse-filter-then-take approach as
history(), then restore chronological order. Preserve gap_detected semantics.
---
Nitpick comments:
In `@agent_api_http/src/v0/events/mod.rs`:
- Around line 106-137: Extract the duplicated CloudEvent-to-SSE conversion into
a shared helper such as to_sse_event, preserving the existing success
serialization and error-event behavior. Replace the serialization match in both
the catchup_events loop and the live_stream map with calls to this helper,
retaining the surrounding ID filtering and stream construction.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: d3c9b4ac-ed0e-4268-a9b2-610d011cd2df
📒 Files selected for processing (4)
agent_api_http/openapi-generated.yamlagent_api_http/src/v0/events/mod.rsagent_store/src/event_source.rsshared-kernel/src/event_bus.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- agent_api_http/openapi-generated.yaml
|
|
||
| let catchup_events = catchup_result.events; | ||
| let mut seen_ids: std::collections::HashSet<String> = catchup_events.iter().map(|e| e.id.clone()).collect(); | ||
|
|
||
| let mut catchup_items = Vec::new(); | ||
| if catchup_result.gap_detected { | ||
| catchup_items.push(Ok(sse::Event::default() | ||
| .event("lagged") | ||
| .data(json!({ "warning": "Last-Event-ID evicted from history" }).to_string()))); | ||
| } | ||
| for cloud_event in catchup_events { | ||
| let event_type = cloud_event.event_type.clone(); | ||
| let event_id = cloud_event.id.clone(); | ||
| catchup_items.push(match serde_json::to_string(&cloud_event) { | ||
| Ok(json_data) => Ok(sse::Event::default().id(event_id).event(event_type).data(json_data)), | ||
| Err(err) => Ok(sse::Event::default() | ||
| .event("error") | ||
| .data(format!("Serialization error: {}", err))), | ||
| }); | ||
| } | ||
|
|
||
| let catchup_stream = futures::stream::iter(catchup_items); | ||
|
|
||
| let live_stream = live_subscription | ||
| .filter(move |result| { | ||
| let is_duplicate = match result { | ||
| Ok(cloud_event) => !seen_ids.insert(cloud_event.id.clone()), | ||
| Err(_) => false, | ||
| }; | ||
| async move { !is_duplicate } | ||
| }) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
seen_ids grows without bound for the life of the SSE connection.
seen_ids starts with the catch-up event IDs. In the live_stream.filter() closure, every live event ID is inserted into seen_ids via !seen_ids.insert(cloud_event.id.clone()), regardless of whether it was a duplicate. SSE connections stay open indefinitely, so seen_ids accumulates one entry per streamed event for as long as the connection lives. Under sustained event volume and many concurrent connections, this causes unbounded per-connection memory growth.
Deduplication only needs to catch the overlap window between subscribe() and history_ascending(). Once a catch-up ID is matched in the live stream, it never needs to be checked again. Use HashSet::remove instead of HashSet::insert so the set only shrinks over time and never grows past the initial catch-up size.
🔧 Proposed fix to bound `seen_ids` memory growth
let live_stream = live_subscription
.filter(move |result| {
let is_duplicate = match result {
- Ok(cloud_event) => !seen_ids.insert(cloud_event.id.clone()),
+ Ok(cloud_event) => seen_ids.remove(&cloud_event.id),
Err(_) => false,
};
async move { !is_duplicate }
})📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let catchup_events = catchup_result.events; | |
| let mut seen_ids: std::collections::HashSet<String> = catchup_events.iter().map(|e| e.id.clone()).collect(); | |
| let mut catchup_items = Vec::new(); | |
| if catchup_result.gap_detected { | |
| catchup_items.push(Ok(sse::Event::default() | |
| .event("lagged") | |
| .data(json!({ "warning": "Last-Event-ID evicted from history" }).to_string()))); | |
| } | |
| for cloud_event in catchup_events { | |
| let event_type = cloud_event.event_type.clone(); | |
| let event_id = cloud_event.id.clone(); | |
| catchup_items.push(match serde_json::to_string(&cloud_event) { | |
| Ok(json_data) => Ok(sse::Event::default().id(event_id).event(event_type).data(json_data)), | |
| Err(err) => Ok(sse::Event::default() | |
| .event("error") | |
| .data(format!("Serialization error: {}", err))), | |
| }); | |
| } | |
| let catchup_stream = futures::stream::iter(catchup_items); | |
| let live_stream = live_subscription | |
| .filter(move |result| { | |
| let is_duplicate = match result { | |
| Ok(cloud_event) => !seen_ids.insert(cloud_event.id.clone()), | |
| Err(_) => false, | |
| }; | |
| async move { !is_duplicate } | |
| }) | |
| let catchup_events = catchup_result.events; | |
| let mut seen_ids: std::collections::HashSet<String> = catchup_events.iter().map(|e| e.id.clone()).collect(); | |
| let mut catchup_items = Vec::new(); | |
| if catchup_result.gap_detected { | |
| catchup_items.push(Ok(sse::Event::default() | |
| .event("lagged") | |
| .data(json!({ "warning": "Last-Event-ID evicted from history" }).to_string()))); | |
| } | |
| for cloud_event in catchup_events { | |
| let event_type = cloud_event.event_type.clone(); | |
| let event_id = cloud_event.id.clone(); | |
| catchup_items.push(match serde_json::to_string(&cloud_event) { | |
| Ok(json_data) => Ok(sse::Event::default().id(event_id).event(event_type).data(json_data)), | |
| Err(err) => Ok(sse::Event::default() | |
| .event("error") | |
| .data(format!("Serialization error: {}", err))), | |
| }); | |
| } | |
| let catchup_stream = futures::stream::iter(catchup_items); | |
| let live_stream = live_subscription | |
| .filter(move |result| { | |
| let is_duplicate = match result { | |
| Ok(cloud_event) => seen_ids.remove(&cloud_event.id), | |
| Err(_) => false, | |
| }; | |
| async move { !is_duplicate } | |
| }) |
🤖 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 `@agent_api_http/src/v0/events/mod.rs` around lines 96 - 126, Update the
live-stream deduplication closure around seen_ids and live_subscription.filter:
check whether each successful event ID is in the catch-up set by removing it,
and treat removal failure as a non-duplicate event. Preserve passing errors
through unchanged, while ensuring seen_ids only shrinks from its initial
catch-up contents and never grows with live events.
| impl EventSource for MongoEventSource { | ||
| /// Opens a change-stream listener on the MongoDB `events` collection. | ||
| /// | ||
| /// Supports resuming from a specific position when `SubscribePosition::From` contains a valid serialized BSON [`ResumeToken`]. | ||
| async fn open(&self, from: SubscribePosition) -> Result<BusEventStream, EventBusError> { |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Confirm whether anything other than EventBusHandle::attach_source
# calls MongoEventSource::open / EventSource::open, and whether any caller
# constructs SubscribePosition::From with a real resume token.
set -euo pipefail
rg -nP 'SubscribePosition::From' --type=rust
rg -nP -C5 '\.open\(' --type=rust -g '!*/target/*'
rg -nP -C5 'attach_source' --type=rustRepository: impierce/ssi-agent
Length of output: 156
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files '**/*.rs' 'Cargo.toml' | sed 's#^\./##' | head -200
echo "== event source files =="
fd -a 'event.*|event_source|event_bus' . | sed 's#^\./##' | head -200
echo "== search symbols =="
rg -n 'SubscribePosition|EventSource|MongoEventSource|Open\(|\.open\(|ChangeStreamEvent|ResumeToken|change\.id|ResumeToken' --glob '*.rs' . || true
echo "== git status/stat =="
git diff --stat || trueRepository: impierce/ssi-agent
Length of output: 9568
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== exact symbol search =="
rg -n 'SubscribePosition::From|SubscribePosition' --glob '*.rs' . || true
rg -n 'EventSource::open|event_source\.open|\.open\(' --glob '*.rs' . || true
rg -n 'attach_source' --glob '*.rs' . || true
rg -n 'ChangeStreamEvent|ResumeToken|change\.id|id:' --glob '*.rs' . || true
echo "== candidate files =="
git ls-files '**/*.rs' | rg 'event_source|event_bus|agent_store|shared.kernel|shared_kernel' || true
echo "== git diff stat =="
git diff --stat || true
git diff -- agent_store/src/event_source.rs shared-kernel/src/event_bus.rs \
2>/dev/null | sed -n '1,220p' || trueRepository: impierce/ssi-agent
Length of output: 50375
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== agent_store/event_source.rs around open/filter_map =="
sed -n '1,180p' agent_store/src/event_source.rs
echo ""
echo "== shared-kernel/event_bus.rs around open/open impl/attach_source =="
sed -n '140,315p' shared-kernel/src/event_bus.rs
echo ""
echo "== mongo_event_source tests/conversions if present =="
rg -n 'MongoEventSource|ResumeToken|change_stream|watch|ResumeAfter|SubscribePosition::From' agent_store shared-kernel --glob '*.rs' || trueRepository: impierce/ssi-agent
Length of output: 12545
Make the resume position visible on reconnect.
ChangeStreamEvent.id is not extracted, and each item is converted into a CloudEvent, which carries no MongoDB resume token/position. BusEventStream still exposes only CloudEvent, so attach_source cannot persist or resume from the last consumed position and reconnects call source.open(SubscribePosition::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 `@agent_store/src/event_source.rs` around lines 20 - 24, Update the
EventSource::open flow and related BusEventStream/CloudEvent conversion so each
MongoDB ChangeStreamEvent extracts and propagates its resume token or position
alongside the emitted event; ensure attach_source can persist the last consumed
position and pass SubscribePosition::From on reconnect instead of always using
SubscribePosition::Live.
| let mut options = mongodb::options::ChangeStreamOptions::default(); | ||
| if let SubscribePosition::From(ref pos) = from { | ||
| // Deserialize resume token from position bytes if available | ||
| if let Ok(resume_token) = bson::from_slice::<mongodb::change_stream::event::ResumeToken>(&pos.0) { | ||
| options.resume_after = Some(resume_token); | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Log resume-token deserialization failures.
If bson::from_slice::<ResumeToken>(&pos.0) fails, options.resume_after stays None and the stream opens without any warning. The caller can believe it resumed correctly while the stream actually starts fresh, silently losing events. Add a tracing::warn! in the failure branch so this degradation is observable, consistent with the malformed-document logging already added below.
🔧 Proposed fix
let mut options = mongodb::options::ChangeStreamOptions::default();
if let SubscribePosition::From(ref pos) = from {
// Deserialize resume token from position bytes if available
- if let Ok(resume_token) = bson::from_slice::<mongodb::change_stream::event::ResumeToken>(&pos.0) {
- options.resume_after = Some(resume_token);
+ match bson::from_slice::<mongodb::change_stream::event::ResumeToken>(&pos.0) {
+ Ok(resume_token) => options.resume_after = Some(resume_token),
+ Err(err) => {
+ tracing::warn!("Failed to deserialize resume token, starting fresh stream: {}", err);
+ }
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let mut options = mongodb::options::ChangeStreamOptions::default(); | |
| if let SubscribePosition::From(ref pos) = from { | |
| // Deserialize resume token from position bytes if available | |
| if let Ok(resume_token) = bson::from_slice::<mongodb::change_stream::event::ResumeToken>(&pos.0) { | |
| options.resume_after = Some(resume_token); | |
| } | |
| } | |
| let mut options = mongodb::options::ChangeStreamOptions::default(); | |
| if let SubscribePosition::From(ref pos) = from { | |
| // Deserialize resume token from position bytes if available | |
| match bson::from_slice::<mongodb::change_stream::event::ResumeToken>(&pos.0) { | |
| Ok(resume_token) => options.resume_after = Some(resume_token), | |
| Err(err) => { | |
| tracing::warn!("Failed to deserialize resume token, starting fresh stream: {}", err); | |
| } | |
| } | |
| } |
🤖 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 `@agent_store/src/event_source.rs` around lines 33 - 39, Update the
resume-token handling in the SubscribePosition::From branch to add an else
failure path for bson::from_slice::<ResumeToken>(&pos.0). Log the
deserialization error with tracing::warn!, while preserving the existing
successful resume_after assignment and fallback behavior.
| impl EventBusHandle { | ||
| /// Publishes a [`CloudEvent`] to all active subscribers and synchronously appends it to the in-memory ring-buffer. | ||
| pub fn publish(&self, event: CloudEvent) { | ||
| let _ = self.sender.send(Arc::new(event.clone())); | ||
|
|
||
| if let Ok(mut lock) = self.history.write() { | ||
| if lock.len() >= self.history_capacity { | ||
| lock.pop_front(); | ||
| } | ||
| lock.push_back(event); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Recover from lock poisoning consistently in publish.
publish only appends to history if let Ok(mut lock) = self.history.write(). history and history_ascending instead recover from a poisoned lock with poisoned.into_inner(). If the lock is ever poisoned, publish silently and permanently stops appending new events to history (the lock stays poisoned), while reads keep working against a frozen buffer. Recover the same way in publish for consistent behavior, and log the poisoning so it's observable.
🔧 Proposed fix
pub fn publish(&self, event: CloudEvent) {
let _ = self.sender.send(Arc::new(event.clone()));
- if let Ok(mut lock) = self.history.write() {
- if lock.len() >= self.history_capacity {
- lock.pop_front();
- }
- lock.push_back(event);
- }
+ let mut lock = match self.history.write() {
+ Ok(guard) => guard,
+ Err(poisoned) => {
+ tracing::error!("Event bus history lock poisoned; recovering");
+ poisoned.into_inner()
+ }
+ };
+ if lock.len() >= self.history_capacity {
+ lock.pop_front();
+ }
+ lock.push_back(event);
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| impl EventBusHandle { | |
| /// Publishes a [`CloudEvent`] to all active subscribers and synchronously appends it to the in-memory ring-buffer. | |
| pub fn publish(&self, event: CloudEvent) { | |
| let _ = self.sender.send(Arc::new(event.clone())); | |
| if let Ok(mut lock) = self.history.write() { | |
| if lock.len() >= self.history_capacity { | |
| lock.pop_front(); | |
| } | |
| lock.push_back(event); | |
| } | |
| } | |
| impl EventBusHandle { | |
| /// Publishes a [`CloudEvent`] to all active subscribers and synchronously appends it to the in-memory ring-buffer. | |
| pub fn publish(&self, event: CloudEvent) { | |
| let _ = self.sender.send(Arc::new(event.clone())); | |
| let mut lock = match self.history.write() { | |
| Ok(guard) => guard, | |
| Err(poisoned) => { | |
| tracing::error!("Event bus history lock poisoned; recovering"); | |
| poisoned.into_inner() | |
| } | |
| }; | |
| if lock.len() >= self.history_capacity { | |
| lock.pop_front(); | |
| } | |
| lock.push_back(event); | |
| } |
🤖 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 `@shared-kernel/src/event_bus.rs` around lines 218 - 229, Update
EventBusHandle::publish to recover the history write guard from a poisoned lock
using into_inner(), matching the recovery behavior in history and
history_ascending, while preserving the existing ring-buffer append logic. Log
the poisoning event when recovery occurs so the condition is observable.
Description of change
This PR introduces the core event streaming infrastructure to
ssi-agent:EventBusHandleinshared-kernel) that automatically receives domain events across all CQRS state initializers.CloudEventenvelopes with full metadata (ID, type, source, subject, timestamp, and payload data).GET /v0/events): Added a Server-Sent Events endpoint allowing clients to stream live events in real-time with catch-up buffer support and query parameter filtering (types,sources,subject,since,until).utoipa::ToSchemaforCloudEventand registered the/v0/eventsroute and schema inopenapi-generated.yaml.Example HTTP requests to start streaming events:
Subscribe to a live real-time stream of all events emitted by UniCore:
Filter the stream to only receive Issuer URL Updated events:
Stream events created since a particular time:
Note: Upon restarting the agent, the Event Bus does not rehydrate past events from persistent storage. Consequently, the /v0/events endpoint only serves events emitted during the current runtime session (since the last startup).
Links to any relevant issues
N/A
How the change has been tested
cargo test --allto verifyEventBusbroadcast fanout, CloudEvent structure generation, SSE catch-up filtering, and CQRS component state initializations.generate_openapi_specandopenapi_spec_is_up_to_datetests inagent_api_http.To verify locally: