[Membership] Addressing recovery from contention on membership writes (#9376)

rkargMsft · ReubenBond · web-flow · commit c723211cfdcf · 2025-03-24T11:50:39.000-07:00
Co-authored-by: Reuben Bond &lt;reuben.bond@gmail.com&gt;
diff --git a/src/Orleans.Runtime/MembershipService/MembershipAgent.cs b/src/Orleans.Runtime/MembershipService/MembershipAgent.cs
@@ -15,6 +15,9 @@ namespace Orleans.Runtime.MembershipService
     /// </summary>
     internal class MembershipAgent : IHealthCheckParticipant, ILifecycleParticipant<ISiloLifecycle>, IDisposable, MembershipAgent.ITestAccessor
     {
+        private static readonly TimeSpan EXP_BACKOFF_CONTENTION_MIN = TimeSpan.FromMilliseconds(200);
+        private static readonly TimeSpan EXP_BACKOFF_CONTENTION_MAX = TimeSpan.FromMinutes(2);
+        private static readonly TimeSpan EXP_BACKOFF_STEP = TimeSpan.FromMilliseconds(1000);
         private readonly CancellationTokenSource cancellation = new CancellationTokenSource();
         private readonly MembershipTableManager tableManager;
         private readonly ILocalSiloDetails localSilo;
@@ -59,27 +62,31 @@ private async Task UpdateIAmAlive()
             if (this.log.IsEnabled(LogLevel.Debug)) this.log.LogDebug("Starting periodic membership liveness timestamp updates");
             try
             {
-                TimeSpan? onceOffDelay = default;
-                while (await this.iAmAliveTimer.NextTick(onceOffDelay) && !this.tableManager.CurrentStatus.IsTerminating())
+                // jitter for initial
+                TimeSpan? overrideDelayPeriod = RandomTimeSpan.Next(this.clusterMembershipOptions.IAmAliveTablePublishTimeout);
+                var exponentialBackoff = new ExponentialBackoff(EXP_BACKOFF_CONTENTION_MIN, EXP_BACKOFF_CONTENTION_MAX, EXP_BACKOFF_STEP);
+                var runningFailures = 0;
+                while (await this.iAmAliveTimer.NextTick(overrideDelayPeriod) && !this.tableManager.CurrentStatus.IsTerminating())
                 {
-                    onceOffDelay = default;
-
                     try
                     {
                         var stopwatch = ValueStopwatch.StartNew();
                         ((ITestAccessor)this).OnUpdateIAmAlive?.Invoke();
                         await this.tableManager.UpdateIAmAlive();
                         if (this.log.IsEnabled(LogLevel.Trace)) this.log.LogTrace("Updating IAmAlive took {Elapsed}", stopwatch.Elapsed);
+                        overrideDelayPeriod = default;
+                        runningFailures = 0;
                     }
                     catch (Exception exception)
                     {
+                        runningFailures += 1;
                         this.log.LogWarning(
                             (int)ErrorCode.MembershipUpdateIAmAliveFailure,
                             exception,
                             "Failed to update table entry for this silo, will retry shortly");
 
-                        // Retry quickly
-                        onceOffDelay = TimeSpan.FromMilliseconds(200);
+                        // Retry quickly and then exponentially back off
+                        overrideDelayPeriod = exponentialBackoff.Next(runningFailures);
                     }
                 }
             }
diff --git a/src/Orleans.Runtime/MembershipService/MembershipTableManager.cs b/src/Orleans.Runtime/MembershipService/MembershipTableManager.cs
@@ -4,13 +4,13 @@
 using System.Linq;
 using System.Reflection;
 using System.Threading;
+using System.Threading.Channels;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using Orleans.Configuration;
 using Orleans.Internal;
 using Orleans.Runtime.Utilities;
-using Orleans.Serialization;
 using Orleans.Serialization.TypeSystem;
 
 namespace Orleans.Runtime.MembershipService
@@ -38,6 +38,13 @@ internal class MembershipTableManager : IHealthCheckParticipant, ILifecycleParti
         private readonly SiloAddress myAddress;
         private readonly AsyncEnumerable<MembershipTableSnapshot> updates;
         private readonly IAsyncTimer membershipUpdateTimer;
+        private readonly CancellationTokenSource _shutdownCts = new();
+
+        private readonly Task _suspectOrKillsListTask;
+        private readonly Channel<SuspectOrKillRequest> _trySuspectOrKillChannel = Channel.CreateBounded<SuspectOrKillRequest>(new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.DropOldest });
+
+        // For testing.
+        internal AutoResetEvent TestingSuspectOrKillIdle = new(false);
 
         private MembershipTableSnapshot snapshot;
 
@@ -71,6 +78,8 @@ public MembershipTableManager(
             this.membershipUpdateTimer = timerFactory.Create(
                 this.clusterMembershipOptions.TableRefreshTimeout,
                 nameof(PeriodicallyRefreshMembershipTable));
+
+            _suspectOrKillsListTask = Task.Run(ProcessSuspectOrKillLists);
         }
 
         internal Func<DateTime> GetDateTimeUtcNow { get; set; } = () => DateTime.UtcNow;
@@ -243,28 +252,32 @@ private async Task PeriodicallyRefreshMembershipTable()
             if (this.log.IsEnabled(LogLevel.Debug)) this.log.LogDebug("Starting periodic membership table refreshes");
             try
             {
-                var targetMilliseconds = (int)this.clusterMembershipOptions.TableRefreshTimeout.TotalMilliseconds;
-                
-                TimeSpan? onceOffDelay = RandomTimeSpan.Next(this.clusterMembershipOptions.TableRefreshTimeout);
-                while (await this.membershipUpdateTimer.NextTick(onceOffDelay))
+                // jitter for initial
+                TimeSpan? overrideDelayPeriod = RandomTimeSpan.Next(this.clusterMembershipOptions.TableRefreshTimeout);
+                var exponentialBackoff = new ExponentialBackoff(EXP_BACKOFF_CONTENTION_MIN, EXP_BACKOFF_CONTENTION_MAX, EXP_BACKOFF_STEP);
+                var runningFailures = 0;
+                while (await this.membershipUpdateTimer.NextTick(overrideDelayPeriod))
                 {
-                    onceOffDelay = default;
-
                     try
                     {
                         var stopwatch = ValueStopwatch.StartNew();
+                        
                         await this.Refresh();
                         if (this.log.IsEnabled(LogLevel.Trace)) this.log.LogTrace("Refreshing membership table took {Elapsed}", stopwatch.Elapsed);
+                        // reset to allow normal refresh period after success
+                        overrideDelayPeriod = default;
+                        runningFailures = 0;
                     }
                     catch (Exception exception)
                     {
+                        runningFailures += 1;
                         this.log.LogWarning(
                             (int)ErrorCode.MembershipUpdateIAmAliveFailure,
                             exception,
-                            "Failed to refresh membership table, will retry shortly");
+                            "Failed to refresh membership table, will retry shortly. Retry attempt {retries}", runningFailures);
 
-                        // Retry quickly
-                        onceOffDelay = TimeSpan.FromMilliseconds(200);
+                        // Retry quickly and then exponentially back off
+                        overrideDelayPeriod = exponentialBackoff.Next(runningFailures);
                     }
                 }
             }
@@ -584,19 +597,13 @@ private async Task<bool> CleanupMyTableEntries(MembershipTableData table)
             if (log.IsEnabled(LogLevel.Debug)) log.LogDebug("CleanupTableEntries: About to DeclareDead {Count} outdated silos in the table: {Silos}", silosToDeclareDead.Count,
                 Utils.EnumerableToString(silosToDeclareDead.Select(tuple => tuple.Item1)));
 
-            var result = true;
-            var nextVersion = table.Version;
-
             foreach (var siloData in silosToDeclareDead)
             {
-                MembershipEntry entry = siloData.Item1;
-                string eTag = siloData.Item2;
-                bool ok = await DeclareDead(entry, eTag, nextVersion, GetDateTimeUtcNow());
-                if (!ok) result = false;
-                nextVersion = nextVersion.Next(); // advance the table version (if write succeded, we advanced the version. if failed, someone else did. It is safe anyway).
+                await _trySuspectOrKillChannel.Writer.WriteAsync(
+                    SuspectOrKillRequest.CreateKillRequest(siloData.Item1.SiloAddress));
             }
 
-            return result;
+            return true;
         }
 
         private void KillMyselfLocally(string reason)
@@ -637,9 +644,89 @@ bool IsFunctionalForMembership(SiloStatus status)
             }
         }
 
+        private class SuspectOrKillRequest
+        {
+            public SiloAddress SiloAddress { get; set; }
+            public SiloAddress OtherSilo { get; set; }
+            public RequestType Type { get; set; }
+
+            public enum RequestType
+            {
+                Unknown = 0,
+                SuspectOrKill,
+                Kill
+            }
+
+            public static SuspectOrKillRequest CreateKillRequest(SiloAddress silo)
+            {
+                return new SuspectOrKillRequest
+                {
+                    SiloAddress = silo,
+                    OtherSilo = null,
+                    Type = RequestType.Kill
+                };
+            }
+
+            public static SuspectOrKillRequest CreateSuspectOrKillRequest(SiloAddress silo, SiloAddress otherSilo)
+            {
+                return new SuspectOrKillRequest
+                {
+                    SiloAddress = silo,
+                    OtherSilo = otherSilo,
+                    Type = RequestType.SuspectOrKill
+                };
+            }
+        }
+
         public async Task<bool> TryKill(SiloAddress silo)
         {
-            var table = await membershipTableProvider.ReadAll();
+            await _trySuspectOrKillChannel.Writer.WriteAsync(SuspectOrKillRequest.CreateKillRequest(silo));
+            return true;
+        }
+
+        public async Task ProcessSuspectOrKillLists()
+        {
+            var backoff = new ExponentialBackoff(EXP_BACKOFF_ERROR_MIN, EXP_BACKOFF_ERROR_MAX,
+                EXP_BACKOFF_STEP);
+            var runningFailureCount = 0;
+            var reader = _trySuspectOrKillChannel.Reader;
+            while (await reader.WaitToReadAsync(_shutdownCts.Token))
+            {
+                while (reader.TryRead(out var request))
+                {
+                    await Task.Delay(backoff.Next(runningFailureCount), _shutdownCts.Token);
+
+                    try
+                    {
+                        switch (request.Type)
+                        {
+                            case SuspectOrKillRequest.RequestType.Kill:
+                                await InnerTryKill(request.SiloAddress, _shutdownCts.Token);
+                                break;
+                            case SuspectOrKillRequest.RequestType.SuspectOrKill:
+                                await InnerTryToSuspectOrKill(request.SiloAddress, request.OtherSilo, _shutdownCts.Token);
+                                break;
+                        }
+                        runningFailureCount = 0;
+                    }
+                    catch (Exception ex)
+                    {
+                        runningFailureCount += 1;
+                        log.LogError(ex, "Error while processing suspect or kill lists. '{FailureCount}' consecutive failures.", runningFailureCount);
+                        await _trySuspectOrKillChannel.Writer.WriteAsync(request, _shutdownCts.Token);
+                    }
+
+                    if (!reader.TryPeek(out _))
+                    {
+                        TestingSuspectOrKillIdle.Set();
+                    }
+                }
+            }
+        }
+
+        private async Task<bool> InnerTryKill(SiloAddress silo, CancellationToken cancellationToken)
+        {
+            var table = await membershipTableProvider.ReadAll().WaitAsync(cancellationToken);
 
             if (log.IsEnabled(LogLevel.Debug))
             {
@@ -685,12 +772,18 @@ public async Task<bool> TryKill(SiloAddress silo)
                 (int)ErrorCode.MembershipMarkingAsDead,
                 "Going to mark silo {SiloAddress} dead as a result of a call to TryKill",
                 entry.SiloAddress);
-            return await DeclareDead(entry, eTag, table.Version, GetDateTimeUtcNow());
+            return await DeclareDead(entry, eTag, table.Version, GetDateTimeUtcNow()).WaitAsync(cancellationToken);
         }
 
         public async Task<bool> TryToSuspectOrKill(SiloAddress silo, SiloAddress indirectProbingSilo = null)
         {
-            var table = await membershipTableProvider.ReadAll();
+            await _trySuspectOrKillChannel.Writer.WriteAsync(SuspectOrKillRequest.CreateSuspectOrKillRequest(silo, indirectProbingSilo));
+            return true;
+        }
+
+        private async Task<bool> InnerTryToSuspectOrKill(SiloAddress silo, SiloAddress indirectProbingSilo, CancellationToken cancellationToken)
+        {
+            var table = await membershipTableProvider.ReadAll().WaitAsync(cancellationToken);
             var now = GetDateTimeUtcNow();
 
             if (log.IsEnabled(LogLevel.Debug)) log.LogDebug("TryToSuspectOrKill: Read Membership table {Table}", table.ToString());
@@ -720,7 +813,9 @@ public async Task<bool> TryToSuspectOrKill(SiloAddress silo, SiloAddress indirec
             {
                 // this should not happen ...
                 log.LogError((int)ErrorCode.MembershipFailedToReadSilo, "Could not find silo entry for silo {Silo} in the table.", silo);
-                throw new KeyNotFoundException($"Could not find silo entry for silo {silo} in the table.");
+                //What is a caller going to do? The silo is not in the table which is what we are trying to achieve.
+                //throw new KeyNotFoundException($"Could not find silo entry for silo {silo} in the table.");
+                return false;
             }
 
             var entry = tuple.Item1.Copy();
@@ -794,7 +889,7 @@ public async Task<bool> TryToSuspectOrKill(SiloAddress silo, SiloAddress indirec
                     activeNonStaleSilos,
                     PrintSuspectList(entry.SuspectTimes));
 
-                return await DeclareDead(entry, eTag, table.Version, now);
+                return await DeclareDead(entry, eTag, table.Version, now).WaitAsync(cancellationToken);
             }
 
             log.LogInformation(
@@ -807,10 +902,10 @@ public async Task<bool> TryToSuspectOrKill(SiloAddress silo, SiloAddress indirec
                 PrintSuspectList(freshVotes));
 
             // If we fail to update here we will retry later.
-            var ok = await membershipTableProvider.UpdateRow(entry, eTag, table.Version.Next());
+            var ok = await membershipTableProvider.UpdateRow(entry, eTag, table.Version.Next()).WaitAsync(cancellationToken);
             if (ok)
             {
-                table = await membershipTableProvider.ReadAll();
+                table = await membershipTableProvider.ReadAll().WaitAsync(cancellationToken);
                 this.ProcessTableUpdate(table, "TrySuspectOrKill");
 
                 // Gossip using the local silo status, since this is just informational to propagate the suspicion vote.
@@ -877,18 +972,22 @@ async Task OnRuntimeGrainServicesStart(CancellationToken ct)
 
             async Task OnRuntimeGrainServicesStop(CancellationToken ct)
             {
+                tasks.Add(_suspectOrKillsListTask);
+                _trySuspectOrKillChannel.Writer.TryComplete();
                 this.membershipUpdateTimer.Dispose();
+                _shutdownCts.Cancel();
 
                 // Allow some minimum time for graceful shutdown.
                 var gracePeriod = Task.WhenAll(Task.Delay(ClusterMembershipOptions.ClusteringShutdownGracePeriod), ct.WhenCancelled());
-                await Task.WhenAny(gracePeriod, Task.WhenAll(tasks));
+                await Task.WhenAny(gracePeriod, Task.WhenAll(tasks)).SuppressThrowing();
             }
         }
 
         public void Dispose()
         {
             this.updates.Dispose();
             this.membershipUpdateTimer.Dispose();
+            _shutdownCts.Dispose();
         }
     }
 }
diff --git a/test/NonSilo.Tests/Membership/ClusterHealthMonitorTests.cs b/test/NonSilo.Tests/Membership/ClusterHealthMonitorTests.cs
@@ -284,6 +284,7 @@ await UntilEqual(expectedNumProbedSilos, () =>
 
                 while (probeCalls.TryDequeue(out var call)) ;
 
+                testRig.Manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(45));
                 // Check that probes match the expected missed probes
                 table = await this.membershipTable.ReadAll();
                 foreach (var siloMonitor in monitoredSilos)
@@ -463,7 +464,9 @@ private async Task ClusterHealthMonitor_StaleJoinOrCreatedSilos_Runner(bool evic
 
             await testRig.Manager.Refresh();
 
+            testRig.Manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(45));
             await Until(() => testRig.TestAccessor.ObservedVersion > lastVersion);
+            
             lastVersion = testRig.TestAccessor.ObservedVersion;
 
             table = await this.membershipTable.ReadAll();
diff --git a/test/NonSilo.Tests/Membership/MembershipTableManagerTests.cs b/test/NonSilo.Tests/Membership/MembershipTableManagerTests.cs
@@ -462,6 +462,7 @@ public async Task MembershipTableManager_TrySuspectOrKill_ButIAmKill()
 
             this.fatalErrorHandler.DidNotReceiveWithAnyArgs().OnFatalException(default, default, default);
             await manager.TryToSuspectOrKill(otherSilos.First().SiloAddress);
+            manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(45));
             this.fatalErrorHandler.ReceivedWithAnyArgs().OnFatalException(default, default, default);
         }
 
@@ -526,6 +527,7 @@ public async Task MembershipTableManager_TrySuspectOrKill_DeclareDead_SmallClust
 
             var victim = otherSilos.First().SiloAddress;
             await manager.TryToSuspectOrKill(victim);
+            manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(45));
             Assert.Equal(SiloStatus.Dead, manager.MembershipTableSnapshot.GetSiloStatus(victim));
         }
 
@@ -638,14 +640,12 @@ public async Task MembershipTableManager_TrySuspectOrKill_DeclareDead_LargerClus
             await this.lifecycle.OnStart();
             await manager.UpdateStatus(SiloStatus.Active);
 
-            // Try to declare an unknown silo dead
-            await Assert.ThrowsAsync<KeyNotFoundException>(() => manager.TryToSuspectOrKill(Silo("123.123.123.123:1@1")));
-
             // Multiple votes from the same node should not result in the node being declared dead.
             var victim = otherSilos.First().SiloAddress;
             await manager.TryToSuspectOrKill(victim);
             await manager.TryToSuspectOrKill(victim);
             await manager.TryToSuspectOrKill(victim);
+            manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(45));
             Assert.Equal(SiloStatus.Active, manager.MembershipTableSnapshot.GetSiloStatus(victim));
 
             // Manually remove our vote and add another silo's vote so we can be the one to kill the silo.
@@ -660,6 +660,7 @@ public async Task MembershipTableManager_TrySuspectOrKill_DeclareDead_LargerClus
             }
 
             await manager.TryToSuspectOrKill(victim);
+            manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(45));
             Assert.Equal(SiloStatus.Dead, manager.MembershipTableSnapshot.GetSiloStatus(victim));
 
             // One down, one to go. Now overshoot votes and kill ourselves instead (due to internal error).
@@ -678,6 +679,7 @@ public async Task MembershipTableManager_TrySuspectOrKill_DeclareDead_LargerClus
 
             this.fatalErrorHandler.DidNotReceiveWithAnyArgs().OnFatalException(default, default, default);
             await manager.TryToSuspectOrKill(victim);
+            manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(45));
             this.fatalErrorHandler.ReceivedWithAnyArgs().OnFatalException(default, default, default);
 
             // We killed ourselves and should not have marked the other silo as dead.

Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,9 @@ namespace Orleans.Runtime.MembershipService`
`15`	`15`	`/// </summary>`
`16`	`16`	`internal class MembershipAgent : IHealthCheckParticipant, ILifecycleParticipant<ISiloLifecycle>, IDisposable, MembershipAgent.ITestAccessor`
`17`	`17`	`{`
	`18`	`+ private static readonly TimeSpan EXP_BACKOFF_CONTENTION_MIN = TimeSpan.FromMilliseconds(200);`
	`19`	`+ private static readonly TimeSpan EXP_BACKOFF_CONTENTION_MAX = TimeSpan.FromMinutes(2);`
	`20`	`+ private static readonly TimeSpan EXP_BACKOFF_STEP = TimeSpan.FromMilliseconds(1000);`
`18`	`21`	`private readonly CancellationTokenSource cancellation = new CancellationTokenSource();`
`19`	`22`	`private readonly MembershipTableManager tableManager;`
`20`	`23`	`private readonly ILocalSiloDetails localSilo;`
`@@ -59,27 +62,31 @@ private async Task UpdateIAmAlive()`
`59`	`62`	`if (this.log.IsEnabled(LogLevel.Debug)) this.log.LogDebug("Starting periodic membership liveness timestamp updates");`
`60`	`63`	`try`
`61`	`64`	`{`
`62`		`- TimeSpan? onceOffDelay = default;`
`63`		`- while (await this.iAmAliveTimer.NextTick(onceOffDelay) && !this.tableManager.CurrentStatus.IsTerminating())`
	`65`	`+ // jitter for initial`
	`66`	`+ TimeSpan? overrideDelayPeriod = RandomTimeSpan.Next(this.clusterMembershipOptions.IAmAliveTablePublishTimeout);`
	`67`	`+ var exponentialBackoff = new ExponentialBackoff(EXP_BACKOFF_CONTENTION_MIN, EXP_BACKOFF_CONTENTION_MAX, EXP_BACKOFF_STEP);`
	`68`	`+ var runningFailures = 0;`
	`69`	`+ while (await this.iAmAliveTimer.NextTick(overrideDelayPeriod) && !this.tableManager.CurrentStatus.IsTerminating())`
`64`	`70`	`{`
`65`		`- onceOffDelay = default;`
`66`		`-`
`67`	`71`	`try`
`68`	`72`	`{`
`69`	`73`	`var stopwatch = ValueStopwatch.StartNew();`
`70`	`74`	`((ITestAccessor)this).OnUpdateIAmAlive?.Invoke();`
`71`	`75`	`await this.tableManager.UpdateIAmAlive();`
`72`	`76`	`if (this.log.IsEnabled(LogLevel.Trace)) this.log.LogTrace("Updating IAmAlive took {Elapsed}", stopwatch.Elapsed);`
	`77`	`+ overrideDelayPeriod = default;`
	`78`	`+ runningFailures = 0;`
`73`	`79`	`}`
`74`	`80`	`catch (Exception exception)`
`75`	`81`	`{`
	`82`	`+ runningFailures += 1;`
`76`	`83`	`this.log.LogWarning(`
`77`	`84`	`(int)ErrorCode.MembershipUpdateIAmAliveFailure,`
`78`	`85`	`exception,`
`79`	`86`	`"Failed to update table entry for this silo, will retry shortly");`
`80`	`87`
`81`		`- // Retry quickly`
`82`		`- onceOffDelay = TimeSpan.FromMilliseconds(200);`
	`88`	`+ // Retry quickly and then exponentially back off`
	`89`	`+ overrideDelayPeriod = exponentialBackoff.Next(runningFailures);`
`83`	`90`	`}`
`84`	`91`	`}`
`85`	`92`	`}`