From 78a4c2e8418334b1c93cd4ce6ca18c31d708a1a8 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Tue, 31 Mar 2026 12:17:56 -0700 Subject: [PATCH 1/7] Add connection liveness check for fault detection --- .../Options/ClusterMembershipOptions.cs | 12 + src/Orleans.Core/Networking/Connection.cs | 24 +- .../Networking/ConnectionManager.cs | 26 +++ src/Orleans.Core/Timers/CoarseStopwatch.cs | 2 +- .../MembershipService/ClusterHealthMonitor.cs | 46 +++- .../Membership/ClusterHealthMonitorTests.cs | 209 +++++++++++++++++- .../Membership/MembershipAgentTests.cs | 7 +- 7 files changed, 321 insertions(+), 5 deletions(-) diff --git a/src/Orleans.Core/Configuration/Options/ClusterMembershipOptions.cs b/src/Orleans.Core/Configuration/Options/ClusterMembershipOptions.cs index 057b2321b67..5ab4d8bbc87 100644 --- a/src/Orleans.Core/Configuration/Options/ClusterMembershipOptions.cs +++ b/src/Orleans.Core/Configuration/Options/ClusterMembershipOptions.cs @@ -138,6 +138,18 @@ public class ClusterMembershipOptions /// public bool EnableIndirectProbes { get; set; } = true; + /// + /// Gets or sets a value indicating whether to consider connection-level message activity when evaluating silo liveness. + /// + /// + /// When enabled, if an active connection to a silo has recently received messages within the monitoring window + /// ( × ), votes to suspect that silo will be suppressed + /// since the connection activity demonstrates the silo is alive. This helps prevent false death declarations + /// when probes fail due to local issues such as GC pauses or thread pool saturation. + /// + /// Connection liveness checks are enabled by default. + public bool EnableConnectionLivenessCheck { get; set; } = true; + /// /// /// Gets or sets a value indicating whether to enable membership eviction of silos when they remain in the Joining or Created state for longer than . /// diff --git a/src/Orleans.Core/Networking/Connection.cs b/src/Orleans.Core/Networking/Connection.cs index 0160e4e0d11..10109af7e26 100644 --- a/src/Orleans.Core/Networking/Connection.cs +++ b/src/Orleans.Core/Networking/Connection.cs @@ -39,6 +39,7 @@ internal abstract partial class Connection private Task? _processIncomingTask; private Task? _processOutgoingTask; private Task? _closeTask; + private CoarseStopwatch _lastMessageReceivedTimestamp; protected Connection( ConnectionContext connection, @@ -74,6 +75,19 @@ protected Connection( public Task Initialized => _initializationTcs.Task; + /// + /// Gets the time elapsed since the last message was received on this connection, + /// or if no message has been received yet. + /// + public TimeSpan? ElapsedSinceLastMessageReceived + { + get + { + if (!_lastMessageReceivedTimestamp.IsRunning) return null; + return _lastMessageReceivedTimestamp.Elapsed; + } + } + public static void ConfigureBuilder(ConnectionBuilder builder) => builder.Run(OnConnectedDelegate); /// @@ -275,6 +289,7 @@ private async Task ProcessIncoming() Exception? error = default; var serializer = this.shared.ServiceProvider.GetRequiredService(); + var prevBufferLength = 0L; try { var input = this._transport!.Input; @@ -284,8 +299,15 @@ private async Task ProcessIncoming() var readResult = await input.ReadAsync(); var buffer = readResult.Buffer; + if (buffer.Length > prevBufferLength) + { + prevBufferLength = buffer.Length; + _lastMessageReceivedTimestamp.Restart(); + } + if (buffer.Length >= requiredBytes) { + prevBufferLength = 0; do { Message? message = default; @@ -307,7 +329,7 @@ private async Task ProcessIncoming() if (!HandleReceiveMessageFailure(message, exception)) { throw; - } + } } } while (requiredBytes == 0); } diff --git a/src/Orleans.Core/Networking/ConnectionManager.cs b/src/Orleans.Core/Networking/ConnectionManager.cs index f6f68902693..553666bcd07 100644 --- a/src/Orleans.Core/Networking/ConnectionManager.cs +++ b/src/Orleans.Core/Networking/ConnectionManager.cs @@ -75,6 +75,32 @@ public bool TryGetConnection(SiloAddress endpoint, [NotNullWhen(true)] out Conne return false; } + /// + /// Gets the minimum elapsed time since any connection to the specified silo last received a message, + /// or if no connections exist or no messages have been received. + /// + /// The silo address to check. + /// The elapsed time since the most recently received message across all connections, or . + public TimeSpan? GetElapsedSinceLastMessageReceived(SiloAddress endpoint) + { + if (!this.connections.TryGetValue(endpoint, out var entry)) + { + return null; + } + + TimeSpan? minElapsed = null; + foreach (var connection in entry.Connections) + { + if (connection.ElapsedSinceLastMessageReceived is { } elapsed + && (minElapsed is null || elapsed < minElapsed)) + { + minElapsed = elapsed; + } + } + + return minElapsed; + } + private async Task GetConnectionAsync(SiloAddress endpoint) { await Task.Yield(); diff --git a/src/Orleans.Core/Timers/CoarseStopwatch.cs b/src/Orleans.Core/Timers/CoarseStopwatch.cs index 0242873fc40..b43e52644cb 100644 --- a/src/Orleans.Core/Timers/CoarseStopwatch.cs +++ b/src/Orleans.Core/Timers/CoarseStopwatch.cs @@ -45,7 +45,7 @@ private CoarseStopwatch(long timestamp) /// /// Returns the elapsed time. /// - public TimeSpan Elapsed => TimeSpan.FromMilliseconds(ElapsedMilliseconds); + public readonly TimeSpan Elapsed => TimeSpan.FromMilliseconds(ElapsedMilliseconds); /// /// Returns a value indicating whether this instance has the default value. diff --git a/src/Orleans.Runtime/MembershipService/ClusterHealthMonitor.cs b/src/Orleans.Runtime/MembershipService/ClusterHealthMonitor.cs index aa6036d7a8b..126545d47cb 100644 --- a/src/Orleans.Runtime/MembershipService/ClusterHealthMonitor.cs +++ b/src/Orleans.Runtime/MembershipService/ClusterHealthMonitor.cs @@ -12,6 +12,7 @@ using Microsoft.Extensions.Options; using Orleans.Configuration; using Orleans.Internal; +using Orleans.Runtime.Messaging; using static Orleans.Runtime.MembershipService.SiloHealthMonitor; namespace Orleans.Runtime.MembershipService @@ -26,6 +27,7 @@ internal partial class ClusterHealthMonitor : IClusterHealthMonitor, ClusterHeal private readonly ILocalSiloDetails localSiloDetails; private readonly IServiceProvider serviceProvider; private readonly IMembershipManager membershipManager; + private readonly ConnectionManager connectionManager; private readonly ILogger log; private readonly IFatalErrorHandler fatalErrorHandler; private readonly IOptionsMonitor clusterMembershipOptions; @@ -51,11 +53,13 @@ public ClusterHealthMonitor( ILogger log, IOptionsMonitor clusterMembershipOptions, IFatalErrorHandler fatalErrorHandler, - IServiceProvider serviceProvider) + IServiceProvider serviceProvider, + ConnectionManager connectionManager) { this.localSiloDetails = localSiloDetails; this.serviceProvider = serviceProvider; this.membershipManager = membershipManager; + this.connectionManager = connectionManager; this.log = log; this.fatalErrorHandler = fatalErrorHandler; this.clusterMembershipOptions = clusterMembershipOptions; @@ -324,15 +328,49 @@ private async Task OnProbeResultInternal(SiloHealthMonitor monitor, ProbeResult { if (probeResult.Status == ProbeResultStatus.Failed && probeResult.FailedProbeCount >= this.clusterMembershipOptions.CurrentValue.NumMissedProbesLimit) { + if (IsConnectionActiveWithinMonitoringWindow(monitor.TargetSiloAddress)) + { + return; + } + await this.membershipManager.TrySuspectSilo(monitor.TargetSiloAddress, null, this.shutdownCancellation.Token).ConfigureAwait(false); } } else if (probeResult.Status == ProbeResultStatus.Failed) { + if (IsConnectionActiveWithinMonitoringWindow(monitor.TargetSiloAddress)) + { + return; + } + await this.membershipManager.TrySuspectSilo(monitor.TargetSiloAddress, probeResult.Intermediary, this.shutdownCancellation.Token).ConfigureAwait(false); } } + /// + /// Checks whether a connection to the specified silo has received a message within the monitoring window + /// ( × ). + /// If so, the silo is demonstrably alive and the vote should be suppressed. + /// + private bool IsConnectionActiveWithinMonitoringWindow(SiloAddress targetSilo) + { + var options = this.clusterMembershipOptions.CurrentValue; + if (!options.EnableConnectionLivenessCheck) + { + return false; + } + + var monitoringWindow = options.ProbeTimeout.Multiply(options.NumMissedProbesLimit); + + if (this.connectionManager.GetElapsedSinceLastMessageReceived(targetSilo) is { } elapsed && elapsed <= monitoringWindow) + { + LogInformationSuppressingVoteDueToActiveConnection(log, targetSilo, elapsed, monitoringWindow); + return true; + } + + return false; + } + bool IHealthCheckable.CheckHealth(DateTime lastCheckTime, [MaybeNullWhen(true)] out string reason) { var ok = true; @@ -464,5 +502,11 @@ private readonly struct ProbedSilosLogRecord(IEnumerable probedSilo Message = "Error disposing monitor for {SiloAddress}." )] private static partial void LogErrorDisposingMonitorForSilo(ILogger logger, Exception exception, SiloAddress siloAddress); + + [LoggerMessage( + Level = LogLevel.Information, + Message = "Suppressing vote to suspect silo {SiloAddress}: connection received a message {Elapsed} ago, within the {MonitoringWindow} monitoring window. The silo is demonstrably alive." + )] + private static partial void LogInformationSuppressingVoteDueToActiveConnection(ILogger logger, SiloAddress siloAddress, TimeSpan elapsed, TimeSpan monitoringWindow); } } diff --git a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs index fb29097c4ae..9356cb2a707 100644 --- a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs +++ b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs @@ -1,11 +1,16 @@ using System.Collections.Concurrent; +using Microsoft.AspNetCore.Connections; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using NonSilo.Tests.Utilities; using NSubstitute; using Orleans.Configuration; using Orleans.Core.Diagnostics; +using Orleans.Messaging; +using Orleans.Placement.Repartitioning; +using Orleans.Runtime; using Orleans.Runtime.MembershipService; +using Orleans.Runtime.Messaging; using Orleans.TestingHost.Diagnostics; using TestExtensions; using Xunit; @@ -34,6 +39,7 @@ public class ClusterHealthMonitorTests private readonly ILocalSiloHealthMonitor localSiloHealthMonitor; private readonly InMemoryMembershipTable membershipTable; private readonly IRemoteSiloProber prober; + private readonly ConnectionManager connectionManager; public ClusterHealthMonitorTests(ITestOutputHelper output) { @@ -70,6 +76,10 @@ public ClusterHealthMonitorTests(ITestOutputHelper output) this.prober = Substitute.For(); this.membershipTable = new InMemoryMembershipTable(new TableVersion(1, "1")); + this.connectionManager = new ConnectionManager( + Options.Create(new ConnectionOptions()), + null, + new NetworkingTrace(this.loggerFactory)); } /// @@ -204,6 +214,155 @@ public async Task ClusterHealthMonitor_SilosWithStaleCreatedOrJoiningState_Disab await ClusterHealthMonitor_StaleJoinOrCreatedSilos_Runner(evictWhenMaxJoinAttemptTimeExceeded: false, numVotesForDeathDeclaration: 3); } + /// + /// Tests that when an active connection has recently received messages from the target silo, + /// the vote to suspect/kill is suppressed even though probes are failing. + /// + [Fact] + public async Task ClusterHealthMonitor_ConnectionCanary_SuppressesVoteWhenConnectionActive() + { + var now = DateTimeOffset.UtcNow; + var clusterMembershipOptions = new ClusterMembershipOptions + { + EnableIndirectProbes = false, + NumProbedSilos = 1, + NumVotesForDeathDeclaration = 1, + }; + + var canaryConnectionManager = new ConnectionManager( + Options.Create(new ConnectionOptions()), + null, + new NetworkingTrace(this.loggerFactory)); + + var testRig = CreateClusterHealthMonitorTestRig(clusterMembershipOptions, canaryConnectionManager); + + // Set up probes to always fail. + var probeCalls = new ConcurrentQueue(); + this.prober.Probe(default, default).ReturnsForAnyArgs(info => + { + probeCalls.Enqueue(info.ArgAt(0)); + return Task.FromException(new Exception("probe failed")); + }); + + await this.lifecycle.OnStart(); + + var targetSilo = Silo("127.0.0.200:100@100"); + await this.membershipTable.InsertRow(Entry(targetSilo, SiloStatus.Active, now), this.membershipTable.Version.Next()); + await testRig.Manager.Refresh(); + await testRig.Manager.UpdateStatus(SiloStatus.Active); + await testRig.Manager.Refresh(); + + await Until(() => testRig.TestAccessor.MonitoredSilos.Count > 0); + + // Register a test connection and simulate recent message activity. + var testConnection = CreateTestConnection(this.loggerFactory); + canaryConnectionManager.OnConnected(targetSilo, testConnection); + SimulateMessageReceived(testConnection); + + // Drive enough probe failures to normally trigger a vote. + for (var i = 0; i < clusterMembershipOptions.NumMissedProbesLimit + 1; i++) + { + if (this.timerCalls.TryDequeue(out var timer)) + { + timer.Completion.TrySetResult(true); + } + + // Keep re-stamping the canary so it stays fresh. + SimulateMessageReceived(testConnection); + await Task.Delay(50); + } + + // Let any pending async work complete. + testRig.Manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(5)); + + // The silo should NOT be dead because the canary detected active connection traffic. + var table = await this.membershipTable.ReadAll(); + var entry = table.Members.SingleOrDefault(m => m.Item1.SiloAddress.Equals(targetSilo)); + Assert.NotNull(entry); + Assert.NotEqual(SiloStatus.Dead, entry.Item1.Status); + + await StopLifecycle(); + } + + /// + /// Tests that when no connections exist to a target silo, the canary does not interfere + /// and the silo is declared dead normally after probe failures. + /// + [Fact] + public async Task ClusterHealthMonitor_ConnectionCanary_AllowsVoteWhenNoConnection() + { + // With no connections in the connection manager, canary returns null -> vote proceeds. + await ClusterHealthMonitor_BasicScenario_Runner(enableIndirectProbes: false, numVotesForDeathDeclaration: 1); + } + + /// + /// Tests that when is disabled, + /// votes proceed normally even when an active connection exists. + /// + [Fact] + public async Task ClusterHealthMonitor_ConnectionCanary_DisabledByOption() + { + var now = DateTimeOffset.UtcNow; + var clusterMembershipOptions = new ClusterMembershipOptions + { + EnableIndirectProbes = false, + NumProbedSilos = 1, + NumVotesForDeathDeclaration = 1, + EnableConnectionLivenessCheck = false, + }; + + var canaryConnectionManager = new ConnectionManager( + Options.Create(new ConnectionOptions()), + null, + new NetworkingTrace(this.loggerFactory)); + + var testRig = CreateClusterHealthMonitorTestRig(clusterMembershipOptions, canaryConnectionManager); + + var probeCalls = new ConcurrentQueue(); + this.prober.Probe(default, default).ReturnsForAnyArgs(info => + { + probeCalls.Enqueue(info.ArgAt(0)); + return Task.FromException(new Exception("probe failed")); + }); + + await this.lifecycle.OnStart(); + + var targetSilo = Silo("127.0.0.200:100@100"); + await this.membershipTable.InsertRow(Entry(targetSilo, SiloStatus.Active, now), this.membershipTable.Version.Next()); + await testRig.Manager.Refresh(); + await testRig.Manager.UpdateStatus(SiloStatus.Active); + await testRig.Manager.Refresh(); + + await Until(() => testRig.TestAccessor.MonitoredSilos.Count > 0); + + // Register a test connection and simulate recent message activity. + var testConnection = CreateTestConnection(this.loggerFactory); + canaryConnectionManager.OnConnected(targetSilo, testConnection); + SimulateMessageReceived(testConnection); + + // Drive enough probe failures to trigger a vote. + for (var i = 0; i < clusterMembershipOptions.NumMissedProbesLimit + 1; i++) + { + if (this.timerCalls.TryDequeue(out var timer)) + { + timer.Completion.TrySetResult(true); + } + + SimulateMessageReceived(testConnection); + await Task.Delay(50); + } + + testRig.Manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(5)); + + // Despite an active connection, the silo SHOULD be dead because the option is disabled. + var table = await this.membershipTable.ReadAll(); + var entry = table.Members.SingleOrDefault(m => m.Item1.SiloAddress.Equals(targetSilo)); + Assert.NotNull(entry); + Assert.Equal(SiloStatus.Dead, entry.Item1.Status); + + await StopLifecycle(); + } + private async Task ClusterHealthMonitor_BasicScenario_Runner(bool enableIndirectProbes, int? numVotesForDeathDeclaration = default, bool otherSilosAreStale = false) { var now = DateTimeOffset.UtcNow; @@ -656,6 +815,11 @@ private class ClusterHealthMonitorTestRig( } private ClusterHealthMonitorTestRig CreateClusterHealthMonitorTestRig(ClusterMembershipOptions clusterMembershipOptions) + { + return CreateClusterHealthMonitorTestRig(clusterMembershipOptions, this.connectionManager); + } + + private ClusterHealthMonitorTestRig CreateClusterHealthMonitorTestRig(ClusterMembershipOptions clusterMembershipOptions, ConnectionManager connManager) { var manager = new MembershipTableManager( localSiloDetails: this.localSiloDetails, @@ -679,7 +843,8 @@ private ClusterHealthMonitorTestRig CreateClusterHealthMonitorTestRig(ClusterMem this.loggerFactory.CreateLogger(), optionsMonitor, this.fatalErrorHandler, - null!); + null!, + connManager); ((ILifecycleParticipant)monitor).Participate(this.lifecycle); @@ -701,5 +866,47 @@ private ClusterHealthMonitorTestRig CreateClusterHealthMonitorTestRig(ClusterMem optionsMonitor: optionsMonitor, testAccessor: testAccessor); } + + /// + /// Simulates message activity on a connection by setting the last-message-received timestamp + /// via reflection (the field is updated via Volatile.Write in production code on the I/O path). + /// + private static void SimulateMessageReceived(Connection connection) + { + var field = typeof(Connection).GetField("_lastMessageReceivedTimestamp", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)!; + field.SetValue(connection, CoarseStopwatch.StartNew()); + } + + /// + /// Creates a minimal test suitable for registering with . + /// + private Connection CreateTestConnection(ILoggerFactory loggerFactory) + { + var features = new Microsoft.AspNetCore.Http.Features.FeatureCollection(); + var context = Substitute.For(); + context.Features.Returns(features); + ConnectionDelegate middleware = _ => Task.CompletedTask; + var messagingTrace = new MessagingTrace(loggerFactory); + var shared = new ConnectionCommon( + Substitute.For(), + null, + messagingTrace, + new NetworkingTrace(loggerFactory), + new NoOpMessageStatisticsSink()); + return new TestConnection(context, middleware, shared); + } + + private sealed class TestConnection(ConnectionContext context, ConnectionDelegate middleware, ConnectionCommon shared) + : Connection(context, middleware, shared) + { + protected override ConnectionDirection ConnectionDirection => ConnectionDirection.SiloToSilo; + protected override IMessageCenter MessageCenter => null; + protected override bool PrepareMessageForSend(Message msg) => true; + protected override void OnReceivedMessage(Message msg) { } + protected override void RecordMessageReceive(Message msg, int numTotalBytes, int headerBytes) { } + protected override void RecordMessageSend(Message msg, int numTotalBytes, int headerBytes) { } + protected override void OnSendMessageFailure(Message message, string error) { } + protected override void RetryMessage(Message msg, Exception ex = null) { } + } } } diff --git a/test/Orleans.Core.Tests/Membership/MembershipAgentTests.cs b/test/Orleans.Core.Tests/Membership/MembershipAgentTests.cs index d0f302648d1..f7bede47e70 100644 --- a/test/Orleans.Core.Tests/Membership/MembershipAgentTests.cs +++ b/test/Orleans.Core.Tests/Membership/MembershipAgentTests.cs @@ -7,6 +7,7 @@ using Orleans.Configuration; using Orleans.Runtime; using Orleans.Runtime.MembershipService; +using Orleans.Runtime.Messaging; using TestExtensions; using Xunit; using Xunit.Abstractions; @@ -93,7 +94,11 @@ public MembershipAgentTests(ITestOutputHelper output) this.loggerFactory.CreateLogger(), optionsMonitor, this.fatalErrorHandler, - null!); + null!, + new ConnectionManager( + Options.Create(new ConnectionOptions()), + null!, + this.loggerFactory.CreateLogger())); ((ILifecycleParticipant)this.clusterHealthMonitor).Participate(this.lifecycle); this.remoteSiloProber = Substitute.For(); From ebd7779d413b3f248ebf4fcf2d01963e455b1893 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Thu, 7 May 2026 13:15:56 -0700 Subject: [PATCH 2/7] Avoid reflection in connection liveness tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/Orleans.Core/Networking/Connection.cs | 4 +++- .../Membership/ClusterHealthMonitorTests.cs | 21 ++++++------------- 2 files changed, 9 insertions(+), 16 deletions(-) diff --git a/src/Orleans.Core/Networking/Connection.cs b/src/Orleans.Core/Networking/Connection.cs index 10109af7e26..3728d36ce8b 100644 --- a/src/Orleans.Core/Networking/Connection.cs +++ b/src/Orleans.Core/Networking/Connection.cs @@ -88,6 +88,8 @@ public TimeSpan? ElapsedSinceLastMessageReceived } } + protected void MarkMessageReceived() => _lastMessageReceivedTimestamp.Restart(); + public static void ConfigureBuilder(ConnectionBuilder builder) => builder.Run(OnConnectedDelegate); /// @@ -302,7 +304,7 @@ private async Task ProcessIncoming() if (buffer.Length > prevBufferLength) { prevBufferLength = buffer.Length; - _lastMessageReceivedTimestamp.Restart(); + MarkMessageReceived(); } if (buffer.Length >= requiredBytes) diff --git a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs index 9356cb2a707..16587c75b01 100644 --- a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs +++ b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs @@ -257,7 +257,7 @@ public async Task ClusterHealthMonitor_ConnectionCanary_SuppressesVoteWhenConnec // Register a test connection and simulate recent message activity. var testConnection = CreateTestConnection(this.loggerFactory); canaryConnectionManager.OnConnected(targetSilo, testConnection); - SimulateMessageReceived(testConnection); + testConnection.SimulateMessageReceived(); // Drive enough probe failures to normally trigger a vote. for (var i = 0; i < clusterMembershipOptions.NumMissedProbesLimit + 1; i++) @@ -268,7 +268,7 @@ public async Task ClusterHealthMonitor_ConnectionCanary_SuppressesVoteWhenConnec } // Keep re-stamping the canary so it stays fresh. - SimulateMessageReceived(testConnection); + testConnection.SimulateMessageReceived(); await Task.Delay(50); } @@ -338,7 +338,7 @@ public async Task ClusterHealthMonitor_ConnectionCanary_DisabledByOption() // Register a test connection and simulate recent message activity. var testConnection = CreateTestConnection(this.loggerFactory); canaryConnectionManager.OnConnected(targetSilo, testConnection); - SimulateMessageReceived(testConnection); + testConnection.SimulateMessageReceived(); // Drive enough probe failures to trigger a vote. for (var i = 0; i < clusterMembershipOptions.NumMissedProbesLimit + 1; i++) @@ -348,7 +348,7 @@ public async Task ClusterHealthMonitor_ConnectionCanary_DisabledByOption() timer.Completion.TrySetResult(true); } - SimulateMessageReceived(testConnection); + testConnection.SimulateMessageReceived(); await Task.Delay(50); } @@ -867,20 +867,10 @@ private ClusterHealthMonitorTestRig CreateClusterHealthMonitorTestRig(ClusterMem testAccessor: testAccessor); } - /// - /// Simulates message activity on a connection by setting the last-message-received timestamp - /// via reflection (the field is updated via Volatile.Write in production code on the I/O path). - /// - private static void SimulateMessageReceived(Connection connection) - { - var field = typeof(Connection).GetField("_lastMessageReceivedTimestamp", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance)!; - field.SetValue(connection, CoarseStopwatch.StartNew()); - } - /// /// Creates a minimal test suitable for registering with . /// - private Connection CreateTestConnection(ILoggerFactory loggerFactory) + private TestConnection CreateTestConnection(ILoggerFactory loggerFactory) { var features = new Microsoft.AspNetCore.Http.Features.FeatureCollection(); var context = Substitute.For(); @@ -907,6 +897,7 @@ protected override void RecordMessageReceive(Message msg, int numTotalBytes, int protected override void RecordMessageSend(Message msg, int numTotalBytes, int headerBytes) { } protected override void OnSendMessageFailure(Message message, string error) { } protected override void RetryMessage(Message msg, Exception ex = null) { } + public void SimulateMessageReceived() => MarkMessageReceived(); } } } From b04be43d955fbe6d42bb427f327f6ea48e6cf9e5 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Thu, 7 May 2026 13:36:57 -0700 Subject: [PATCH 3/7] Fix connection liveness tests for current main Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Membership/ClusterHealthMonitorTests.cs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs index 16587c75b01..2155e95c138 100644 --- a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs +++ b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs @@ -79,7 +79,7 @@ public ClusterHealthMonitorTests(ITestOutputHelper output) this.connectionManager = new ConnectionManager( Options.Create(new ConnectionOptions()), null, - new NetworkingTrace(this.loggerFactory)); + this.loggerFactory.CreateLogger()); } /// @@ -232,7 +232,7 @@ public async Task ClusterHealthMonitor_ConnectionCanary_SuppressesVoteWhenConnec var canaryConnectionManager = new ConnectionManager( Options.Create(new ConnectionOptions()), null, - new NetworkingTrace(this.loggerFactory)); + this.loggerFactory.CreateLogger()); var testRig = CreateClusterHealthMonitorTestRig(clusterMembershipOptions, canaryConnectionManager); @@ -314,7 +314,7 @@ public async Task ClusterHealthMonitor_ConnectionCanary_DisabledByOption() var canaryConnectionManager = new ConnectionManager( Options.Create(new ConnectionOptions()), null, - new NetworkingTrace(this.loggerFactory)); + this.loggerFactory.CreateLogger()); var testRig = CreateClusterHealthMonitorTestRig(clusterMembershipOptions, canaryConnectionManager); @@ -881,7 +881,7 @@ private TestConnection CreateTestConnection(ILoggerFactory loggerFactory) Substitute.For(), null, messagingTrace, - new NetworkingTrace(loggerFactory), + loggerFactory.CreateLogger(), new NoOpMessageStatisticsSink()); return new TestConnection(context, middleware, shared); } From 8c1ff77466264d347abdda496501e2d4d800aadb Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Mon, 10 Aug 2026 18:47:37 -0700 Subject: [PATCH 4/7] Harden connection liveness tracking Use atomic timestamp access across the receive and membership threads and include the new option in the checked-in public API surface. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3182b65a-654c-4cb7-b172-3dc16a1df7b2 --- src/Orleans.Core/Networking/Connection.cs | 9 +++++---- src/api/Orleans.Core/Orleans.Core.cs | 2 ++ 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/src/Orleans.Core/Networking/Connection.cs b/src/Orleans.Core/Networking/Connection.cs index 3728d36ce8b..b9b38296057 100644 --- a/src/Orleans.Core/Networking/Connection.cs +++ b/src/Orleans.Core/Networking/Connection.cs @@ -39,7 +39,7 @@ internal abstract partial class Connection private Task? _processIncomingTask; private Task? _processOutgoingTask; private Task? _closeTask; - private CoarseStopwatch _lastMessageReceivedTimestamp; + private long _lastMessageReceivedTimestamp; protected Connection( ConnectionContext connection, @@ -83,12 +83,13 @@ public TimeSpan? ElapsedSinceLastMessageReceived { get { - if (!_lastMessageReceivedTimestamp.IsRunning) return null; - return _lastMessageReceivedTimestamp.Elapsed; + var timestamp = Volatile.Read(ref _lastMessageReceivedTimestamp); + if (timestamp == 0) return null; + return TimeSpan.FromMilliseconds(CoarseStopwatch.GetTimestamp() - timestamp); } } - protected void MarkMessageReceived() => _lastMessageReceivedTimestamp.Restart(); + protected void MarkMessageReceived() => Volatile.Write(ref _lastMessageReceivedTimestamp, CoarseStopwatch.GetTimestamp()); public static void ConfigureBuilder(ConnectionBuilder builder) => builder.Run(OnConnectedDelegate); diff --git a/src/api/Orleans.Core/Orleans.Core.cs b/src/api/Orleans.Core/Orleans.Core.cs index 998d47a1dfa..16150922ebc 100644 --- a/src/api/Orleans.Core/Orleans.Core.cs +++ b/src/api/Orleans.Core/Orleans.Core.cs @@ -443,6 +443,8 @@ public partial class ClusterMembershipOptions public System.TimeSpan DefunctSiloExpiration { get { throw null; } set { } } + public bool EnableConnectionLivenessCheck { get { throw null; } set { } } + public bool EnableIndirectProbes { get { throw null; } set { } } public bool EvictWhenMaxJoinAttemptTimeExceeded { get { throw null; } set { } } From 0b9ea76de471e1196e27419e30cd285a724b7beb Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Mon, 10 Aug 2026 18:58:01 -0700 Subject: [PATCH 5/7] Fix membership option documentation Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3182b65a-654c-4cb7-b172-3dc16a1df7b2 --- .../Configuration/Options/ClusterMembershipOptions.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Orleans.Core/Configuration/Options/ClusterMembershipOptions.cs b/src/Orleans.Core/Configuration/Options/ClusterMembershipOptions.cs index 5ab4d8bbc87..32dae280035 100644 --- a/src/Orleans.Core/Configuration/Options/ClusterMembershipOptions.cs +++ b/src/Orleans.Core/Configuration/Options/ClusterMembershipOptions.cs @@ -149,7 +149,7 @@ public class ClusterMembershipOptions /// /// Connection liveness checks are enabled by default. public bool EnableConnectionLivenessCheck { get; set; } = true; - /// + /// /// Gets or sets a value indicating whether to enable membership eviction of silos when they remain in the Joining or Created state for longer than . /// From 1aed8a442eb6292c3fd158d2f3d78cb96ae3a9f4 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Mon, 10 Aug 2026 19:03:09 -0700 Subject: [PATCH 6/7] Fix test connection nullability Match the nullable exception signature introduced on current main. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3182b65a-654c-4cb7-b172-3dc16a1df7b2 --- test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs index 2155e95c138..b350e47fa03 100644 --- a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs +++ b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs @@ -896,7 +896,7 @@ protected override void OnReceivedMessage(Message msg) { } protected override void RecordMessageReceive(Message msg, int numTotalBytes, int headerBytes) { } protected override void RecordMessageSend(Message msg, int numTotalBytes, int headerBytes) { } protected override void OnSendMessageFailure(Message message, string error) { } - protected override void RetryMessage(Message msg, Exception ex = null) { } + protected override void RetryMessage(Message msg, Exception? ex = null) { } public void SimulateMessageReceived() => MarkMessageReceived(); } } From 9b4f398ba09cfed90684d54d68ff0e98c4499338 Mon Sep 17 00:00:00 2001 From: Reuben Bond Date: Mon, 10 Aug 2026 19:19:55 -0700 Subject: [PATCH 7/7] Update liveness tests for current runtime APIs Use current metrics dependencies and membership polling after rebasing onto main. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3182b65a-654c-4cb7-b172-3dc16a1df7b2 --- .../Membership/ClusterHealthMonitorTests.cs | 49 ++++++++++++++----- 1 file changed, 37 insertions(+), 12 deletions(-) diff --git a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs index b350e47fa03..924e5896a38 100644 --- a/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs +++ b/test/Orleans.Core.Tests/Membership/ClusterHealthMonitorTests.cs @@ -1,5 +1,6 @@ using System.Collections.Concurrent; using Microsoft.AspNetCore.Connections; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using NonSilo.Tests.Utilities; @@ -78,7 +79,7 @@ public ClusterHealthMonitorTests(ITestOutputHelper output) this.membershipTable = new InMemoryMembershipTable(new TableVersion(1, "1")); this.connectionManager = new ConnectionManager( Options.Create(new ConnectionOptions()), - null, + null!, this.loggerFactory.CreateLogger()); } @@ -231,14 +232,14 @@ public async Task ClusterHealthMonitor_ConnectionCanary_SuppressesVoteWhenConnec var canaryConnectionManager = new ConnectionManager( Options.Create(new ConnectionOptions()), - null, + null!, this.loggerFactory.CreateLogger()); var testRig = CreateClusterHealthMonitorTestRig(clusterMembershipOptions, canaryConnectionManager); // Set up probes to always fail. var probeCalls = new ConcurrentQueue(); - this.prober.Probe(default, default).ReturnsForAnyArgs(info => + this.prober.Probe(default!, default).ReturnsForAnyArgs(info => { probeCalls.Enqueue(info.ArgAt(0)); return Task.FromException(new Exception("probe failed")); @@ -272,8 +273,8 @@ public async Task ClusterHealthMonitor_ConnectionCanary_SuppressesVoteWhenConnec await Task.Delay(50); } - // Let any pending async work complete. - testRig.Manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(5)); + await Until(() => probeCalls.Count >= clusterMembershipOptions.NumMissedProbesLimit); + await Task.Delay(100); // The silo should NOT be dead because the canary detected active connection traffic. var table = await this.membershipTable.ReadAll(); @@ -313,13 +314,13 @@ public async Task ClusterHealthMonitor_ConnectionCanary_DisabledByOption() var canaryConnectionManager = new ConnectionManager( Options.Create(new ConnectionOptions()), - null, + null!, this.loggerFactory.CreateLogger()); var testRig = CreateClusterHealthMonitorTestRig(clusterMembershipOptions, canaryConnectionManager); var probeCalls = new ConcurrentQueue(); - this.prober.Probe(default, default).ReturnsForAnyArgs(info => + this.prober.Probe(default!, default).ReturnsForAnyArgs(info => { probeCalls.Enqueue(info.ArgAt(0)); return Task.FromException(new Exception("probe failed")); @@ -352,7 +353,11 @@ public async Task ClusterHealthMonitor_ConnectionCanary_DisabledByOption() await Task.Delay(50); } - testRig.Manager.TestingSuspectOrKillIdle.WaitOne(TimeSpan.FromSeconds(5)); + await Until(async () => + { + var snapshot = await this.membershipTable.ReadAll(); + return snapshot.Members.Any(m => m.Item1.SiloAddress.Equals(targetSilo) && m.Item1.Status == SiloStatus.Dead); + }); // Despite an active connection, the silo SHOULD be dead because the option is disabled. var table = await this.membershipTable.ReadAll(); @@ -781,6 +786,13 @@ private static async Task Until(Func condition) Assert.True(maxTimeout > 0); } + private static async Task Until(Func> condition) + { + var maxTimeout = 40_000; + while (!await condition() && (maxTimeout -= 10) > 0) await Task.Delay(10); + Assert.True(maxTimeout > 0); + } + private static async Task WaitForMembershipSnapshot(DiagnosticEventCollector membershipEvents, Func condition) { var diagnosticEvent = await membershipEvents.WaitForEventAsync( @@ -876,11 +888,24 @@ private TestConnection CreateTestConnection(ILoggerFactory loggerFactory) var context = Substitute.For(); context.Features.Returns(features); ConnectionDelegate middleware = _ => Task.CompletedTask; - var messagingTrace = new MessagingTrace(loggerFactory); + var services = new ServiceCollection(); + services.AddMetrics(); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + var serviceProvider = services.BuildServiceProvider(); + var orleansInstruments = serviceProvider.GetRequiredService(); + var messagingInstruments = serviceProvider.GetRequiredService(); + var messagingTrace = new MessagingTrace( + loggerFactory, + messagingInstruments, + serviceProvider.GetRequiredService()); var shared = new ConnectionCommon( - Substitute.For(), - null, + serviceProvider, + null!, messagingTrace, + orleansInstruments, + messagingInstruments, loggerFactory.CreateLogger(), new NoOpMessageStatisticsSink()); return new TestConnection(context, middleware, shared); @@ -890,7 +915,7 @@ private sealed class TestConnection(ConnectionContext context, ConnectionDelegat : Connection(context, middleware, shared) { protected override ConnectionDirection ConnectionDirection => ConnectionDirection.SiloToSilo; - protected override IMessageCenter MessageCenter => null; + protected override IMessageCenter MessageCenter => null!; protected override bool PrepareMessageForSend(Message msg) => true; protected override void OnReceivedMessage(Message msg) { } protected override void RecordMessageReceive(Message msg, int numTotalBytes, int headerBytes) { }