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

@@ -7,7 +7,7 @@ ms.topic: reference
 
 # Orleans NuGet packages
 
-All official packages use the `Microsoft.Orleans` prefix and are published on [NuGet.org](https://www.nuget.org/profiles/Orleans). Keep Orleans package versions aligned within an application.
+Official packages use the `Microsoft.Orleans` prefix. Find released packages on [NuGet.org](https://www.nuget.org/profiles/Orleans). Keep Orleans package versions aligned within an application.
 
 ## Start with a metapackage
 
@@ -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` | Experimental grain-scoped inbox and outbox messaging contracts, awaiting their first NuGet publication. |
 
 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/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/DurableDictionary.cs

@@ -16,7 +16,7 @@ public interface IDurableDictionary<TKey, TValue> : IDictionary<TKey, TValue> wh
 
 [DebuggerTypeProxy(typeof(IDurableDictionaryDebugView<,>))]
 [DebuggerDisplay("Count = {Count}")]
-internal class DurableDictionary<TKey, TValue> : IDurableDictionary<TKey, TValue>, IStateMachine, IDurableDictionaryCommandHandler<TKey, TValue> where TKey : notnull
+internal class DurableDictionary<TKey, TValue> : IDurableDictionary<TKey, TValue>, IStateMachine, IReportsStagedChanges, IDurableDictionaryCommandHandler<TKey, TValue> where TKey : notnull
 {
     private readonly IDurableDictionaryCommandCodec<TKey, TValue> _codec;
     private readonly Dictionary<TKey, TValue> _items = [];

src/Orleans.Journaling/DurableList.cs

@@ -27,7 +27,7 @@ public interface IDurableList<T> : IList<T>
 
 [DebuggerTypeProxy(typeof(IDurableCollectionDebugView<>))]
 [DebuggerDisplay("Count = {Count}")]
-internal sealed class DurableList<T> : IDurableList<T>, IStateMachine, IDurableListCommandHandler<T>
+internal sealed class DurableList<T> : IDurableList<T>, IStateMachine, IReportsStagedChanges, IDurableListCommandHandler<T>
 {
     private readonly IDurableListCommandCodec<T> _codec;
     private readonly List<T> _items = [];

src/Orleans.Journaling/DurableQueue.cs

@@ -65,7 +65,7 @@ public interface IDurableQueue<T> : IEnumerable<T>, IReadOnlyCollection<T>
 
 [DebuggerTypeProxy(typeof(DurableQueueDebugView<>))]
 [DebuggerDisplay("Count = {Count}")]
-internal sealed class DurableQueue<T> : IDurableQueue<T>, IStateMachine, IDurableQueueCommandHandler<T>
+internal sealed class DurableQueue<T> : IDurableQueue<T>, IStateMachine, IReportsStagedChanges, IDurableQueueCommandHandler<T>
 {
     private readonly IDurableQueueCommandCodec<T> _codec;
     private readonly Queue<T> _items = new();

src/Orleans.Journaling/DurableSet.cs

@@ -40,7 +40,7 @@ public interface IDurableSet<T> : ISet<T>, IReadOnlyCollection<T>, IReadOnlySet<
 
 [DebuggerTypeProxy(typeof(IDurableCollectionDebugView<>))]
 [DebuggerDisplay("Count = {Count}")]
-internal sealed class DurableSet<T> : IDurableSet<T>, IStateMachine, IDurableSetCommandHandler<T>
+internal sealed class DurableSet<T> : IDurableSet<T>, IStateMachine, IReportsStagedChanges, IDurableSetCommandHandler<T>
 {
     private readonly IDurableSetCommandCodec<T> _codec;
     private readonly HashSet<T> _items = [];

src/Orleans.Journaling/DurableTaskCompletionSource.cs

@@ -42,7 +42,7 @@ public interface IDurableTaskCompletionSource<T>
 }
 
 [DebuggerDisplay("Status = {Status}")]
-internal sealed class DurableTaskCompletionSource<T> : IDurableTaskCompletionSource<T>, IStateMachine, IDurableTaskCompletionSourceCommandHandler<T>
+internal sealed class DurableTaskCompletionSource<T> : IDurableTaskCompletionSource<T>, IStateMachine, IReportsStagedChanges, IDurableTaskCompletionSourceCommandHandler<T>
 {
     private readonly IDurableTaskCompletionSourceCommandCodec<T> _codec;
     private readonly DeepCopier<T> _copier;
@@ -52,6 +52,7 @@ internal sealed class DurableTaskCompletionSource<T> : IDurableTaskCompletionSou
     private DurableTaskCompletionSourceStatus _status;
     private T? _value;
     private Exception? _exception;
+    private JournalStreamWriter _writer;
 
     public DurableTaskCompletionSource(
         [ServiceKey] string key,
@@ -90,6 +91,7 @@ public bool TrySetResult(T value)
         }
 
         _status = DurableTaskCompletionSourceStatus.Completed;
+        _writer.MarkModified();
         _value = _copier.Copy(value);
         return true;
     }
@@ -102,6 +104,7 @@ public bool TrySetException(Exception exception)
         }
 
         _status = DurableTaskCompletionSourceStatus.Faulted;
+        _writer.MarkModified();
         _exception = _exceptionCopier.Copy(exception);
         return true;
     }
@@ -114,6 +117,7 @@ public bool TrySetCanceled()
         }
 
         _status = DurableTaskCompletionSourceStatus.Canceled;
+        _writer.MarkModified();
         return true;
     }
 
@@ -154,6 +158,7 @@ private void OnValuePersisted()
 
     void IStateMachine.Reset(JournalStreamWriter writer)
     {
+        _writer = writer;
         _status = DurableTaskCompletionSourceStatus.Pending;
         _value = default;
         _exception = null;
@@ -198,12 +203,14 @@ private void WriteState(JournalStreamWriter writer)
     void IDurableTaskCompletionSourceCommandHandler<T>.ApplyCompleted(T value)
     {
         _status = DurableTaskCompletionSourceStatus.Completed;
+        _writer.MarkModified();
         _value = value;
     }
 
     void IDurableTaskCompletionSourceCommandHandler<T>.ApplyFaulted(Exception exception)
     {
         _status = DurableTaskCompletionSourceStatus.Faulted;
+        _writer.MarkModified();
         _exception = exception;
     }
 

src/Orleans.Journaling/DurableValue.cs

@@ -16,11 +16,12 @@ public interface IDurableValue<T>
 }
 
 [DebuggerDisplay("{Value}")]
-internal sealed class DurableValue<T> : IDurableValue<T>, IStateMachine, IDurableValueCommandHandler<T>
+internal sealed class DurableValue<T> : IDurableValue<T>, IStateMachine, IReportsStagedChanges, IDurableValueCommandHandler<T>
 {
     private readonly IDurableValueCommandCodec<T> _codec;
     private T? _value;
     private bool _isDirty;
+    private JournalStreamWriter _writer;
 
     public DurableValue(
         [ServiceKey] string key,
@@ -54,7 +55,11 @@ public T? Value
 
     private void OnValuePersisted() => OnPersisted?.Invoke();
 
-    public void OnModified() => _isDirty = true;
+    public void OnModified()
+    {
+        _isDirty = true;
+        _writer.MarkModified();
+    }
 
     void IStateMachine.ReplayEntry(JournalEntry entry, JournalReplayContext context) =>
         context.GetRequiredCommandCodec(entry.FormatKey, _codec).Apply(entry.Reader, this);
@@ -64,6 +69,7 @@ void IStateMachine.ReplayEntry(JournalEntry entry, JournalReplayContext context)
 
     void IStateMachine.Reset(JournalStreamWriter writer)
     {
+        _writer = writer;
         _value = default;
         _isDirty = false;
     }

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>
@@ -81,4 +84,5 @@ public interface IJournaledStateManager : IAsyncDisposable
     /// concurrent writers and should not be used for correctness decisions.
     /// </remarks>
     long PendingWriteByteCount => -1;
+
 }

src/Orleans.Journaling/IReportsStagedChanges.cs

@@ -0,0 +1,12 @@
+namespace Orleans.Journaling;
+
+/// <summary>
+/// Marks a state whose every change advances <see cref="JournaledStateManager.StagedChangeCount"/>: the state commits a
+/// journal entry when the change is made, or calls <see cref="JournalStreamWriter.MarkModified"/>.
+/// </summary>
+/// <remarks>
+/// A state which journals changes it can't observe, such as an object mutated in place and serialized whole when a write
+/// captures it, doesn't implement this interface. While one is registered, the manager can't tell whether a state staged
+/// a change, so <see cref="JournaledStateManager.StagedChangeCount"/> is negative.
+/// </remarks>
+internal interface IReportsStagedChanges;

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,25 @@ 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 prepares every registered state before capturing each append or snapshot, whether or not the state
+    /// has pending changes, so a state can establish prerequisites for changes which other states hold, or fail a write
+    /// whose pending changes must not commit. Preparation is idempotent. 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/JournalBufferWriter.cs

@@ -24,6 +24,7 @@ public abstract class JournalBufferWriter : IDisposable, IBufferWriter<byte>
 #endif
     private readonly ArcBufferWriter _buffer = new();
     private int _committedLength;
+    private long _stagedChangeCount;
     private bool _hasActiveEntry;
     private bool _disposed;
 
@@ -269,6 +270,28 @@ internal void AppendPreservedEntry(JournalStreamId streamId, IPreservedJournalEn
         }
     }
 
+    /// <summary>
+    /// Gets or sets a count which advances whenever a state stages a change and never decreases between writes.
+    /// </summary>
+    /// <remarks>The manager restores it after a write captures pending changes, which stages nothing new.</remarks>
+    internal long StagedChangeCount
+    {
+        get
+        {
+            lock (_lock) return _stagedChangeCount;
+        }
+        set
+        {
+            lock (_lock) _stagedChangeCount = value;
+        }
+    }
+
+    /// <summary>Records a change which a state stages in memory and journals when a write captures it.</summary>
+    internal void MarkModified()
+    {
+        lock (_lock) _stagedChangeCount++;
+    }
+
     internal void CommitActiveEntry(JournalStreamId streamId)
     {
         lock (_lock)
@@ -279,6 +302,7 @@ internal void CommitActiveEntry(JournalStreamId streamId)
             {
                 FinishEntry(streamId);
                 _committedLength = _buffer.Length;
+                _stagedChangeCount++;
             }
             catch
             {

src/Orleans.Journaling/JournalStreamWriter.cs

@@ -26,6 +26,9 @@ internal JournalStreamWriter(JournalStreamId id, JournalBufferWriter writer)
 
     internal bool IsInitialized => _writer is not null;
 
+    /// <summary>Records a change which the state journals when a write captures it.</summary>
+    internal void MarkModified() => _writer?.MarkModified();
+
     /// <summary>
     /// Appends a format-owned entry for retired or unknown state preservation.
     /// </summary>