fix(directory): deactivate recovery duplicates

src/Orleans.Runtime/Catalog/Catalog.cs

@@ -312,23 +312,25 @@ await Parallel.ForEachAsync(activations, options, (kv, _) =>
             }).WaitAsync(cancellationToken);
         }
 
-        public async Task DeleteActivations(
+        public Task DeleteActivations(
             List<GrainAddress> addresses,
             DeactivationReasonCode reasonCode,
             string reasonText,
             CancellationToken cancellationToken = default)
         {
+            cancellationToken.ThrowIfCancellationRequested();
             var deactivationReason = new DeactivationReason(reasonCode, reasonText);
-            await Parallel.ForEachAsync(addresses, cancellationToken, (activationAddress, cancellationToken) =>
+            foreach (var activationAddress in addresses)
             {
-                if (TryGetGrainContext(activationAddress.GrainId, out var grainContext))
+                if (TryGetGrainContext(activationAddress.GrainId, out var grainContext)
+                    && grainContext.Address.Equals(activationAddress))
                 {
-                    grainContext.Deactivate(deactivationReason, cancellationToken);
-                    return new ValueTask(grainContext.Deactivated);
+                    // Deactivate synchronously makes the exact loser unroutable; teardown and deregistration continue asynchronously.
+                    grainContext.Deactivate(deactivationReason, CancellationToken.None);
                 }
+            }
 
-                return ValueTask.CompletedTask;
-            });
+            return Task.CompletedTask;
         }
 
         void ILifecycleParticipant<ISiloLifecycle>.Participate(ISiloLifecycle lifecycle)

src/Orleans.Runtime/Catalog/ICatalog.cs

@@ -10,12 +10,17 @@ namespace Orleans.Runtime
     internal interface ICatalog : ISystemTarget
     {
         /// <summary>
-        /// Delete activations from this silo
+        /// Begins deactivating the specified activations on this silo.
         /// </summary>
-        /// <param name="activationAddresses"></param>
-        /// <param name="reasonCode"></param>
-        /// <param name="reasonText"></param>
-        /// <returns></returns>
+        /// <param name="activationAddresses">The exact activation addresses to deactivate.</param>
+        /// <param name="reasonCode">The reason code for deactivation.</param>
+        /// <param name="reasonText">The reason text for deactivation.</param>
+        /// <param name="cancellationToken">The token which cancels the request before deactivation begins.</param>
+        /// <returns>
+        /// A task which completes after each matching activation has entered deactivation and no longer accepts
+        /// application messages. Deactivation callbacks, directory deregistration, disposal, and removal continue
+        /// asynchronously.
+        /// </returns>
         [Alias("C4A56D7C")]
         Task DeleteActivations(
             List<GrainAddress> activationAddresses,

src/Orleans.Runtime/GrainDirectory/GrainDirectoryPartition.cs

@@ -18,6 +18,7 @@ internal sealed partial class GrainDirectoryPartition : SystemTarget, IGrainDire
 {
     private const float MaxActivationQueryRangePercent = 5f;
     private const int MaxActivationQueryRangeCount = 32;
+    private const int DuplicateActivationCleanupBatchSize = 1_000;
     private static readonly IBackoffProvider ClusterMemberRetryBackoff = new ExponentialBackoff(
         TimeSpan.FromMilliseconds(100),
         TimeSpan.FromSeconds(5),
@@ -793,6 +794,7 @@ private async Task RecoverPartitionRange(DirectoryMembershipSnapshot current, Ri
         var stopwatch = ValueStopwatch.StartNew();
         GrainRuntime.CheckRuntimeContext(this);
         LogDebugRecoveringActivations(_logger, addedRange, current.Version);
+        Dictionary<SiloAddress, List<GrainAddress>>? duplicateActivations = null;
 
         await foreach (var activations in GetRegisteredActivations(current, addedRange, isValidation: false))
         {
@@ -801,26 +803,100 @@ private async Task RecoverPartitionRange(DirectoryMembershipSnapshot current, Ri
             {
                 DebugAssertOwnership(current, entry.GrainId);
                 LogTraceRecoveredEntry(_logger, entry, current.Version);
-                RecoverEntry(_directory, entry);
+                if (RecoverEntry(_directory, entry) is { SiloAddress: { } siloAddress } duplicate)
+                {
+                    duplicateActivations ??= [];
+                    if (!duplicateActivations.TryGetValue(siloAddress, out var duplicates))
+                    {
+                        duplicateActivations[siloAddress] = duplicates = [];
+                    }
+
+                    duplicates.Add(duplicate);
+                }
             }
         }
 
+        if (duplicateActivations is not null)
+        {
+            RemoveRecoveredWinners(_directory, duplicateActivations);
+            await DeactivateDuplicateActivationsAsync(duplicateActivations);
+        }
+
         _directoryInstruments.RangeRecoveryCount.Add(1);
         _directoryInstruments.RangeRecoveryDuration.Record((long)stopwatch.Elapsed.TotalMilliseconds);
         LogDebugCompletedRecoveringActivations(_logger, addedRange, current.Version, stopwatch.Elapsed);
     }
 
-    internal static void RecoverEntry(Dictionary<GrainId, GrainAddress> directory, GrainAddress recovered)
+    internal static GrainAddress? RecoverEntry(Dictionary<GrainId, GrainAddress> directory, GrainAddress recovered)
     {
         // During a rolling upgrade, LocalGrainDirectory does not participate in DistributedGrainDirectory's
         // recovery-registration barrier and can report an activation superseded in a newer membership view.
         // This is the only expected case where recovery returns different registrations for one grain. Preserve
         // the newest view while retaining recovery's existing last-response tie-breaking within the same view.
-        if (!directory.TryGetValue(recovered.GrainId, out var existing)
-            || recovered.MembershipVersion >= existing.MembershipVersion)
+        if (!directory.TryGetValue(recovered.GrainId, out var existing))
         {
             directory[recovered.GrainId] = recovered;
+            return null;
         }
+
+        if (recovered.Equals(existing))
+        {
+            if (recovered.MembershipVersion > existing.MembershipVersion)
+            {
+                directory[recovered.GrainId] = recovered;
+            }
+
+            return null;
+        }
+
+        if (recovered.MembershipVersion >= existing.MembershipVersion)
+        {
+            directory[recovered.GrainId] = recovered;
+            return existing;
+        }
+
+        return recovered;
+    }
+
+    internal static void RemoveRecoveredWinners(
+        Dictionary<GrainId, GrainAddress> directory,
+        Dictionary<SiloAddress, List<GrainAddress>> duplicateActivations)
+    {
+        foreach (var duplicates in duplicateActivations.Values)
+        {
+            duplicates.RemoveAll(candidate =>
+                directory.TryGetValue(candidate.GrainId, out var winner)
+                && candidate.Equals(winner));
+        }
+    }
+
+    private async Task DeactivateDuplicateActivationsAsync(
+        Dictionary<SiloAddress, List<GrainAddress>> duplicateActivations)
+    {
+        var tasks = duplicateActivations
+            .Where(static pair => pair.Value.Count > 0)
+            .Select(async pair =>
+            {
+                var catalog = _grainFactory.GetSystemTarget<ICatalog>(Constants.CatalogType, pair.Key);
+                foreach (var batch in pair.Value.Chunk(DuplicateActivationCleanupBatchSize))
+                {
+                    await InvokeOnClusterMember(
+                        pair.Key,
+                        async cancellationToken =>
+                        {
+                            await catalog.DeleteActivations(
+                                [.. batch],
+                                DeactivationReasonCode.DuplicateActivation,
+                                "This grain has been activated elsewhere",
+                                cancellationToken).WaitAsync(cancellationToken);
+                            return true;
+                        },
+                        false,
+                        nameof(ICatalog.DeleteActivations));
+                }
+            });
+
+        await Task.WhenAll(tasks);
     }
 
     private async IAsyncEnumerable<List<GrainAddress>> GetRegisteredActivations(DirectoryMembershipSnapshot current, RingRange range, bool isValidation)

test/Orleans.GrainDirectory.Tests/GrainDirectory/GrainDirectoryPartitionTests.cs

@@ -174,7 +174,7 @@ public void GetSnapshotTransferRanges_ReturnsOnlyPreviousOwnerIntersections()
     }
 
     [Fact]
-    public void RecoverEntry_PreservesNewestRegistrationRegardlessOfResponseOrder()
+    public void RecoverEntry_PreservesNewestRegistrationAndReportsDuplicateRegardlessOfResponseOrder()
     {
         var grainId = GrainId.Create("recovery-test", "grain");
         var oldRegistration = new GrainAddress
@@ -193,17 +193,111 @@ public void RecoverEntry_PreservesNewestRegistrationRegardlessOfResponseOrder()
         };
 
         Dictionary<GrainId, GrainAddress> oldThenNew = [];
-        GrainDirectoryPartition.RecoverEntry(oldThenNew, oldRegistration);
-        GrainDirectoryPartition.RecoverEntry(oldThenNew, newRegistration);
+        var oldThenNewInitialDuplicate = GrainDirectoryPartition.RecoverEntry(oldThenNew, oldRegistration);
+        var oldThenNewDuplicate = GrainDirectoryPartition.RecoverEntry(oldThenNew, newRegistration);
 
         Dictionary<GrainId, GrainAddress> newThenOld = [];
-        GrainDirectoryPartition.RecoverEntry(newThenOld, newRegistration);
-        GrainDirectoryPartition.RecoverEntry(newThenOld, oldRegistration);
+        var newThenOldInitialDuplicate = GrainDirectoryPartition.RecoverEntry(newThenOld, newRegistration);
+        var newThenOldDuplicate = GrainDirectoryPartition.RecoverEntry(newThenOld, oldRegistration);
 
+        Assert.Null(oldThenNewInitialDuplicate);
+        Assert.Equal(oldRegistration, oldThenNewDuplicate);
         Assert.Equal(newRegistration, oldThenNew[grainId]);
+        Assert.Null(newThenOldInitialDuplicate);
+        Assert.Equal(oldRegistration, newThenOldDuplicate);
         Assert.Equal(newRegistration, newThenOld[grainId]);
     }
 
+    [Fact]
+    public void RemoveRecoveredWinners_DoesNotDeactivateWinnerReobservedAtNewerVersion()
+    {
+        var grainId = GrainId.Create("recovery-test", "grain");
+        var activationId = ActivationId.NewId();
+        var initialRegistration = new GrainAddress
+        {
+            GrainId = grainId,
+            ActivationId = activationId,
+            SiloAddress = TestSiloAddress,
+            MembershipVersion = new MembershipVersion(1),
+        };
+        var intermediateRegistration = new GrainAddress
+        {
+            GrainId = grainId,
+            ActivationId = ActivationId.NewId(),
+            SiloAddress = ReplacementSiloAddress,
+            MembershipVersion = new MembershipVersion(2),
+        };
+        var finalRegistration = new GrainAddress
+        {
+            GrainId = grainId,
+            ActivationId = activationId,
+            SiloAddress = TestSiloAddress,
+            MembershipVersion = new MembershipVersion(3),
+        };
+        Dictionary<GrainId, GrainAddress> directory = [];
+        Dictionary<SiloAddress, List<GrainAddress>> duplicateActivations = [];
+
+        AddDuplicate(GrainDirectoryPartition.RecoverEntry(directory, initialRegistration));
+        AddDuplicate(GrainDirectoryPartition.RecoverEntry(directory, intermediateRegistration));
+        AddDuplicate(GrainDirectoryPartition.RecoverEntry(directory, finalRegistration));
+        GrainDirectoryPartition.RemoveRecoveredWinners(directory, duplicateActivations);
+
+        Assert.Same(finalRegistration, directory[grainId]);
+        Assert.Empty(duplicateActivations[TestSiloAddress]);
+        Assert.Equal([intermediateRegistration], duplicateActivations[ReplacementSiloAddress]);
+
+        void AddDuplicate(GrainAddress? duplicate)
+        {
+            if (duplicate?.SiloAddress is not { } siloAddress)
+            {
+                return;
+            }
+
+            if (!duplicateActivations.TryGetValue(siloAddress, out var duplicates))
+            {
+                duplicateActivations[siloAddress] = duplicates = [];
+            }
+
+            duplicates.Add(duplicate);
+        }
+    }
+
+    [Fact]
+    public void RecoverEntry_RefreshesMembershipVersionForSameActivation()
+    {
+        var grainId = GrainId.Create("recovery-test", "grain");
+        var activationId = ActivationId.NewId();
+        var initialRegistration = new GrainAddress
+        {
+            GrainId = grainId,
+            ActivationId = activationId,
+            SiloAddress = TestSiloAddress,
+            MembershipVersion = new MembershipVersion(1),
+        };
+        var refreshedRegistration = new GrainAddress
+        {
+            GrainId = grainId,
+            ActivationId = activationId,
+            SiloAddress = TestSiloAddress,
+            MembershipVersion = new MembershipVersion(3),
+        };
+        var intermediateRegistration = new GrainAddress
+        {
+            GrainId = grainId,
+            ActivationId = ActivationId.NewId(),
+            SiloAddress = ReplacementSiloAddress,
+            MembershipVersion = new MembershipVersion(2),
+        };
+        Dictionary<GrainId, GrainAddress> directory = [];
+
+        Assert.Null(GrainDirectoryPartition.RecoverEntry(directory, initialRegistration));
+        Assert.Null(GrainDirectoryPartition.RecoverEntry(directory, refreshedRegistration));
+        var duplicate = GrainDirectoryPartition.RecoverEntry(directory, intermediateRegistration);
+
+        Assert.Same(refreshedRegistration, directory[grainId]);
+        Assert.Equal(intermediateRegistration, duplicate);
+    }
+
     private static void AssertRanges(RingRange previousOwnerRange, RingRange addedRange, params RingRange[] expected)
     {
         var actual = GrainDirectoryPartition.GetSnapshotTransferRanges(previousOwnerRange, addedRange).ToArray();