feat(durable-messaging): compose hosting and public delivery - #10693
ReubenBond wants to merge 82 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
Adds a new durable, grain-scoped inbox/outbox messaging primitive (Microsoft.Orleans.DurableMessaging) built on Orleans Journaling + Durable Jobs, alongside supporting journaling participant/observer capabilities and Durable Jobs execution semantics needed for reliability, routing, and recovery.
Changes:
- Introduces
Microsoft.Orleans.DurableMessaging(envelopes, routing handlers, inbox/outbox contracts, DI/hosting integration, diagnostics, and instrumentation). - Extends Journaling with composable grain participants and write/recovery observers, plus rollback fencing and codec fixes for snapshot reference-scope replay.
- Extends Durable Jobs with feature handler registry, explicit
RetryAtdisposition/reset-rescheduling support, and improved execution deduplication.
Show a summary per file
| File | Description |
|---|---|
| test/Orleans.Journaling.Tests/OrleansBinaryCommandCodecTests.cs | Adds regression tests ensuring pre-change snapshot payloads replay with independent reference scopes. |
| test/Orleans.Journaling.Tests/JournaledGrainParticipantTests.cs | Adds integration coverage for composed journaling participants initialization and failure propagation. |
| test/Orleans.Journaling.Tests/DurableListDirectWriteTests.cs | Updates test double to implement new revert/rollback surface. |
| test/Orleans.Journaling.Tests/DurableCollectionDirectWriteTests.cs | Updates test double to implement new revert/rollback surface. |
| test/Orleans.DurableMessaging.Tests/Support/SnapshotProbe.cs | Adds snapshot-based test synchronization utility. |
| test/Orleans.DurableMessaging.Tests/Support/HandlerProbe.cs | Adds barrier utility to block/release handlers deterministically in tests. |
| test/Orleans.DurableMessaging.Tests/Support/DurableMessagingTestGrains.cs | Adds durable-messaging test grain, message/effect models, and handler behavior for scenarios. |
| test/Orleans.DurableMessaging.Tests/Support/DurableMessagingClusterFixture.cs | Adds reusable in-process cluster fixture with Durable Jobs + Journaling + Durable Messaging wiring. |
| test/Orleans.DurableMessaging.Tests/Support/ControlledJournalStorageProvider.cs | Adds controllable journal storage provider for injecting write blocking/failures in tests. |
| test/Orleans.DurableMessaging.Tests/Orleans.DurableMessaging.Tests.csproj | Introduces new Durable Messaging test project. |
| test/Orleans.DurableMessaging.Tests/Hosting/PublicDurableMessagingRegistrationTests.cs | Validates DI registrations, options behavior, and absence of friend-access requirements. |
| test/Orleans.DurableMessaging.Tests/Functional/MultiSiloDurableMessagingFailoverTests.cs | Adds multi-silo failover recovery test for stable inbox job ownership and exactly-once effects. |
| test/Orleans.DurableMessaging.Tests/Functional/InboxCapacityBehaviorTests.cs | Adds backpressure + recovery behavior tests under capacity constraints. |
| test/Orleans.DurableMessaging.Tests/Functional/DedupeExpiryBehaviorTests.cs | Adds dedupe expiry/compaction behavior test allowing later reprocessing. |
| test/Orleans.DurableMessaging.Tests/Contracts/HandlerRoutingContractTests.cs | Adds routing contract tests for exact, prefix, correlation, and typed handler behaviors. |
| test/Orleans.DurableMessaging.Tests/Contracts/DurableEnvelopeContractTests.cs | Adds envelope/builder/serializer contract tests including reply-to and context. |
| test/Orleans.DurableMessaging.Tests/Contracts/DeliveryAndOptionsContractTests.cs | Adds contracts for delivery result/status and options validation defaults/boundaries. |
| test/Orleans.DurableJobs.Tests/DurableJobs/JobShardTests.cs | Adds shard-level test distinguishing reschedule reset vs failure retry attempt semantics. |
| test/Orleans.DurableJobs.Tests/DurableJobs/JobShardManagerTestsRunner.cs | Adds reassignment test ensuring reschedule reset semantics persist through shard reassignment. |
| test/Orleans.DurableJobs.Tests/DurableJobs/DurableJobFeatureHandlerTests.cs | Adds coverage for registry behavior and durable job run-result compatibility semantics. |
| test/Orleans.Core.Tests/Orleans.Core.Tests.csproj | Links HierarchicalKey tests into core test project. |
| test/Orleans.Core.Tests/DurableJobs/ShardExecutorTests.cs | Adds executor tests for RetryAt, unknown disposition handling, and legacy shard behavior. |
| test/Orleans.Core.Tests/DurableJobs/DurableJobsExtensionsTests.cs | Adds DI scope/registry behavior tests and explicit replacement/decoration rejection. |
| test/Orleans.Core.Tests/DurableJobs/DurableJobReceiverExtensionTests.cs | Updates execution identity to (JobId, RunId), adds retention/dedup + feature handler precedence tests. |
| test/Benchmarks/Journaling/DurableListJournalBenchmarks.cs | Updates benchmark test double to implement new revert/rollback surface. |
| src/Orleans.Journaling/JournaledStateManager.cs | Adds rollback support flag, observer notifications, recovery fencing, and revert-pending-changes support. |
| src/Orleans.Journaling/IJournaledStateObserver.cs | Introduces observer interface for write/recovery boundaries. |
| src/Orleans.Journaling/IJournaledStateManager.cs | Adds SupportsRollback, observer registration, and RevertPendingChangesAsync surface. |
| src/Orleans.Journaling/IJournaledGrainParticipant.cs | Introduces participant initialization hook for composed journaling features. |
| src/Orleans.Journaling/Formats/OrleansBinary/OrleansBinaryDurableSetCommandCodec.cs | Fixes snapshot replay to reset reference scope per element. |
| src/Orleans.Journaling/Formats/OrleansBinary/OrleansBinaryDurableQueueCommandCodec.cs | Fixes snapshot replay to reset reference scope per element. |
| src/Orleans.Journaling/Formats/OrleansBinary/OrleansBinaryDurableListCommandCodec.cs | Fixes snapshot replay to reset reference scope per element. |
| src/Orleans.Journaling/Formats/OrleansBinary/OrleansBinaryDurableDictionaryCommandCodec.cs | Fixes snapshot replay to reset reference scope per key/value entry. |
| src/Orleans.Journaling/Formats/OrleansBinary/OrleansBinaryCommandCodecHelpers.cs | Adds helper to read values using independent serializer sessions (reference-scope reset). |
| src/Orleans.Journaling/DurableGrain.cs | Initializes composed participants before recovery and documents WriteStateAsync. |
| src/Orleans.DurableMessaging/RoutePrefixHandler.cs | Adds prefix-based routing handler base class and documentation. |
| src/Orleans.DurableMessaging/RouteKeyHandler.cs | Adds exact-route routing handler base class and documentation. |
| src/Orleans.DurableMessaging/README.md | Adds package readme with configuration and semantics overview. |
| src/Orleans.DurableMessaging/Orleans.DurableMessaging.csproj | Adds new packable Durable Messaging project. |
| src/Orleans.DurableMessaging/InboxHandlerContext.cs | Adds handler context implementation for sending outbox messages and building envelopes. |
| src/Orleans.DurableMessaging/IInboxHandlerContext.cs | Adds public handler context contract. |
| src/Orleans.DurableMessaging/IInboxHandler.cs | Adds handler interfaces including typed handler adapter and documentation. |
| src/Orleans.DurableMessaging/IDurableOutbox.cs | Adds outbox public contract. |
| src/Orleans.DurableMessaging/IDurableMessagingDiagnostics.cs | Adds diagnostics contract + internal implementation for dead letters. |
| src/Orleans.DurableMessaging/IDurableInboxExtension.cs | Adds grain extension contract for durable inbox delivery. |
| src/Orleans.DurableMessaging/IDurableInbox.cs | Adds inbox public contract including handler registration and lookup. |
| src/Orleans.DurableMessaging/Hosting/DurableMessagingExtensions.cs | Adds ISiloBuilder/IServiceCollection registration, option validation, and rollback requirement checks. |
| src/Orleans.DurableMessaging/DurableMessagingPumpResults.cs | Adds pump execution result tracking and one-shot timer handle helper. |
| src/Orleans.DurableMessaging/DurableMessagingInstruments.cs | Adds meters/counters/histograms for inbox/outbox behavior and depth tracking. |
| src/Orleans.DurableMessaging/DurableMessagingGrainParticipant.cs | Ensures messaging services materialize via journaling participant initialization. |
| src/Orleans.DurableMessaging/DurableMessageState.cs | Adds durable state models for inbox/outbox attempts and dead letters. |
| src/Orleans.DurableMessaging/DurableInbox.cs | Adds inbox implementation with handler registration/selection and storage-backed message access. |
| src/Orleans.DurableMessaging/DurableEnvelopeData.cs | Adds deferred (slice-based) envelope body/context storage and accessors. |
| src/Orleans.DurableMessaging/DurableEnvelope.cs | Adds durable envelope type (routing, correlation, reply-to, metadata, and payload). |
| src/Orleans.DurableMessaging/DeliveryStatus.cs | Adds delivery status enum contract. |
| src/Orleans.DurableMessaging/DeliveryResult.cs | Adds delivery result struct contract and factories. |
| src/Orleans.DurableMessaging/CorrelationHandler.cs | Adds correlation-hierarchy-based routing handler base class. |
| src/Orleans.DurableMessaging/Configuration/DurableInboxOptions.cs | Adds durable messaging options and validation contract. |
| src/Orleans.DurableJobs/ShardExecutor.cs | Adds RetryAt handling with reset-rescheduling path and explicit legacy-shard failure. |
| src/Orleans.DurableJobs/JournaledJobShard.cs | Implements reset-rescheduling in journaled shard via IResettableJobShard. |
| src/Orleans.DurableJobs/JobShard.cs | Adds reset-rescheduling support to base shard and introduces IResettableJobShard. |
| src/Orleans.DurableJobs/IDurableJobReceiverExtension.cs | Updates receiver extension for feature-handler lookup, new execution identity, and retention semantics. |
| src/Orleans.DurableJobs/IDurableJobHandlerRegistry.cs | Adds activation-scoped feature handler registry + lookup implementation. |
| src/Orleans.DurableJobs/Hosting/DurableJobsOptions.cs | Adds completed-attempt retention option and validation. |
| src/Orleans.DurableJobs/Hosting/DurableJobsExtensions.cs | Registers registry in DI and explicitly rejects replacement/decoration. |
| src/Orleans.DurableJobs/DurableJobRunResult.cs | Adds RetryAt disposition and related fields/compatibility semantics. |
| src/api/Orleans.Journaling/Orleans.Journaling.cs | Updates generated API surface for journaling observer/participant/rollback additions. |
| src/api/Orleans.DurableJobs/Orleans.DurableJobs.cs | Updates generated API surface for feature handlers/registry and RetryAt semantics. |
| src/api/Orleans.Core.Abstractions/Orleans.Core.Abstractions.cs | Updates generated API surface to include HierarchicalKey type and codec. |
| Orleans.slnx | Adds Durable Messaging projects (src + tests) to the solution. |
| docs/site/src/data/unpublished-api-packages.json | Marks Durable Messaging package as unpublished. |
| docs/site/src/data/external-link-allowlist.json | Allow-lists NuGet link for unpublished Durable Messaging package. |
| docs/site/src/content/docs/toc.yml | Adds Durable Messaging doc page to Journaling section. |
| docs/site/src/content/docs/resources/nuget-packages.md | Adds Durable Messaging to NuGet package catalog and updates guidance blurb. |
| docs/site/src/content/docs/grains/durable-messaging.md | Adds end-user documentation for durable messaging semantics and requirements. |
Review details
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
- Files reviewed: 83/83 changed files
- Comments generated: 4
- Review effort level: Lite
There was a problem hiding this comment.
Review details
Suppressed comments (4)
Previously missed (1) — in code that hasn't changed since the last review.
src/Orleans.DurableMessaging/DurableInbox.cs:43
- Exact-route handler storage uses the default string comparer, which is culture-sensitive. Route keys are protocol identifiers and should use ordinal comparisons (matching the rest of this PR’s routing semantics). Use
StringComparer.Ordinalfor_exactRouteHandlersto avoid surprising behavior under non-invariant cultures (e.g., Turkish-I issues).
_inbox = inbox;
_processed = processed;
_handlers = new List<IInboxHandler>();
_exactRouteHandlers = new Dictionary<string, IInboxHandler>();
_capacity = capacity;
src/Orleans.Journaling/IJournaledStateObserver.cs:16
- The remarks contradict the interface contract: they say the manager does not notify observers before a write, but this interface defines OnWriteStarted and JournaledStateManager invokes it before capturing state. This is confusing for implementers; update the remarks to reflect the actual callback sequence.
src/Orleans.DurableMessaging/RouteKeyHandler.cs:21 - The remarks suggest implementing
IInboxHandlerdirectly for prefix-based routing, but this PR also introducesRoutePrefixHandlerfor that exact scenario. Updating the docs helps steer users to the supported helper type and keeps the guidance consistent.
src/Orleans.DurableMessaging/InboxHandlerContext.cs:125 - The XML docs reference
IStateMachineManager.WriteStateAsync(), which doesn't exist in this package and appears to be a leftover name. This should point toIJournaledStateManager.WriteStateAsync()to avoid misleading API consumers.
- Files reviewed: 83/83 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
Review details
Suppressed comments (2)
Previously missed (2) — in code that hasn't changed since the last review.
src/Orleans.DurableJobs/IDurableJobReceiverExtension.cs:185
- LongPollGetJobStatusAsync creates a CancellationTokenSource but never cancels it. That means the Task.Delay will always run to completion even when the job finishes quickly, causing avoidable timer allocations/work per call.
using var cts = new CancellationTokenSource();
var longPollDuration = TimeSpan.FromTicks(Math.Min(_shared.MessagingOptions.ResponseTimeout.Divide(2).Ticks, _shared.Options.JobStatusPollInterval.Ticks));
await Task.WhenAny(Task.Delay(longPollDuration, cts.Token), state.Task);
if (!state.Task.IsCompleted)
{
return DurableJobRunResult.PollAfter(_shared.Options.JobStatusPollInterval);
}
src/Orleans.DurableMessaging/Hosting/DurableMessagingExtensions.cs:58
- The DurableInboxOptions validation predicate swallows all exceptions and turns them into a generic validation failure. That can hide unexpected exceptions (eg, NullReferenceException) and makes misconfiguration harder to diagnose. Consider only converting known validation exceptions into a failed predicate and letting other exceptions bubble.
optionsBuilder.Validate(
options =>
{
try
{
options.Validate();
return true;
}
catch
{
return false;
}
},
"DurableInboxOptions validation failed.");
- Files reviewed: 83/83 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
Review details
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
test/Orleans.DurableMessaging.Tests/Support/SnapshotProbe.cs:33
- WaitAsync adds a waiter to the per-grain list and then returns
waiter.Completion.Task.WaitAsync(...), but if that wait times out/throws, the waiter remains in_waitersindefinitely. This can leak memory and can also keep evaluating stale predicates on subsequent Publish calls. Consider awaiting with a cleanupfinallythat removes the waiter (and optionally removes the empty list from_waiters).
- Files reviewed: 85/85 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
Review details
Suppressed comments (4)
Previously missed (3) — in code that hasn't changed since the last review.
src/Orleans.Journaling/JournaledStateManager.cs:309
- WorkLoop captures a stable observer snapshot into the local
observersarray, butOnWriteStarted()is invoked by iterating_observersdirectly. If an observer is registered duringOnWritePreparingAsync(or from within another observer callback), it can change_observersmid-iteration, causing inconsistent callback sequencing and potentialInvalidOperationExceptiondue to collection mutation during enumeration. Use the capturedobserverssnapshot forOnWriteStartedto keep the write boundary consistent and mutation-safe.
src/Orleans.Journaling/JournaledStateManager.cs:493 OnWriteCompleted()is invoked by iterating_observersdirectly, even though a stableobserverssnapshot was captured earlier for this write. If observers are registered during the write pipeline, enumerating the live HashSet can throw (collection modified) or cause some observers to seeOnWriteCompletedwithout the correspondingOnWritePreparingAsync/OnWriteStarted. Use the per-write snapshot for completion callbacks.
This issue also appears on line 516 of the same file.
src/Orleans.Journaling/JournaledStateManager.cs:826
- Recovery completion notifies observers by enumerating
_observersdirectly inside a lock. If any observer callback registers another observer (re-entrantly) this can mutate the HashSet during enumeration and throw, potentially breaking recovery. Take a snapshot of_observersbefore iterating so callbacks are mutation-safe and sequencing is stable.
src/Orleans.Journaling/JournaledStateManager.cs:516
- In the no-op write path (
!hasCommittedBuffer),OnWriteCompleted()is called by iterating_observersdirectly. This has the same collection-mutation risk as the committed write path and can also notify observers which were registered afterOnWritePreparingAsyncran for this write. Prefer using the stableobserverssnapshot captured at the start of the work item.
- Files reviewed: 85/85 changed files
- Comments generated: 0 new
- Review effort level: Lite
|
Addressed the latest observer-snapshot review findings in \3d4829b8d. Each write now uses one stable observer snapshot for preparation, start, and completion (including no-op writes), and recovery snapshots observers before callbacks so re-entrant registration cannot mutate an active enumeration. Focused regression coverage passes on net8.0 and net10.0. |
There was a problem hiding this comment.
Review details
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
test/Orleans.DurableMessaging.Tests/Support/SnapshotProbe.cs:33
- WaitAsync adds a waiter to the per-grain list but never removes it if the 30s timeout elapses (WaitAsync throws). That can leak waiters across a test run and slow down Publish() due to ever-growing lists.
- Files reviewed: 85/85 changed files
- Comments generated: 1
- Review effort level: Lite
|
Addressed the remaining suppressed review findings in two focused commits:\n\n- \�222b4b77\ removes timed-out SnapshotProbe waiters so stale predicates are not retained or evaluated; regression coverage passes on net8.0 and net10.0.\n- \�d214b07b\ narrows durable inbox options validation to the documented \ArgumentOutOfRangeException, allowing unexpected faults to surface; the options-contract test passes on net8.0 and net10.0. |
There was a problem hiding this comment.
Review details
Suppressed comments (1)
src/Orleans.Journaling/IJournaledStateManager.cs:58
IJournaledStateManageris a public interface andRevertPendingChangesAsyncis added as a required member (no default implementation). This is a breaking change for any external implementations ofIJournaledStateManager. Consider providing a default interface implementation (similar toRegisterObserver) which throwsNotSupportedException, and useSupportsRollbackto indicate capability.
- Files reviewed: 86/86 changed files
- Comments generated: 0 new
- Review effort level: Lite
There was a problem hiding this comment.
Review details
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
src/Orleans.DurableMessaging/DurableOutbox.cs:168
- The XML docs reference
IStateMachineManager.WriteStateAsync(), but durable messaging commits are driven by journaling. This should referenceIJournaledStateManager.WriteStateAsync()to match the rest of the package docs and avoid misleading API consumers.
/// The message is persisted atomically with grain state when IStateMachineManager.WriteStateAsync()
/// is called. The background pump will deliver the message to the target grain ONLY AFTER
/// the message has been durably persisted.
- Files reviewed: 86/86 changed files
- Comments generated: 0 new
- Review effort level: Lite
Problem
Applications need grain-scoped durable messaging whose committed business effects and outgoing messages survive activation loss, with clear preparation, persistence, delivery and ownership boundaries.
Solution
Complete the public consumer layer with both
AddDurableMessaginghosting overloads, packaging, compiled application examples and public-host/provider-cutover integration.PrepareSendAsyncsnapshots and validates envelopes, reserves their identities and confirms a durable self-wakeup before shared mutations. The opaqueIPreparedOutboxBatchretains those prerequisites. Ordinary methods await preparation, apply complete safe-to-commit updates and callSend(batch)synchronously, then await the ordinary journal write within the batch's disposal scope. Handlers usePrepareAsync -> Action; inbox processing stages completion and deduplication with the action's prepared changes.Existing standard journaled dictionaries and values provide the eight inbox states and six outbox collections. The existing outbox sequence value associates an immutable message/owner cohort with actual storage acknowledgement, so only acknowledged messages become eligible for dispatch and later staged work belongs to a later write. Public hosting resolves canonical collections against the actual advanced owner and supplies the sequence codec from the configured scope and format.
Messaging owns operation completion, preparation retirement and delivery cleanup. The journal manager owns persistence failures, caller-wait cancellation and initialization coordination. Failed initial recovery can be explicitly retried on the same manager through a complete reset and replay; admitted write/delete failures remain terminal. One lifetime work-loop task handles recovery attempts and subsequent journal work, waiting for a newly queued explicit initialization request after a failed attempt. Initialization admission and owner shutdown share the manager lock. Full deletion follows the terminal owner workflow: stop and drain messaging, await actual journal deletion, then dispose or deactivate. Subsequent work uses a fresh owner.
The handler facade retains started and late acquisitions through their actual lifetimes. Its returned-ValueTask consumption check preserves caught preparation failures and safe alternative processing.
AsTask()transfers retrieval to a caller-owned task whose outcome the caller handles. Batch ownership, attempt and lifetime checks precede repeated-send handling.Grain-local durable state controls duplicate and orphan wakeups using the logical ownership generation and exact scheduler-returned job ID and shard. Healthy owners retain their physical handle; feature startup repairs wholly absent ownership after successful journal recovery. Transport is at-least-once and unordered. Retained
(SenderId, MessageId)deduplication defines the effectively-once window, with constant-time depth accounting and bounded metric dimensions.Applications select messaging through
IDurableMessagingGrainorDurableGrain. Activation setup consumes resolved interleaving properties and placement, then resolves the scoped graph before recovery. Grain-bound managers own lifecycle enrollment; standalone owners retain explicit state/dependency lifetimes and initialization/start/stop ordering.Dependency and review scope
This final layer depends on #11285 at
b6194aa166b3b4355ecf7563e0e9ae9d3404ed65. Published source head:720d2c880a0054730a166a1753f4babd9762ee62.Immutable final-layer comparison.
The inbox layer #11284 is at
b3bca7bd6ba4e54bb9cd84a76eb0eba506d23b56, rebased onto selected main8bc9fd244427351ad24ccc039a9e7642a42c1cc4. That base includes the merged Journaling foundation #11326 (b3184ebffc11f8c6494c5a3956ecb2f27b1c5c05). The open contracts PR #11282 retains published headfb911fa47667b91215371d9f43298c5178619de0; the inbox and upper layers contain a patch-equivalent replay of its eleven commits through2b8a7b072f254767c85789bd096e835516a6b2cc. Each cumulative PR targetsmainand provides its immutable incremental comparison.Public-consumer coverage exercises both hosting overloads, canonical scoped state identity, marker-selected grains, actual delivery/replay, terminal deletion with fresh-owner reuse and all twelve named-provider cutover cases. Storage and deactivation probes observe actual outcomes. Callback-ownership probes execute on the captured activation scheduler and assert context identity across startup barriers and stopped-owner checks. Explicit case mappings distinguish retained behavior from retired generic validation and deferred-wrapper assumptions. The exact timer/coalescing, arm-before-clock and release-before-duplicate-RPC ordering fixes from #11318 remain preserved.
The package includes both target-framework assemblies, XML documentation and the runtime README. Within-package framework/API validation is enabled; the package-specific released baseline is selected after its first release establishes one. Source-backed examples demonstrate ordinary prepared sends and handler replies; reply examples preserve correlation when the incoming envelope supplies a key. Historical heads and intermediate work remain preserved in checkpoints. All PR merges remain human decisions.
Microsoft Reviewers: Open in CodeFlow