feat(durable-messaging): deliver application-keyed outbox intents

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.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" />
@@ -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.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/journaling/configuration.md

@@ -150,7 +150,7 @@ Serializer naming policies affect application payload values. Journal command na
 
 ## Migrate a journal format
 
-Providers expose the persisted format key as <xref:Orleans.Journaling.IJournalMetadata.FormatKey> and <xref:Orleans.Journaling.JournalMetadata.FormatKey>. Recovery selects the stored reader independently of the configured write format. When they differ, the next write creates a full snapshot using the configured format and updates the metadata. <xref:Orleans.Journaling.Json.JsonLinesJournalFormat.JournalFormatKey> supplies the JSON Lines format key.
+Providers expose the persisted format key as <xref:Orleans.Journaling.IJournalMetadata.FormatKey> and <xref:Orleans.Journaling.JournalMetadata.FormatKey>. Recovery selects the stored reader independently of the configured write format. When stored format metadata is absent, recovery uses the configured format. New empty journals use the configured write format. When the stored and configured keys differ, the next write creates a full snapshot using the configured format and updates the metadata. <xref:Orleans.Journaling.Json.JsonLinesJournalFormat.JournalFormatKey> supplies the JSON Lines format key.
 
 Use this deployment sequence:
 

docs/site/src/content/docs/grains/journaling/runtime-behavior.md

@@ -25,6 +25,8 @@ During <xref:Orleans.Runtime.GrainLifecycleStage.SetupState>, the manager:
 1. Resets and replays each registered durable state.
 1. Completes activation setup after replay finishes.
 
+Recovery uses the configured format when stored format metadata is absent. New empty journals use the configured write format.
+
 <xref:Orleans.Grain.OnActivateAsync*> and requests observe recovered durable state after setup succeeds, whether the grain derives directly from <xref:Orleans.Grain>, from an application-owned base, or from <xref:Orleans.Journaling.DurableGrain>. A storage read, format, codec, or malformed-data failure fails activation and preserves the stored journal for diagnosis and recovery.
 
 Provider registration makes Journaling services available. Per-grain journal I/O begins only for activations which resolve the manager, directly or through durable-state dependencies. Grains which use other persistence models keep their existing activation behavior.
@@ -68,6 +70,46 @@ Orleans executes that synchronous block on a single activation thread. Another g
 the operation awaits, so keep shared state safe to commit at each await. Any caller's write can include
 staged mutations from other calls.
 
+## Journal operation hooks
+
+Activation-scoped features coordinate prerequisites and completion through
+<xref:Orleans.Journaling.IJournaledStateManager.Hooks>. The owner exposes a stable, lazily
+allocated list of <xref:Orleans.Journaling.IJournaledStateHook> registrations. Inspect and
+deduplicate feature registrations on the owner's logical execution context while persistence
+is quiescent. Registration survives recovery and deletion; the standard manager rejects hook-list
+mutation while persistence is queued or running.
+Every journal owner implementation provides this list and runs its registered callbacks at the
+operation boundaries below. Delegating owners forward the list to the inner owner.
+
+Each actual append, snapshot, or deletion runs ordinary before callbacks in list order, outside
+the manager lock. At most one <xref:Orleans.Journaling.IJournaledStateCaptureHook> supplies the
+final prerequisite. Its before callback runs last, and the work loop awaits it directly before
+synchronous capture or storage deletion. Prerequisites cover changes staged during asynchronous
+preparation, including the final hook's own I/O wait. Preserve operation-local bookkeeping for
+the captured batch separately from changes staged later.
+
+Storage acknowledgement and registered-state acknowledgement or reset precede after callbacks.
+All after callbacks run in list order, including for successful zero-byte writes. Coalesced callers
+share callbacks for the actual operation. Features implement the before and after callbacks on
+their own hook, retaining feature identity and operation-local bookkeeping there.
+
+A failed prerequisite reports <xref:Orleans.Journaling.JournaledStatePreCommitException> with pending
+state retained for an explicit persistence retry after the prerequisite is restored. A failed after
+callback reports <xref:Orleans.Journaling.JournaledStatePostCommitException> with persistence completed.
+Remaining after callbacks run, multiple failures are aggregated, and the manager stays usable.
+The feature's durable recovery protocol resumes interrupted post-persistence work. Storage and
+state-processing failures retain the manager's fencing and fresh-recovery behavior.
+
+Hook callbacks receive the owner's shutdown token. Cancelling a caller's wait leaves the owned
+operation running through its actual outcome. Disposal drains owned hooks and storage before releasing
+journal resources, including when cancellation callbacks or cleanup fail. Concurrent disposal callers
+share this completion. Hook implementations complete without calling initialization, persistence,
+or disposal on their own owner. Awaiting an operation serialized behind the current callback creates
+a circular dependency. Shutdown closes work admission and cancels queued operations while the current
+operation drains to its actual storage and hook outcome. For deletion, the feature owner stops
+admission and drains feature operations before
+queuing the whole-journal reset.
+
 ## Consistency and competing writers
 
 Orleans grain placement normally supplies a single active writer for a grain identity. Journal storage providers also use optimistic concurrency to protect the journal when a stale or competing writer reaches storage.

docs/site/src/content/docs/grains/timers.md

@@ -32,9 +32,13 @@ Register timers with <xref:Orleans.GrainBaseExtensions.RegisterGrainTimer*>. <xr
 
 ### Callback scheduling
 
-A timer callback never overlaps itself. Orleans waits for the callback task to complete and then measures the period before scheduling the next callback. Callback duration therefore adds to the interval between callback starts.
+Orleans completes each callback task before scheduling the next callback for that timer. It then measures the period, so callback duration adds to the interval between callback starts.
 
-Timer callbacks are local-only messages addressed to their activation. They participate in normal turn scheduling and stay on the activation which registered them.
+Timer callbacks are local-only messages addressed to their activation. They participate in normal turn scheduling and start a fresh call chain. Each callback receives a fresh request context, while the state passed at registration remains available to the callback.
+
+A zero due time queues a callback directly on the activation. A zero period queues the next callback after the current callback completes. Both use the same activation scheduling, interleaving, and idle-lifetime rules as delayed ticks.
+
+The configured <xref:System.TimeProvider> controls delayed tick timing and resolution; Orleans queues the provider's tick notifications on the activation.
 
 ### Interleaving
 
@@ -50,9 +54,11 @@ With <xref:Orleans.Runtime.GrainTimerCreationOptions.KeepAlive> set to `true`, e
 
 ## Change or stop a timer
 
-Call <xref:Orleans.Runtime.IGrainTimer.Change*> to replace the due time and period. The new due time schedules the next callback, and the new period applies after that callback completes. A change made inside a running callback takes effect after the callback completes.
+Call <xref:Orleans.Runtime.IGrainTimer.Change*> to replace the due time and period. When a callback is already queued or running, it keeps its turn and the change takes effect after it completes. Repeated changes use the latest due time and period for the following schedule.
+
+A physical tick already dispatched by the provider can arrive after a change to another delayed schedule. Orleans admits that tick according to the activation's scheduling rules. After its callback completes, the configured period determines the next tick, or a change made during the callback supplies the next due time.
 
-Dispose <xref:Orleans.Runtime.IGrainTimer> to cancel its callback token and stop future callbacks. Orleans also cancels the token and disposes the timer when the activation begins deactivating.
+Dispose <xref:Orleans.Runtime.IGrainTimer> to invalidate queued ticks, cancel the token of an admitted callback, and stop further scheduling. Queued messages drain through activation scheduling. Orleans also cancels the token and disposes the timer when the activation begins deactivating.
 
 ## Handle callback failures
 

docs/site/src/content/docs/snippets/compiled/Grains/JournalingSnippets.cs

@@ -9,6 +9,20 @@
 
 namespace Documentation.Grains.Journaling;
 
+// <journal_operation_hooks>
+internal static class JournalHookRegistration
+{
+    internal static void Register(IJournaledStateManager owner, IJournaledStateHook featureHook)
+    {
+        var hooks = owner.Hooks;
+        if (!hooks.Contains(featureHook))
+        {
+            hooks.Add(featureHook);
+        }
+    }
+}
+// </journal_operation_hooks>
+
 // <composed_shopping_cart>
 public interface IShoppingCartGrain : IGrainWithStringKey
 {

src/Orleans.Core/Messaging/MessageFactory.cs

@@ -27,7 +27,10 @@ public MessageFactory(DeepCopier deepCopier, ILogger<MessageFactory> logger, Mes
             _seed = unchecked((ulong)Random.Shared.NextInt64());
         }
 
-        public Message CreateMessage(object? body, InvokeMethodOptions options)
+        public Message CreateMessage(object? body, InvokeMethodOptions options) =>
+            CreateMessage(body, options, RequestContextExtensions.Export(_deepCopier));
+
+        public Message CreateMessage(object? body, InvokeMethodOptions options, Dictionary<string, object>? requestContextData)
         {
             var message = new Message
             {
@@ -37,7 +40,7 @@ public Message CreateMessage(object? body, InvokeMethodOptions options)
                 IsUnordered = (options & InvokeMethodOptions.Unordered) != 0,
                 IsAlwaysInterleave = (options & InvokeMethodOptions.AlwaysInterleave) != 0,
                 BodyObject = body,
-                RequestContextData = RequestContextExtensions.Export(_deepCopier),
+                RequestContextData = requestContextData,
             };
 
             return message;