Feature/outbox wakeup signal - #100
Conversation
…ed outbox messages
…n instead of re-joining it
Reviewer's GuideAdds optional, loss-tolerant outbox wakeup notifications and coalescing wake-aware polling for faster same-process processing, signals inbox polling after delivery commits, and separates outbox publish telemetry into new root spans linked to their origin and worker-loop contexts. Sequence diagram for outbox commit wakeup and pollingsequenceDiagram
participant UOW as UnitOfWork
participant Store as EfCoreOutboxStore
participant Coordinator as OutboxWakeupCoordinator
participant Notifier as IOutboxWakeupNotifier
participant PubSub as DaprPubSub
participant Poller as OutboxBackgroundService
participant Processor as OutboxProcessor
UOW->>Store: StoreAsync(envelope)
Store->>Coordinator: OnOutboxMessageStored()
Coordinator->>UOW: OnCompleted(callback)
UOW->>UOW: CommitAsync()
UOW-->>Coordinator: callback
Coordinator-)Notifier: NotifyAsync(cancellationToken)
Notifier-)PubSub: PublishEventAsync(OutboxWakeupEvent)
PubSub-)Poller: wakeup event
Poller->>Poller: WaitAsync(timeout)
Poller->>Processor: poll and process messages
Sequence diagram for inbox delivery wakeupsequenceDiagram
participant Client
participant Controller as EventsController
participant UOW as UnitOfWork
participant Signal as IPollingWakeSignal_IInboxProcessor
participant Poller as InboxBackgroundService
participant Processor as InboxProcessor
Client->>Controller: ProcessEventAsync()
Controller->>UOW: CommitAsync()
UOW-->>Controller: committed inbox row
Controller->>Signal: Signal()
Signal-->>Poller: wake pending wait
Poller->>Processor: poll and process inbox
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
Important
This repository does not receive automatic reviews because it has fewer than 10 stars. ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 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 |
Up to standards ✅🟢 Issues
|
| Metric | Results |
|---|---|
| Complexity | 10 |
| Duplication | 19 |
NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="framework/src/BBT.Aether.Core/BBT/Aether/Events/OutboxWakeupEvent.cs" line_range="9" />
<code_context>
+/// as "poll now"; the payload is deliberately empty and delivery is best-effort — the adaptive
+/// polling interval remains the safety net for lost or early signals.
+/// </summary>
+[EventName("aether.outbox.wakeup")]
+public sealed class OutboxWakeupEvent;
</code_context>
<issue_to_address>
**issue (broader_impact):** The outbox wakeup event is published and the polling signal is registered, but no `IEventHandler<OutboxWakeupEvent>` or equivalent subscription forwards deliveries to `IPollingWakeSignal<IOutboxProcessor>.Signal()`. Consequently, successful wakeup publishes are consumed by no in-process component and outbox workers continue waiting for their normal polling interval.
**Triggers:** When `WakeupSignalEnabled` is true and an outbox message is committed.
**Suggested fix:** Add and register an `IEventHandler<OutboxWakeupEvent>` that resolves `IPollingWakeSignal<IOutboxProcessor>` and calls `Signal()`.
</issue_to_address>
### Comment 2
<location path="framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/DaprOutboxWakeupNotifier.cs" line_range="19-23" />
<code_context>
+ private readonly string _topic = topicNameStrategy.GetTopicName(typeof(OutboxWakeupEvent));
+
+ public Task NotifyAsync(CancellationToken cancellationToken = default)
+ => daprClient.PublishEventAsync(
+ eventBusOptions.PubSubName,
+ _topic,
+ new OutboxWakeupEvent(),
+ cancellationToken);
+}
</code_context>
<issue_to_address>
**issue (bug_risk):** `DaprOutboxWakeupNotifier` publishes a raw `OutboxWakeupEvent`, while the existing distributed event bus publishes serialized `CloudEventEnvelope` objects and its invokers deserialize that envelope shape. A wakeup subscriber using the framework's normal event handling path therefore cannot deserialize or dispatch this payload.
**Triggers:** When the wakeup topic is delivered through the framework's standard Dapr event-bus subscription pipeline.
**Suggested fix:** Publish the wakeup using the same CloudEvent envelope serialization and metadata path as `DaprEventBus`, or implement a dedicated raw-payload subscription that deserializes `OutboxWakeupEvent` directly.
</issue_to_address>Sourcery assessment
Needs a human reviewer. 2 findings to address first, and if the coordinator or topic wiring is wrong, it can emit stray wakeup messages and cause workers to poll or process existing outbox rows prematurely or repeatedly. Reverting stops future nudges, but already-published wakeup messages and their operational effects cannot be recalled; the impact is bounded and the normal polling path remains available.
Blocking findings: framework/src/BBT.Aether.Core/BBT/Aether/Events/OutboxWakeupEvent.cs:9, framework/src/BBT.Aether.Infrastructure/BBT/Aether/Events/DaprOutboxWakeupNotifier.cs:23
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| /// as "poll now"; the payload is deliberately empty and delivery is best-effort — the adaptive | ||
| /// polling interval remains the safety net for lost or early signals. | ||
| /// </summary> | ||
| [EventName("aether.outbox.wakeup")] |
There was a problem hiding this comment.
issue (broader_impact): The outbox wakeup event is published and the polling signal is registered, but no IEventHandler<OutboxWakeupEvent> or equivalent subscription forwards deliveries to IPollingWakeSignal<IOutboxProcessor>.Signal(). Consequently, successful wakeup publishes are consumed by no in-process component and outbox workers continue waiting for their normal polling interval.
Triggers: When WakeupSignalEnabled is true and an outbox message is committed.
Suggested fix: Add and register an IEventHandler<OutboxWakeupEvent> that resolves IPollingWakeSignal<IOutboxProcessor> and calls Signal().
| => daprClient.PublishEventAsync( | ||
| eventBusOptions.PubSubName, | ||
| _topic, | ||
| new OutboxWakeupEvent(), | ||
| cancellationToken); |
There was a problem hiding this comment.
issue (bug_risk): DaprOutboxWakeupNotifier publishes a raw OutboxWakeupEvent, while the existing distributed event bus publishes serialized CloudEventEnvelope objects and its invokers deserialize that envelope shape. A wakeup subscriber using the framework's normal event handling path therefore cannot deserialize or dispatch this payload.
Triggers: When the wakeup topic is delivered through the framework's standard Dapr event-bus subscription pipeline.
Suggested fix: Publish the wakeup using the same CloudEvent envelope serialization and metadata path as DaprEventBus, or implement a dedicated raw-payload subscription that deserializes OutboxWakeupEvent directly.
1 participant
Summary by Sourcery
Enable low-latency event processing by waking outbox and inbox pollers when new work is committed while preserving adaptive polling as a fallback.
New Features:
Enhancements:
Tests: