Repository navigation
feat(messaging): process inbox messages - #11284
ReubenBond wants to merge 113 commits into
Conversation
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Unresolved lifecycle, selection-mutation, scheduling, recovery, retention, validation, and outbox-contract issues remain.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 2
Open (3)
What changed in this PR
Adds a non-packable Durable Messaging inbox runtime integrated with Journaling and Durable Jobs, including recoverable ownership, transactional handler processing, retries, deduplication, and dead-lettering.
Changes:
- Adds durable messaging contracts, routing, diagnostics, and configuration.
- Implements journal-backed acceptance, recovery, pump coordination, and handler processing.
- Adds Journaling integration and comprehensive contract and functional tests.
| File | Reviewed change |
|---|---|
test/Orleans.DurableMessaging.Tests/Support/SnapshotProbeTests.cs |
Snapshot probe tests |
test/Orleans.DurableMessaging.Tests/Support/SnapshotProbe.cs |
Snapshot waiting helper |
test/Orleans.DurableMessaging.Tests/Support/ReceiverTestServices.cs |
Test service composition |
test/Orleans.DurableMessaging.Tests/Support/JournaledTestOutbox.cs |
Journal-backed test outbox |
test/Orleans.DurableMessaging.Tests/Support/HandlerProbe.cs |
Handler synchronization helper |
test/Orleans.DurableMessaging.Tests/Support/DurableMessagingMetricProbe.cs |
Metrics test probe |
test/Orleans.DurableMessaging.Tests/Support/DurableMessagingClusterFixture.cs |
Messaging test cluster |
test/Orleans.DurableMessaging.Tests/Support/DurableMessagingBehaviorTestBase.cs |
Shared behavior test base |
test/Orleans.DurableMessaging.Tests/Support/ControlledJournalStorageProvider.cs |
Journal failure injection |
test/Orleans.DurableMessaging.Tests/Support/ControlledDurableJobManager.cs |
Controlled job scheduling |
test/Orleans.DurableMessaging.Tests/Orleans.DurableMessaging.Tests.csproj |
Test project configuration |
test/Orleans.DurableMessaging.Tests/Functional/MessagingRoutingAndDeadLetterTests.cs |
Routing and dead-letter tests |
test/Orleans.DurableMessaging.Tests/Functional/InboxHandlerTransactionTests.cs |
Handler transaction tests |
test/Orleans.DurableMessaging.Tests/Functional/InboxCapacityBehaviorTests.cs |
Capacity behavior tests |
test/Orleans.DurableMessaging.Tests/Functional/InboxCallbackOwnershipTests.cs |
Callback ownership tests |
test/Orleans.DurableMessaging.Tests/Functional/InboxAcceptanceBehaviorTests.cs |
Acceptance behavior tests |
test/Orleans.DurableMessaging.Tests/Functional/DurableMessagingClusterCollection.cs |
Test collection definition |
test/Orleans.DurableMessaging.Tests/Functional/DedupeExpiryBehaviorTests.cs |
Deduplication expiry tests |
test/Orleans.DurableMessaging.Tests/Contracts/HandlerRoutingContractTests.cs |
Handler routing contracts |
test/Orleans.DurableMessaging.Tests/Contracts/DurableMessagingPumpResultsTests.cs |
Pump result contracts |
test/Orleans.DurableMessaging.Tests/Contracts/DurableEnvelopeContractTests.cs |
Envelope contracts |
test/Orleans.DurableMessaging.Tests/Contracts/DeliveryAndOptionsContractTests.cs |
Delivery and options contracts |
test/Orleans.DurableMessaging.Tests/Contracts/ActivationValidationTests.cs |
Activation validation contracts |
src/Orleans.Journaling/IJournaledStateObserver.cs |
Journal lifecycle observer API |
src/Orleans.Journaling/IJournaledStateManager.cs |
Observer registration API |
src/Orleans.Journaling/IJournaledGrainParticipant.cs |
Participant contract |
src/Orleans.Journaling/DurableGrain.cs |
Participant initialization |
src/Orleans.DurableMessaging/RoutePrefixHandler.cs |
Prefix route handler |
src/Orleans.DurableMessaging/RouteKeyHandler.cs |
Exact route handler |
src/Orleans.DurableMessaging/README.md |
Runtime documentation |
src/Orleans.DurableMessaging/Orleans.DurableMessaging.csproj |
Messaging project configuration |
src/Orleans.DurableMessaging/InboxHandlerSelectionContext.cs |
Handler selection context |
src/Orleans.DurableMessaging/InboxHandlerContext.cs |
Handler context implementation |
src/Orleans.DurableMessaging/IInboxHandlerContext.cs |
Handler context API |
src/Orleans.DurableMessaging/IInboxHandler.cs |
Handler contract |
src/Orleans.DurableMessaging/IDurableOutbox.cs |
Outbox contract |
src/Orleans.DurableMessaging/IDurableMessagingDiagnostics.cs |
Diagnostics API |
src/Orleans.DurableMessaging/IDurableInboxExtension.cs |
Inbox extension API |
src/Orleans.DurableMessaging/IDurableInbox.cs |
Inbox API |
src/Orleans.DurableMessaging/DurableMessagingTime.cs |
Time arithmetic helpers |
src/Orleans.DurableMessaging/DurableMessagingStateNames.cs |
Durable state names |
src/Orleans.DurableMessaging/DurableMessagingStateManagerCapabilities.cs |
State manager capability checks |
src/Orleans.DurableMessaging/DurableMessagingPumpResults.cs |
Pump execution results |
src/Orleans.DurableMessaging/DurableMessagingPumpCoordinator.cs |
Pump coordination |
src/Orleans.DurableMessaging/DurableMessagingJobOwnership.cs |
Job ownership metadata |
src/Orleans.DurableMessaging/DurableMessagingInstruments.cs |
Messaging metrics |
src/Orleans.DurableMessaging/DurableMessagingGrainParticipant.cs |
Messaging participant |
src/Orleans.DurableMessaging/DurableMessagingActivationValidator.cs |
Activation validation |
src/Orleans.DurableMessaging/DurableMessageState.cs |
Retry and dead-letter state |
src/Orleans.DurableMessaging/DurableInbox.cs |
Inbox storage and routing |
src/Orleans.DurableMessaging/DurableEnvelopeData.cs |
Deferred payload access |
src/Orleans.DurableMessaging/DurableEnvelope.cs |
Durable envelope model |
src/Orleans.DurableMessaging/DurableDeadLetterRetention.cs |
Dead-letter retention |
src/Orleans.DurableMessaging/DeliveryStatus.cs |
Delivery status values |
src/Orleans.DurableMessaging/DeliveryResult.cs |
Delivery result contract |
src/Orleans.DurableMessaging/CorrelationHandler.cs |
Correlation routing |
src/Orleans.DurableMessaging/Configuration/DurableInboxOptions.cs |
Inbox configuration |
src/api/Orleans.Journaling/Orleans.Journaling.cs |
Generated Journaling API |
Orleans.slnx |
Solution project registration |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Code coverage
Report-only conclusion: current-main baseline stale. The newest successful coverage run tested ebfde71, not current main ceeebcf. Coverage combines every CI test matrix job, including providers, CodeGen, .NET 8/10, Linux, Windows, and macOS, using canonical physical source and branch identities. The comparison remains report-only while normal line and branch variance is calibrated. Coverage details |
There was a problem hiding this comment.
Copilot review overview
🔵 Needs a closer look
Six unresolved moderate findings and one nit remain, covering state rollback, option validation, shutdown and recovery handling, test-outbox idempotency, and an unused injected field.
Review effort: Lite
Findings: 2
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
One critical build issue and five moderate correctness, lifecycle, and test-isolation issues remain unresolved.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 3
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Six unresolved findings remain, including build-blocking and critical correctness issues.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 5
Open (6)
Composed keys with multi-segment child values lose segments · New Import CancellationToken in the handler context contract · New Remove the unused grain factory dependency Do not dispose the shutdown token while timers can still enter Handler selection does not enforce its read-only boundary Honor idempotent duplicate sends in the test outbox
4d18916 to
b05c432
Compare
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Unresolved ownership validation, compilation, and routing/performance issues require changes before approval.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 6
Open (6)
Reject delivery when recovered ownership is invalid · New Import CancellationToken in the handler context contract Composed keys with multi-segment child values lose segments Remove the unused grain factory dependency Do not dispose the shutdown token while timers can still enter Handler selection does not enforce its read-only boundary
Resolved since last review (1)
b05c432 to
c7b6bd9
Compare
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Unresolved ownership-race, retry/dead-letter accounting, deduplication, and shutdown-fault handling issues block approval.
Get a fresh assessment by requesting another Copilot review.
Review effort: Lite
Findings: 1
Open (2)
Resolved since last review (7)
Reject delivery when recovered ownership is invalid Import CancellationToken in the handler context contract Composed keys with multi-segment child values lose segments Remove the unused grain factory dependency Do not dispose the shutdown token while timers can still enter Handler selection does not enforce its read-only boundary Honor idempotent duplicate sends in the test outbox
c7b6bd9 to
67a4a2e
Compare
…drop racy waits Every retry fixture made the grain clock and the Durable Jobs clock the same object, so reading either clock passed. The retry fixture now runs its own job clock days apart from the grain clock, and a test shows that advancing only the grain clock neither retries nor dead-letters a message. Two acceptance tests asserted a wait by sleeping or by checking IsCompleted right after a call, which passed by race. They now assert the ordering contract directly: a duplicate is answered only once the completion it depends on is durable, and a duplicate of a durable pending message is answered at once.
…every write The inbox ensures wake-ups for work which other states hold, and vetoes writes after a messaging failure, so it relies on the manager preparing every registered state before each append or snapshot, including states without pending changes. The contract described only a state's own prerequisites, which would permit an optimization that silently drops those wake-ups. State the guarantee and pin it with a test.
Wake-ups for due work now fire up to RecheckDelay later, because the live loop is expected to perform the work. The loop started only on a delivery or a job run, so recovered work, and work staged by an activation which no messaging call had reached, waited for the wake-up. Start the loop once activation completes, when the activation services calls.
DeliverAsync interleaves with other requests, so a delivery reaches a grain whose running request is suspended, which is what lets a request wait for a message.
…plication window An inbox recognizes a redelivery only while it keeps the message's processed record. Before recording a new message, it compares the sender's redelivery deadline with the time its window covers on the Durable Jobs clock, which the deadline uses, and answers DeadLettered with DeduplicationWindowExceededCode instead of accepting a message whose redelivery it could process again. The inbox stages nothing for the rejection and the answer doesn't depend on its state, so every redelivery receives it. Duplicates are checked first, so a message already recorded stays a duplicate. The rejection logs an error naming the configuration to fix.
Apply the coordinator-selected ordering: known equal or earlier sends retain their answer, and expiration precedes the future bound after rollback. New sends and retention extensions remain bounded.
…port them Messaging fails the activation when a failed handler attempt staged changes, so they can't commit with its retry record, and decides that by comparing StagedChangeCount before and after the attempt. A JournaledPersistentState is mutated in place and serialized whole when a write captures it, so its changes never advance the count: a handler which mutated it and threw would have its partial state committed by the retry write. States which report every change now implement the internal IReportsStagedChanges: the built-in collections, values, and task completion sources, and the messaging inbox. While any other state is registered, StagedChangeCount is negative, which messaging already treats as staged, so a failed attempt recovers from the journal instead.
CompactProcessed removes processed records while enumerating them, which Dictionary supports on .NET Core 3.0 and later, and so on both frameworks Messaging targets. The test places expired records first and between retained ones, so compaction keeps enumerating after each removal.
Options validation accepts any positive DeduplicationWindow, including TimeSpan.MaxValue. The floor subtracts the window from the clock's ticks, which stays above long.MinValue for every representable time, and clamps at the minimum time. The test runs delivery, deduplication, and snapshot compaction with that window.
A handler attempt which fails after staging changes fails the activation, so recovery discards those changes, but it also discards the attempt's failure record. The message then reran with the same attempt count, so backoff and InboxRetryTimeout never applied: a handler which always failed after staging looped indefinitely, and so did any failure on a grain whose state can't report its changes. A graceful stop now journals a clean-stop record when messages are pending, and recovery charges one attempt to each due message unless the journal ends with that record. The charge commits before the pump runs a handler, and a recovery which finds a clean stop journals that it resumed, so a later unclean stop is charged. The happy path writes nothing extra; the cost falls on deactivation with pending work and on recovery. A crash also charges due messages which hadn't been dispatched, so attempts can overcount by one per unclean stop. A charge starts a message's retry window but dead-letters only a message whose window an earlier failure started, so one crash never dead-letters a message by itself.
The future bound, its exemption for recorded and floor-expired identities, and equality admission shipped without tests. Pin each branch, including receiver clock rollback, on a Durable Jobs clock which stands still between reading the bound and delivering.
There was a problem hiding this comment.
🔵 Needs a closer look
Two unresolved critical issues remain: a dictionary is modified during enumeration, causing compaction to throw, and an async test without await triggers a warnings-as-errors build failure.
No open issues 🎉
Review effort: Lite
There was a problem hiding this comment.
🔵 Needs a closer look
Human assessment is needed because the change contains intertwined state-management and diagnostic-copying behavior with critical hidden findings that could corrupt persisted state or misattribute local sends during handler failure.
No open issues 🎉
Review effort: Lite


Durable messaging stack: #11282 (contracts) → #11284 (inbox) → #11285 (outbox) → #10693 (integration, documentation, samples, NativeAOT smoke, benchmarks, and playground). This layer builds on #11282.
Problem
An inbox must durably accept commands, run handlers after acknowledgement, and commit business effects and completion together. Identity history must be reclaimable while delayed delivery, lost replies, and clock rollback preserve recorded decisions.
Solution
DeduplicationWindow. A completed record becomes removable once the durable floor covers its latest observedSentAt. Acknowledged snapshots and recovery reclaim it; ordinary and scalar acknowledgements remain constant-time.SentAtextensions: a redelivery with a laterSentAtextends the identity's retention before the inbox answers it. A processed extension is a self-contained completion upsert, so replay succeeds when a preceding snapshot omitted the older record. A pending extension is a delta and keeps the payload, subject, andDeliverAtconflict checks. A missing pending entry fails replay explicitly, and staging asserts ownership.Decision for review: deduplication precedes the
SentAtchecksThis ordering was chosen while you were away; please confirm it.
Inbox.DeliverAsyncevaluates a delivery in this order, wherenowis the receiver's Durable Jobs time andwindowisDeduplicationWindow:SentAt > now + window, the delivery throwsArgumentOutOfRangeExceptionwithout staging anything, unless the inbox already holds the identity with an equal or laterSentAt, or the identity is untracked andSentAtis at or below the admission floor. Equality withnow + windowis admitted.Duplicate, or theDeadLetteredrejection a sink first recorded. A laterSentAtfirst stages its retention extension, and the answer waits until the record is durable.Expired.max(durable floor, now - window)is the floor; the inbox persists any floor advance before it answers.Why this order:
SentAtnow exceedsnow + windowstill receives its recorded answer. An unconditional bound would turn a delivered message into a transient error, which the sender keeps retrying.Expired. The record for a pending or completed identity is evidence of its outcome. Only the durable, non-decreasing floor permits forgetting that evidence, so step 2 precedes step 3 whatever the clock does.The alternatives were expiration-first (forgets lost acknowledgements under skew) and an unconditional future bound (breaks recorded answers after rollback).
InboxSentAtBoundTestspins the bound, its equality boundary, and both rollback exemptions on a stationary Durable Jobs clock;DedupeExpiryBehaviorTestspins that a tracked identity staysDuplicateuntil its floor is durable.Rationale
The journal owns atomicity and acknowledgement; the inbox owns admission, recorded outcomes, and latest-observed times. Bounded extensions keep history reclaimable; the durable floor preserves expiration, and known outcomes survive backward clock movement. #11285 supplies the outbox runtime; this layer remains independently buildable.