fix(durable-messaging): enforce activation safety

Author: ReubenBond

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

@@ -107,7 +107,13 @@ 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. The Journaling implementation must provide
+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

src/Orleans.DurableMessaging/DurableInboxExtension.cs

@@ -1022,8 +1022,12 @@ internal async Task ResumeProcessingAsync(bool replaceExisting, CancellationToke
         }
     }
 
-    public Task OnStart(CancellationToken cancellationToken) =>
-        ResumeProcessingAsync(replaceExisting: true, cancellationToken);
+    public Task OnStart(CancellationToken cancellationToken)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        DurableMessagingActivationValidator.Validate(_grainContext);
+        return ResumeProcessingAsync(replaceExisting: true, cancellationToken);
+    }
 
     public Task OnStop(CancellationToken cancellationToken)
     {

src/Orleans.DurableMessaging/DurableMessagingActivationValidator.cs

@@ -0,0 +1,41 @@
+using Orleans.Concurrency;
+
+namespace Orleans.DurableMessaging;
+
+internal static class DurableMessagingActivationValidator
+{
+    public static void Validate(IGrainContext grainContext)
+    {
+        var grain = grainContext.GrainInstance
+            ?? throw new InvalidOperationException("Durable Messaging activation requires an initialized grain instance.");
+        var grainType = grain.GetType();
+        if (grainType.IsDefined(typeof(StatelessWorkerAttribute), inherit: true))
+        {
+            throw new InvalidOperationException(
+                $"Durable Messaging requires one activation per grain identity, but grain type '{grainType}' is a stateless worker.");
+        }
+
+        if (grainType.IsDefined(typeof(ReentrantAttribute), inherit: true)
+            || grainType.IsDefined(typeof(MayInterleaveAttribute), inherit: true))
+        {
+            throw new InvalidOperationException(
+                $"Durable Messaging requires non-reentrant grain execution, but grain type '{grainType}' enables interleaving.");
+        }
+
+        var grainInterfaces = grainType
+            .GetInterfaces()
+            .Where(static type => typeof(IGrain).IsAssignableFrom(type))
+            .ToArray();
+        var interleavableMethod = grainInterfaces
+            .SelectMany(static type => type.GetInterfaces().Append(type))
+            .Distinct()
+            .SelectMany(static type => type.GetMethods())
+            .FirstOrDefault(static method => method.IsDefined(typeof(AlwaysInterleaveAttribute), inherit: true));
+        if (interleavableMethod is not null)
+        {
+            throw new InvalidOperationException(
+                $"Durable Messaging grain type '{grainType}' implements interleavable method "
+                + $"'{interleavableMethod.DeclaringType}.{interleavableMethod.Name}'.");
+        }
+    }
+}

src/Orleans.DurableMessaging/DurableOutbox.cs

@@ -653,6 +653,7 @@ private void DeadLetterExpiredMessage(
     public Task OnStart(CancellationToken cancellationToken = default)
     {
         cancellationToken.ThrowIfCancellationRequested();
+        DurableMessagingActivationValidator.Validate(_grainContext);
         EnsureMetricsActive();
         if (Count > 0)
         {

src/Orleans.DurableMessaging/Hosting/DurableMessagingExtensions.cs

@@ -19,6 +19,8 @@ namespace Orleans.Hosting;
 /// </summary>
 public static class DurableMessagingExtensions
 {
+    private const string DurableMessagingJournalFormatKey = "orleans-binary";
+
     /// <summary>
     /// Adds durable inbox and outbox messaging support to the silo.
     /// </summary>
@@ -35,6 +37,8 @@ public static IServiceCollection AddDurableMessaging(this IServiceCollection ser
     {
         services.AddDurableJobs();
         services.TryAddSingleton(TimeProvider.System);
+        services.PostConfigure<JournaledStateManagerOptions>(
+            options => options.JournalFormatKey = DurableMessagingJournalFormatKey);
 
         var optionsBuilder = services.AddOptions<DurableInboxOptions>();
         if (configureOptions is not null)

test/Orleans.DurableMessaging.Tests/Contracts/DeliveryAndOptionsContractTests.cs

@@ -1,5 +1,6 @@
 using System.Reflection;
 using System.Runtime.CompilerServices;
+using NSubstitute;
 using Orleans.DurableMessaging.Configuration;
 using Orleans.Runtime;
 using Xunit;
@@ -142,6 +143,10 @@ public async Task InboxLifecycleStart_CancellationInterruptsBlockedResume()
             "Orleans.DurableMessaging.DurableInboxExtension",
             throwOnError: true)!;
         var extension = (ILifecycleObserver)RuntimeHelpers.GetUninitializedObject(extensionType);
+        var grainContext = Substitute.For<IGrainContext>();
+        grainContext.GrainInstance.Returns(new object());
+        extensionType.GetField("_grainContext", BindingFlags.Instance | BindingFlags.NonPublic)!
+            .SetValue(extension, grainContext);
         extensionType.GetField("_gate", BindingFlags.Instance | BindingFlags.NonPublic)!
             .SetValue(extension, new SemaphoreSlim(0, 1));
         extensionType.GetField("_metricsActive", BindingFlags.Instance | BindingFlags.NonPublic)!

test/Orleans.DurableMessaging.Tests/Contracts/DurableOutboxDeliveryBatchTests.cs

@@ -576,6 +576,7 @@ public OutboxFixture(
             grainFactory.GetGrain<IDurableInboxExtension>(Arg.Any<GrainId>()).Returns(inbox);
             var grainContext = Substitute.For<IGrainContext>();
             grainContext.GrainId.Returns(SenderId);
+            grainContext.GrainInstance.Returns(new object());
             grainContext.ObservableLifecycle.Returns(Substitute.For<IGrainLifecycle>());
 
             TimerRegistry = timerRegistry ?? Substitute.For<ITimerRegistry>();

test/Orleans.DurableMessaging.Tests/Functional/MultiSiloDurableMessagingFailoverTests.cs

@@ -28,7 +28,7 @@ await sender.SendAsync(
             "messages/failover",
             new DurableTestMessage(logicalId, 81, "failover"));
         await barrier.WaitUntilEnteredAsync();
-        var before = await receiver.GetSnapshotAsync();
+        var before = fixture.GetSnapshot(receiver);
         var owner = fixture.Cluster.Silos.Single(
             silo => silo.SiloAddress.ToParsableString() == before.SiloAddress);
 

test/Orleans.DurableMessaging.Tests/Functional/PublicDurableMessagingBehaviorTests.cs

@@ -25,6 +25,12 @@ public sealed class PublicDurableMessagingBehaviorTests : IAsyncLifetime
     public ValueTask InitializeAsync() => fixture.InitializeAsync();
     public ValueTask DisposeAsync() => fixture.DisposeAsync();
 
+    [Fact]
+    public void DefaultHosting_UsesBinaryJournalFormat()
+    {
+        Assert.Equal("orleans-binary", fixture.Storage.JournalFormatKey);
+    }
+
     [Fact]
     public async Task ApplicationJournaledStateNamesDoNotCollideWithMessagingState()
     {
@@ -49,7 +55,7 @@ public async Task Deliver_AcceptedOnlyAfterInboxAndStableJobOwnershipAreDurable(
         await barrier.WaitUntilEnteredAsync();
 
         Assert.False(delivery.IsCompleted);
-        var staged = await receiver.GetSnapshotAsync();
+        var staged = fixture.GetSnapshot(receiver);
         Assert.Equal(1, staged.InboxCount);
         Assert.Empty(staged.Effects);
 
@@ -115,7 +121,7 @@ public async Task ConcurrentWriteCannotCaptureInboxAcceptanceBeforeScheduling()
         await schedule.WaitUntilEnteredAsync();
 
         var exception = await Assert.ThrowsAsync<InvalidOperationException>(
-            () => receiver.RetryWriteStateAsync());
+            () => fixture.WriteStateAsync(receiver).AsTask());
         Assert.Contains("waiting for job scheduling", exception.Message, StringComparison.Ordinal);
 
         schedule.Continue();
@@ -133,12 +139,12 @@ public async Task RecoveryDuringInboxScheduling_PreventsFalseAcceptance()
 
         var delivery = DeliverAsync(receiver, envelope.Value);
         await schedule.WaitUntilEnteredAsync();
-        await receiver.RevertStateAsync();
+        await fixture.RevertStateAsync(receiver);
         schedule.Continue();
 
         var exception = await Assert.ThrowsAsync<InvalidOperationException>(() => delivery);
         Assert.Contains("interrupted by state recovery", exception.Message, StringComparison.Ordinal);
-        Assert.Equal(0, (await receiver.GetSnapshotAsync()).InboxCount);
+        Assert.Equal(0, fixture.GetSnapshot(receiver).InboxCount);
     }
 
     [Fact]
@@ -348,7 +354,7 @@ public async Task RecoveryDuringHandler_LeavesMessageRetryable()
 
         Assert.Equal(DeliveryStatus.Accepted, (await DeliverAsync(receiver, envelope.Value)).Status);
         await handler.WaitUntilEnteredAsync();
-        await receiver.RevertStateAsync();
+        await fixture.RevertStateAsync(receiver);
         handler.Release();
 
         var completed = await fixture.WaitForEffectCountAsync(receiver, 1);
@@ -368,7 +374,7 @@ public async Task RecoveryDuringHandlerFailure_DiscardsStaleFailureAccounting()
 
         Assert.Equal(DeliveryStatus.Accepted, (await DeliverAsync(receiver, envelope.Value)).Status);
         await handler.WaitUntilEnteredAsync();
-        await receiver.RevertStateAsync();
+        await fixture.RevertStateAsync(receiver);
         handler.Release();
 
         var completed = await fixture.WaitForEffectCountAsync(receiver, 1);
@@ -390,7 +396,7 @@ public async Task ConcurrentDuplicateDeliveries_ConvergeToOneEffectWithinRetenti
 
         Assert.Equal(DeliveryStatus.Accepted, first.Status);
         Assert.False(second.IsCompleted);
-        Assert.Equal(1, (await receiver.GetSnapshotAsync()).InboxCount);
+        Assert.Equal(1, fixture.GetSnapshot(receiver).InboxCount);
 
         barrier.Release();
         Assert.Equal(DeliveryStatus.Duplicate, (await second).Status);
@@ -539,7 +545,7 @@ await sender.SendAndDeactivateAsync(
             "messages/outbox-crash-window",
             NewMessage(52, "durable-wakeup"));
         await receiverWrite.WaitUntilEnteredAsync();
-        var committed = await sender.GetSnapshotAsync();
+        var committed = fixture.GetSnapshot(sender);
 
         Assert.Equal(1, committed.OutboxCount);
         Assert.False(string.IsNullOrEmpty(committed.OutboxJobId));
@@ -733,7 +739,7 @@ public async Task InboxJobClearWriteFailure_RevertsThenRecoversAfterActivationLo
         Assert.Equal(DeliveryStatus.Accepted, (await DeliverAsync(receiver, envelope.Value)).Status);
         await handler.WaitUntilEnteredAsync();
         fixture.Storage.FailWrite(JournalId.FromGrainId(receiver.GetGrainId()), matchingWrite: 2);
-        await receiver.DeactivateOnNextRecoveryAsync();
+        fixture.DeactivateOnNextRecovery(receiver);
         handler.Release();
 
         var recovered = await fixture.SnapshotProbe.WaitAsync(
@@ -760,7 +766,7 @@ public async Task DeliveryIntoEmptyInbox_ReplacesStalePersistedJobOwnership()
 
         Assert.Equal(DeliveryStatus.Accepted, (await DeliverAsync(receiver, envelope.Value)).Status);
         await handler.WaitUntilEnteredAsync();
-        var accepted = await receiver.GetSnapshotAsync();
+        var accepted = fixture.GetSnapshot(receiver);
 
         Assert.NotNull(accepted.InboxJobId);
         Assert.NotEqual(staleJobId, accepted.InboxJobId);
@@ -780,7 +786,7 @@ public async Task Inbox_StaleGenerationCompletesWithoutClearingNewerOwner()
 
         Assert.Equal(DeliveryStatus.Accepted, (await DeliverAsync(receiver, envelope.Value)).Status);
         await handler.WaitUntilEnteredAsync();
-        var owned = await receiver.GetSnapshotAsync();
+        var owned = fixture.GetSnapshot(receiver);
         Assert.False(string.IsNullOrEmpty(owned.InboxJobId));
         var completionBaseline = fixture.Metrics.GetCount("orleans-durablejobs-jobs-completed");
         var manager = fixture.Cluster.Silos[0].ServiceProvider.GetRequiredService<ILocalDurableJobManager>();
@@ -801,7 +807,7 @@ await fixture.Metrics.WaitForCountAsync(
             "orleans-durablejobs-jobs-completed",
             completionBaseline + 1);
 
-        Assert.Equal(owned.InboxJobId, (await receiver.GetSnapshotAsync()).InboxJobId);
+        Assert.Equal(owned.InboxJobId, fixture.GetSnapshot(receiver).InboxJobId);
         handler.Release();
         Assert.Equal("newer-inbox-owner", Assert.Single((await fixture.WaitForEffectCountAsync(receiver, 1)).Effects).Value);
     }
@@ -820,7 +826,7 @@ await sender.SendAsync(
             "messages/stale-outbox-generation",
             NewMessage(61, "newer-outbox-owner"));
         await receiverWrite.WaitUntilEnteredAsync();
-        var owned = await sender.GetSnapshotAsync();
+        var owned = fixture.GetSnapshot(sender);
         Assert.False(string.IsNullOrEmpty(owned.OutboxJobId));
         var completionBaseline = fixture.Metrics.GetCount("orleans-durablejobs-jobs-completed");
         var manager = fixture.Cluster.Silos[0].ServiceProvider.GetRequiredService<ILocalDurableJobManager>();
@@ -843,7 +849,7 @@ await fixture.Metrics.WaitForCountAsync(
                 "orleans-durablejobs-jobs-completed",
                 completionBaseline + 1);
 
-            Assert.Equal(owned.OutboxJobId, (await sender.GetSnapshotAsync()).OutboxJobId);
+            Assert.Equal(owned.OutboxJobId, fixture.GetSnapshot(sender).OutboxJobId);
         }
         finally
         {
@@ -868,10 +874,10 @@ await sender.SendAsync(
             "messages/recovery-visibility",
             NewMessage(62, "recovery-visibility"));
         await receiverWrite.WaitUntilEnteredAsync();
-        var owned = await sender.GetSnapshotAsync();
+        var owned = fixture.GetSnapshot(sender);
         var ownershipId = Assert.IsType<string>(owned.OutboxJobId);
         var recoveryRead = fixture.Storage.BlockRead(JournalId.FromGrainId(sender.GetGrainId()));
-        var recovery = sender.RevertStateAsync();
+        var recovery = fixture.RevertStateAsync(sender).AsTask();
         await recoveryRead.WaitUntilEnteredAsync();
         var handlerBaseline = fixture.Metrics.GetCount("orleans-durablejobs-handler-executions-started");
         var completionBaseline = fixture.Metrics.GetCount("orleans-durablejobs-jobs-completed");
@@ -995,7 +1001,7 @@ await independentSender.SendAsync(
         var independent = await fixture.WaitForEffectCountAsync(independentReceiver, 1);
 
         Assert.Equal("independent", Assert.Single(independent.Effects).Value);
-        Assert.Empty((await blocked.GetSnapshotAsync()).Effects);
+        Assert.Empty(fixture.GetSnapshot(blocked).Effects);
         barrier.Release();
         Assert.Equal("blocked", Assert.Single((await fixture.WaitForEffectCountAsync(blocked, 1)).Effects).Value);
     }