feat(durable-messaging): add durable inbox and outbox

Orleans.slnx

@@ -41,6 +41,7 @@
     <Project Path="src/Orleans.CodeGenerator/Orleans.CodeGenerator.csproj" />
     <Project Path="src/Orleans.Core.Abstractions/Orleans.Core.Abstractions.csproj" />
     <Project Path="src/Orleans.Core/Orleans.Core.csproj" />
+    <Project Path="src/Orleans.DurableMessaging/Orleans.DurableMessaging.csproj" />
     <Project Path="src/Orleans.EventSourcing/Orleans.EventSourcing.csproj" />
     <Project Path="src/Orleans.Journaling/Orleans.Journaling.csproj" />
     <Project Path="src/Orleans.Persistence.Memory/Orleans.Persistence.Memory.csproj" />
@@ -148,6 +149,7 @@
     <Project Path="test/Orleans.GrainDirectory.Tests/Orleans.GrainDirectory.Tests.csproj" />
     <Project Path="test/Orleans.Streaming.Tests/Orleans.Streaming.Tests.csproj" />
     <Project Path="test/Orleans.BroadcastChannel.Tests/Orleans.BroadcastChannel.Tests.csproj" />
+    <Project Path="test/Orleans.DurableMessaging.Tests/Orleans.DurableMessaging.Tests.csproj" />
     <Project Path="test/Orleans.EventSourcing.Tests/Orleans.EventSourcing.Tests.csproj" />
     <Project Path="test/Orleans.DurableJobs.Tests/Orleans.DurableJobs.Tests.csproj" />
   </Folder>

docs/site/src/content/docs/grains/durable-messaging.md

@@ -0,0 +1,128 @@
+---
+title: Durable messaging
+description: Understand the durable inbox and outbox guarantees, recovery model, and operating limits.
+ms.date: 08/20/2026
+ms.topic: conceptual
+---
+
+# Durable messaging
+
+The `Microsoft.Orleans.DurableMessaging` package provides a grain-scoped inbox and
+outbox built on Orleans Journaling and Durable Jobs. It is intended for application
+messages whose state effects and outgoing messages must survive activation loss.
+
+## Message and routing model
+
+Each <xref:Orleans.DurableMessaging.DurableEnvelope> identifies its sender, target,
+route, message ID, optional <xref:Orleans.DurableMessaging.HierarchicalKey> correlation, optional
+`ReplyTo` grain ID, and an opaque serialized body. A receiver evaluates registered
+handlers in registration order. Handlers can select envelopes by exact route, route
+prefix, correlation hierarchy, or arbitrary metadata, and typed handlers deserialize
+the body only when selected. The receiving grain verifies that the envelope target
+matches its own identity before deduplication or persistence.
+
+The preview correlation key type now belongs to this package as
+<xref:Orleans.DurableMessaging.HierarchicalKey>. Draft consumers of
+`Orleans.HierarchicalKey` should update their namespace import. Its serialized alias
+and member identifiers remain unchanged, so envelopes written by the earlier draft
+remain readable.
+
+`ReplyTo` is general message metadata. Applications decide which route and body to use
+for a follow-up message.
+
+## Commit and delivery guarantees
+
+Durable Messaging has the following boundaries:
+
+- Calling <xref:Orleans.DurableMessaging.IDurableOutbox.Send*> stages an envelope in the
+  grain journal. Before journal capture, Durable Messaging allocates a stable job ID and
+  durably schedules the outbox job. The envelope, job ownership, and other journaled
+  grain effects then become durable in one commit. The job polls safely while the
+  envelope is provisional, and dispatch starts only after that commit succeeds.
+- Sending an equivalent envelope with the same `MessageId` more than once is idempotent,
+  whether the original is provisional or durable. Reusing that ID with different routing,
+  correlation, body, or request-context content throws without changing the outbox.
+- A failed inbox-handler attempt restores the last durable journal version before
+  retry or dead-letter accounting commits. The failed attempt's staged effects and
+  outgoing envelopes are discarded at that boundary.
+- Inbox handlers stage journaled effects and outgoing envelopes. Durable Messaging
+  commits those changes together with inbox completion after the handler returns;
+  handlers cannot create an earlier journal commit or delete boundary.
+- Deleting the grain journal discards staged inbox and outbox work and clears the
+  corresponding volatile pump bookkeeping before a later write begins.
+- A receiver allocates a stable ownership token and places it in a scheduled inbox job
+  before committing both the envelope and ownership, and returns `Accepted` only after
+  both are durable.
+- Transport is **at-least-once**. A crash after receiver acceptance but before durable
+  outbox removal can send the same envelope again.
+- The receiver deduplicates by `(SenderId, MessageId)`. Duplicate deliveries converge
+  on one set of handler effects while that deduplication record is retained. After
+  <xref:Orleans.DurableMessaging.Configuration.DurableInboxOptions.DeduplicationWindow>
+  expires, the same envelope can be accepted and processed again. The expired record and
+  replay acceptance are committed atomically.
+- Delivery is **unordered**. Inbox and outbox storage are dictionaries, and retries can
+  reorder envelopes. Applications which require ordering must carry sequence numbers
+  and make their handlers converge on application-defined order.
+
+The inbox and outbox use independent Durable Jobs. A blocked inbox handler on one grain
+doesn't stop another grain's outbox. Monotonic ownership generations fence job
+callbacks. Scheduling uses an internal stable physical job ID for each grain, pump, and
+ownership generation, so retrying an ambiguous response while the original schedule is
+active returns that job instead of creating another one. Completed-generation
+tombstones let delayed duplicates terminate. A job which wakes before its ownership
+commit or activation recovery is visible polls the same attempt instead of completing.
+After recovery, a scheduled generation with no committed owner and no work is a
+confirmed orphan and completes, so Durable Jobs removes it. If recovered work has no
+matching owner, recovery schedules and commits a new generation before the old
+generation terminates. Callbacks for the recovered and replacement generations both poll
+until replacement ownership commits, preserving the existing durable wake-up if
+scheduling or persistence must retry. Ownership-clear write failures restore the
+preceding generation, so the current job remains responsible. Pump callbacks execute as
+non-interleaving grain timer turns so that infrastructure writes can't commit
+provisional state from a concurrently running handler.
+
+## Backpressure, retries, and dead letters
+
+The inbox rejects new, nonduplicate envelopes with `Backpressured` when it reaches
+<xref:Orleans.DurableMessaging.Configuration.DurableInboxOptions.MaxCapacity>. The
+sender retains and retries the envelope. Handler failures restore the preceding durable
+state before retry accounting is committed. Messages move to the appropriate inbox or
+outbox dead-letter collection after their configured attempt or age limit. Use
+<xref:Orleans.DurableMessaging.IDurableMessagingDiagnostics> to inspect those records.
+After an operator or application has handled a record, remove it with
+<xref:Orleans.DurableMessaging.IDurableMessagingDiagnostics.RemoveInboxDeadLetter*>
+or <xref:Orleans.DurableMessaging.IDurableMessagingDiagnostics.RemoveOutboxDeadLetter*>
+so dead-letter storage remains bounded by the application's retention policy.
+Removal is staged in the grain transaction and becomes durable with the grain's next
+journal write.
+
+Malformed typed bodies are isolated during handler deserialization and follow the same
+retry and dead-letter path; they don't prevent later envelopes from being recovered.
+A successfully decoded null body is delivered as null. Typed handler parameters are
+explicitly null-capable and handlers which require a non-null body must validate it.
+
+## Deployment requirements
+
+Configure Durable Jobs storage and Journaling storage before enabling Durable
+Messaging. Grains which use Durable Messaging derive from
+<xref:Orleans.Journaling.DurableGrain>; its activation lifecycle initializes the
+journaled state manager and materializes the inbox and outbox participants before
+message recovery begins. Durable Messaging selects the built-in `orleans-binary`
+journal format so opaque envelope bodies and request-context slices recover exactly.
+Durable Messaging grains use non-reentrant execution: they don't apply `Reentrant`,
+`MayInterleave`, `AlwaysInterleave`, or `StatelessWorker`. A single non-interleaving
+activation owns each grain journal and pump, so infrastructure writes cannot commit
+provisional application state or compete with another activation for the same ownership.
+The Journaling implementation must provide
+<xref:Orleans.Journaling.IJournaledStateManager.RevertPendingChangesAsync*> and accept
+<xref:Orleans.Journaling.IJournaledStateManager.RegisterObserver*> so Durable Messaging
+receives commit and recovery notifications. Activation reports a
+durable-messaging-specific diagnostic when observer registration is unsupported. Use
+shared, production-grade storage for multi-silo deployments. In-memory Durable Jobs and
+journal storage are suitable only for development and tests.
+
+Capacity and retention settings bound storage growth and define the effectively-once
+window. Monitor inbox depth, outbox depth, retry failures, dead letters, and oldest
+pending-message age. Keep deduplication retention longer than the maximum expected
+outbox retry age. The `orleans-durable-messaging-orphaned-jobs-reclaimed` counter
+identifies terminal cleanup of schedule-before-commit crash remnants.

docs/site/src/content/docs/resources/nuget-packages.md

@@ -77,8 +77,9 @@ Memory persistence distributes records across cluster storage grains but isn't d
 | [Microsoft.Orleans.Reminders.Redis](https://www.nuget.org/packages/Microsoft.Orleans.Reminders.Redis) | Redis reminders. |
 | [Microsoft.Orleans.DurableJobs](https://www.nuget.org/packages/Microsoft.Orleans.DurableJobs) | Distributed scheduling for durable one-time jobs. |
 | [Microsoft.Orleans.DurableJobs.AzureStorage](https://www.nuget.org/packages/Microsoft.Orleans.DurableJobs.AzureStorage) | Azure Blob Storage for durable jobs. |
+| [Microsoft.Orleans.DurableMessaging](https://www.nuget.org/packages/Microsoft.Orleans.DurableMessaging) | Experimental grain-scoped durable inbox and outbox messaging. |
 
-Use reminders for recurring durable callbacks and durable jobs for scheduled one-time work. Grain timers are activation-scoped and use the core runtime rather than a provider package.
+Use reminders for recurring durable callbacks and durable jobs for scheduled one-time work. Durable Messaging composes Durable Jobs with Journaling for recoverable, at-least-once grain messages. Grain timers are activation-scoped and use the core runtime rather than a provider package.
 
 ## Streams and broadcast channels
 

docs/site/src/content/docs/toc.yml

@@ -112,6 +112,8 @@ items:
             href: grains/journaling/operations.md
           - name: Samples
             href: grains/journaling/samples.md
+      - name: Durable messaging
+        href: grains/durable-messaging.md
       - name: Transactions
         href: grains/transactions.md
   - name: Serialization and code generation

docs/site/src/data/external-link-allowlist.json

@@ -9,6 +9,7 @@
     "https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html": "AWS serves this public CLI configuration page to browsers but rejects the bounded automated HEAD/GET probe with HTTP 403.",
     "https://docs.aws.amazon.com/sdk-for-net/v4/developer-guide/creds-assign.html": "AWS serves this public SDK credential-resolution page to browsers but rejects the bounded automated HEAD/GET probe with HTTP 403.",
     "https://docs.aws.amazon.com/streams/latest/dev/introduction.html": "AWS serves this public Kinesis overview to browsers but rejects the bounded automated HEAD/GET probe with HTTP 403.",
+    "https://www.nuget.org/packages/Microsoft.Orleans.DurableMessaging": "The new Durable Messaging package is documented but not yet published; remove this entry and its unpublished API-package entry after publication.",
     "https://en.wikipedia.org/wiki/Kalman_filter": "Wikipedia serves this public article to browsers but rejects the bounded automated HEAD/GET probe with HTTP 403.",
     "https://www.f5.com/company/blog/nginx/nginx-power-of-two-choices-load-balancing-algorithm": "The F5 site serves this canonical NGINX article to browsers but rejects the bounded automated HEAD/GET probe with HTTP 403."
   }

docs/site/src/data/unpublished-api-packages.json

@@ -1,4 +1,6 @@
 {
   "description": "Generated API assemblies which are not currently published as standalone NuGet packages.",
-  "packages": {}
+  "packages": {
+    "Microsoft.Orleans.DurableMessaging": "The new package is awaiting its first NuGet publication."
+  }
 }

src/Orleans.DurableJobs/IDurableJobReceiverExtension.cs

@@ -30,7 +30,7 @@ internal sealed partial class DurableJobReceiverExtension : IDurableJobReceiverE
     private readonly IGrainContext _grain;
     private readonly DurableJobReceiverExtensionShared _shared;
     private readonly IDurableJobHandlerLookup _featureHandlers;
-    private readonly Dictionary<(string JobId, long ExecutionGeneration, int DequeueCount), JobAttemptState> _jobAttempts = [];
+    private readonly Dictionary<(string ShardId, string JobId, long ExecutionGeneration, int DequeueCount), JobAttemptState> _jobAttempts = [];
 
     public DurableJobReceiverExtension(
         IGrainContext grain,
@@ -145,7 +145,7 @@ private async Task<DurableJobRunResult> ExecuteHandlerAsync(
     }
 
     private ValueTask<DurableJobRunResult> GetJobStatusAsync(
-        (string JobId, long ExecutionGeneration, int DequeueCount) key,
+        (string ShardId, string JobId, long ExecutionGeneration, int DequeueCount) key,
         IJobRunContext context,
         JobAttemptState state,
         bool newJob,
@@ -181,7 +181,7 @@ private ValueTask<DurableJobRunResult> GetJobStatusAsync(
         return ValueTask.FromCanceled<DurableJobRunResult>(new CancellationToken(canceled: true));
 
         async ValueTask<DurableJobRunResult> LongPollGetJobStatusAsync(
-            (string JobId, long ExecutionGeneration, int DequeueCount) key,
+            (string ShardId, string JobId, long ExecutionGeneration, int DequeueCount) key,
             IJobRunContext context,
             JobAttemptState state,
             CancellationToken attemptCancellationToken)
@@ -218,7 +218,7 @@ async ValueTask<DurableJobRunResult> LongPollGetJobStatusAsync(
     }
 
     private DurableJobRunResult GetSuccessfulResult(
-        (string JobId, long ExecutionGeneration, int DequeueCount) key,
+        (string ShardId, string JobId, long ExecutionGeneration, int DequeueCount) key,
         JobAttemptState state)
     {
         var result = state.Task.Result;
@@ -239,16 +239,16 @@ private DurableJobRunResult GetSuccessfulResult(
         return result;
     }
 
-    private void RemoveJobAttempt((string JobId, long ExecutionGeneration, int DequeueCount) key, JobAttemptState state)
+    private void RemoveJobAttempt((string ShardId, string JobId, long ExecutionGeneration, int DequeueCount) key, JobAttemptState state)
     {
         if (_jobAttempts.TryGetValue(key, out var current) && ReferenceEquals(current, state))
         {
             _jobAttempts.Remove(key);
         }
     }
 
-    private static (string JobId, long ExecutionGeneration, int DequeueCount) GetExecutionKey(IJobRunContext context)
-        => (context.Job.Id, context.Job.ExecutionGeneration, context.DequeueCount);
+    private static (string ShardId, string JobId, long ExecutionGeneration, int DequeueCount) GetExecutionKey(IJobRunContext context)
+        => (context.Job.ShardId, context.Job.Id, context.Job.ExecutionGeneration, context.DequeueCount);
 
     internal sealed class TestAccessor(DurableJobReceiverExtension extension)
     {

src/Orleans.DurableJobs/InMemoryJobQueue.cs

@@ -101,6 +101,23 @@ public bool RemoveJob(string jobId)
         }
     }
 
+    public bool TryGetJob(string jobId, out DurableJob? job)
+    {
+        ArgumentException.ThrowIfNullOrWhiteSpace(jobId);
+        lock (_syncLock)
+        {
+            if (_jobsIdToBucket.TryGetValue(jobId, out var bucket)
+                && bucket.TryGetJob(jobId, out var entry))
+            {
+                job = entry.Job;
+                return true;
+            }
+
+            job = null;
+            return false;
+        }
+    }
+
     /// <summary>
     /// Returns whether the queue still contains the supplied durable job.
     /// </summary>

src/Orleans.DurableJobs/JobShard.cs

@@ -231,18 +231,19 @@ public async Task<DurableJobMutationResult> TryStartAttemptAsync(
                 throw new ArgumentOutOfRangeException(nameof(request), "Scheduled time is out of shard bounds.");
             }
 
-            var jobId = Guid.NewGuid().ToString();
-            var job = new DurableJob
+            if (request.JobId is { } requestedJobId
+                && _jobQueue.TryGetJob(requestedJobId, out var existingJob))
             {
-                Id = jobId,
-                TargetGrainId = request.Target,
-                Name = request.JobName,
-                DueTime = request.DueTime,
-                ShardId = Id,
-                Metadata = request.Metadata,
-                TraceParent = request.TraceParent,
-                TraceState = request.TraceState,
-            };
+                if (!request.Matches(existingJob!))
+                {
+                    throw new InvalidOperationException(
+                        $"Durable job ID '{requestedJobId}' is already scheduled with different properties.");
+                }
+
+                return existingJob;
+            }
+
+            var job = request.CreateJob(Id);
 
             await PersistAddJobAsync(job, cancellationToken);
             _jobQueue.Enqueue(job, 0);

src/Orleans.DurableJobs/JournaledJobShardState.cs

@@ -78,17 +78,19 @@ private JournaledJobShardState(
             throw new ArgumentOutOfRangeException(nameof(request), "Scheduled time is out of shard bounds.");
         }
 
-        var job = new DurableJob
+        if (request.JobId is { } requestedJobId
+            && _jobQueue.TryGetJob(requestedJobId, out var existingJob))
         {
-            Id = Guid.NewGuid().ToString(),
-            TargetGrainId = request.Target,
-            Name = request.JobName,
-            DueTime = request.DueTime,
-            ShardId = Id,
-            Metadata = request.Metadata,
-            TraceParent = request.TraceParent,
-            TraceState = request.TraceState,
-        };
+            if (!request.Matches(existingJob!))
+            {
+                throw new InvalidOperationException(
+                    $"Durable job ID '{requestedJobId}' is already scheduled with different properties.");
+            }
+
+            return existingJob;
+        }
+
+        var job = request.CreateJob(Id);
 
         Write(DurableJobShardJournalRecord.ForSchedule(job));
         ApplySchedule(job);

src/Orleans.DurableJobs/LocalDurableJobManager.cs

@@ -175,6 +175,7 @@ private static ScheduleJobRequest EnsureScheduleRequestHasTraceContext(ScheduleJ
 
         return new ScheduleJobRequest
         {
+            JobId = request.JobId,
             Target = request.Target,
             JobName = request.JobName,
             DueTime = request.DueTime,
@@ -634,7 +635,7 @@ public void AddWritableShard(DateTimeOffset shardKey, IJobShard shard, int strip
     }
 
     private WritableShardKey GetWritableShardKey(ScheduleJobRequest request)
-        => new(GetShardStartTime(request.DueTime), GetShardStripe());
+        => new(GetShardStartTime(request.DueTime), GetShardStripe(request.JobId));
 
     private IDictionary<string, string> CreateShardMetadata(WritableShardKey shardKey)
     {
@@ -655,13 +656,18 @@ private DateTimeOffset GetShardStartTime(DateTimeOffset scheduledTime)
         return new DateTimeOffset(bucketTicks, TimeSpan.Zero);
     }
 
-    private int GetShardStripe()
+    private int GetShardStripe(string? stableJobId)
     {
         if (_options.ShardStripeCount <= 1)
         {
             return 0;
         }
 
+        if (stableJobId is not null)
+        {
+            return (int)(StableHash.ComputeHash(stableJobId) % (uint)_options.ShardStripeCount);
+        }
+
         // Round-robin assignment. Stripe selection is a write-side fan-out knob only:
         // the persisted job location is the shard id, not (StartTime, Stripe), so consistency
         // across calls/silos is not required and round-robin distributes evenly under any input skew.