feat(messaging): deliver outbox messages

Orleans.slnx

@@ -42,6 +42,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.Messaging/Orleans.Messaging.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" />
@@ -152,6 +153,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.Messaging.Tests/Orleans.Messaging.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/resources/nuget-packages.md

@@ -76,6 +76,7 @@ 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.Messaging](https://www.nuget.org/packages/Microsoft.Orleans.Messaging) | Experimental grain-scoped 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.
 

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

@@ -11,6 +11,7 @@
     "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/sdkfornet/v4/apidocs/items/Util/TLoggingConfig.html": "AWS serves this public .NET SDK logging configuration reference 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.Messaging": "The new 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://twitter.com/intent/follow?screen_name=msftorleans": "Twitter serves this public Orleans follow intent to browsers and intermittently returns HTTP 403 to the bounded automated HEAD/GET probe; successful responses also occur.",
     "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.Messaging": "The new package is awaiting its first NuGet publication."
+  }
 }

src/Orleans.Core/Orleans.Core.csproj

@@ -76,6 +76,7 @@
     <InternalsVisibleTo Include="Orleans.Persistence.Firestore.Tests" />
     <InternalsVisibleTo Include="Orleans.GrainDirectory.Tests" />
     <InternalsVisibleTo Include="Orleans.Journaling" />
+    <InternalsVisibleTo Include="Orleans.Messaging" />
     <InternalsVisibleTo Include="Orleans.NativeAotSmoke" />
     <InternalsVisibleTo Include="Orleans.Placement.Tests" />
     <InternalsVisibleTo Include="Orleans.Redis.Tests" />

src/Orleans.Journaling/IJournaledStateManager.cs

@@ -52,6 +52,9 @@ public interface IJournaledStateManager : IAsyncDisposable
     /// Stage mutations only after establishing that they are safe to commit. Pending changes are shared
     /// by all callers using this manager. Storage acknowledgement establishes durability.
     /// A failed journal operation fences the manager; recovery requires a new manager and state machine instances.
+    /// Before capture, the manager prepares each registered state through <see cref="IStateMachine.PrepareWriteAsync"/>.
+    /// A preparation failure reports <see cref="JournaledStatePreCommitException"/> and retains pending changes
+    /// for an explicit retry.
     /// Cancellation stops the caller's wait; an already queued write continues to its storage outcome.
     /// </remarks>
     /// <param name="cancellationToken">The token used to cancel the caller's wait.</param>

src/Orleans.Journaling/IStateMachine.cs

@@ -15,6 +15,7 @@ namespace Orleans.Journaling;
 /// </item>
 /// <item>
 /// When the application requests a write, the journaled state manager calls
+/// <see cref="PrepareWriteAsync"/> to establish the state's persistence prerequisites, then
 /// <see cref="WritePendingEntries"/> (and occasionally <see cref="WriteSnapshot"/>) to materialize
 /// the pending changes, then flushes the journal to durable storage.
 /// </item>
@@ -68,6 +69,24 @@ public interface IStateMachine
     /// </remarks>
     void OnRecoveryCompleted() { }
 
+    /// <summary>
+    /// Establishes the persistence prerequisites for this state's pending changes.
+    /// </summary>
+    /// <param name="cancellationToken">The token which cancels preparation when the journal owner stops.</param>
+    /// <returns>A completion representing readiness to capture this state.</returns>
+    /// <remarks>
+    /// The manager invokes preparation before capturing an append or snapshot.
+    /// Preparation is idempotent and prepares this state's own prerequisites. After asynchronous
+    /// preparation completes, the manager repeats preparation for all states to include changes
+    /// staged on interleaved turns. Prepared prerequisites complete synchronously on subsequent calls
+    /// and remain valid through capture. A complete synchronous pass flows directly into capture.
+    /// Preparation completes without awaiting another operation on its own journal manager,
+    /// since that operation is serialized behind the current write.
+    /// A failure reports <see cref="JournaledStatePreCommitException"/> and retains pending changes
+    /// for an explicit write retry.
+    /// </remarks>
+    ValueTask PrepareWriteAsync(CancellationToken cancellationToken = default) => ValueTask.CompletedTask;
+
     /// <summary>
     /// Writes pending state changes to the journal.
     /// </summary>

src/Orleans.Journaling/JournaledStateManager.cs

@@ -282,6 +282,31 @@ private async Task WorkLoop()
                                 "The journaled state operation was queued before deletion reset its state.");
                         }
 
+                        if (workItem is AppendJournalWorkItem or WriteSnapshotWorkItem)
+                        {
+                            try
+                            {
+                                for (var prepared = false; !prepared;)
+                                {
+                                    await PrepareStates(ref prepared, _shutdownCancellation.Token).ConfigureAwait(true);
+                                }
+
+                                _shutdownCancellation.Token.ThrowIfCancellationRequested();
+                            }
+                            catch (OperationCanceledException) when (_shutdownCancellation.IsCancellationRequested)
+                            {
+                                workItem.TrySetCanceled(_shutdownCancellation.Token);
+                                return;
+                            }
+                            catch (Exception exception)
+                            {
+                                LogStatePreparationFailed(_shared.Logger, exception);
+                                storageActivity?.SetStatus(ActivityStatusCode.Error, "Journal state preparation failed.");
+                                workItem.SetException(new JournaledStatePreCommitException(exception));
+                                continue;
+                            }
+                        }
+
                         // Note that the implementation of each command is inlined to avoid allocating unnecessary async states.
                         // We are ok sacrificing some code organization for performance in the inner loop.
                         switch (workItem)
@@ -596,6 +621,28 @@ private async Task WorkLoop()
         }
     }
 
+    // Completes when every state is synchronously ready; otherwise returns the first incomplete preparation.
+    private ValueTask PrepareStates(ref bool prepared, CancellationToken cancellationToken)
+    {
+        prepared = true;
+        lock (_lock)
+        {
+            foreach (var state in _states.Values)
+            {
+                var preparation = state.PrepareWriteAsync(cancellationToken);
+                if (!preparation.IsCompletedSuccessfully)
+                {
+                    prepared = false;
+                    return preparation;
+                }
+
+                preparation.GetAwaiter().GetResult();
+            }
+        }
+
+        return ValueTask.CompletedTask;
+    }
+
     private void Fence(Exception exception)
     {
         lock (_lock)
@@ -1352,6 +1399,11 @@ void IStateMachine.WritePendingEntries(JournalStreamWriter writer) { }
         Message = "Error processing work items.")]
     private static partial void LogErrorProcessingWorkItems(ILogger logger, Exception exception);
 
+    [LoggerMessage(
+        Level = LogLevel.Error,
+        Message = "Journal state preparation failed; pending changes remain staged.")]
+    private static partial void LogStatePreparationFailed(ILogger logger, Exception exception);
+
     [LoggerMessage(
         Level = LogLevel.Information,
         Message = "State \"{Name}\" was not found. I have substituted a placeholder for graceful time-based retirement.")]

src/Orleans.Journaling/JournaledStatePreCommitException.cs

@@ -0,0 +1,21 @@
+namespace Orleans.Journaling;
+
+/// <summary>
+/// Reports a state preparation failure which prevented a journal write.
+/// </summary>
+/// <remarks>
+/// Pending changes remain staged and safe to commit. The caller can restore the prerequisite
+/// and explicitly retry persistence, or retire its owner and recover from durable state.
+/// </remarks>
+[GenerateSerializer]
+public sealed class JournaledStatePreCommitException : Exception
+{
+    /// <summary>
+    /// Initializes a state preparation failure.
+    /// </summary>
+    /// <param name="innerException">The original preparation failure.</param>
+    public JournaledStatePreCommitException(Exception innerException)
+        : base("The journal write was prevented by a state preparation failure.", innerException)
+    {
+    }
+}

src/Orleans.Journaling/README.md

@@ -207,15 +207,29 @@ provider and requested state name.
 
 - `Reset`: reset in-memory state and bind a `JournalStreamWriter`.
 - `ReplayEntry`: reconstruct state from a recorded operation.
+- `PrepareWriteAsync`: establish asynchronous persistence prerequisites for pending state.
 - `WritePendingEntries`: emit pending operations into the supplied writer.
 - `WriteSnapshot`: emit a reconstructible snapshot into the supplied writer.
 - `OnRecoveryCompleted`: finish reconstruction before application use.
 - `OnWriteCompleted`: publish effects that depend on storage acknowledgement.
 
+Command-backed states can use `IDurableValueCommandCodec<TCommand>` to encode a
+complete command before applying it locally. A snapshot is a command in the same
+application-defined schema which reconstructs the whole state. Annotate command data
+for Orleans serialization and supply System.Text.Json metadata when using the JSON codec.
+
 Recovery resets state machines and replays durable entries. A failed initialization can be retried on the
 same manager and registered states. `JournalReplayContext.ResolveStateMachine` routes entries to the
 state machine for their stream.
 
+Write preparation runs before append or snapshot capture.
+Each state prepares its own prerequisites idempotently. An asynchronous completion
+leaves prepared prerequisites synchronously ready; the manager then rechecks all states
+to include mutations staged on interleaved turns. A complete synchronous pass flows
+directly into capture. Preparation failures retain pending changes for an explicit
+write retry and report `JournaledStatePreCommitException`. The owner's shutdown token
+cancels preparation, and disposal drains its actual completion.
+
 `IJournaledStateManager` is independent of the grain-facing `IDurableStateManager` and extends
 `IAsyncDisposable`. Its owner API provides `RegisterStateMachine`, `TryGetStateMachine`,
 `InitializeAsync`, `WriteStateAsync`, whole-journal `DeleteStateAsync`, and `PendingWriteByteCount`.