From b90f0ad2993e72459867d806cdf19d1b4b490a9d Mon Sep 17 00:00:00 2001 From: Anne Erdtsieck Date: Wed, 5 Aug 2026 15:04:01 +0200 Subject: [PATCH] Ask for the active node numbers once per node, not once per database MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DurabilityAgent's recovery timer needs the active node numbers to spot messages orphaned by a departed node. The only call that yields them, LoadAllNodesAsync, also selects the entire assignment table so it can populate ActiveAgents — which this caller never reads — and there is one durability agent per message database. On a sharded deployment that turns a per-node fact into a per-database query. At 512 databases on the default five-second ScheduledJobPollingTime and ~10,000 agents, measured on the main store: 76 calls and 772,000 rows a second, with Client:ClientWrite at 164 of 170 average active sessions while Lock:transactionid sat at 0.04 — the server's time went into writing those result sets. Each call also held a pooled connection, so the heartbeat writes timed out, the leader declared healthy nodes stale, and reassigning their agents churned the very table being read. ActiveNodeNumberCache fetches it at most once per polling interval for the whole node. The caller already tolerates data one interval old, since that is how often it looks, so behaviour is unchanged and the query count drops by the database count. Same shape as PersistenceMetricsSweeper for the metrics polling in GH-3375. Failures are deliberately not cached: the durability agent keeps its own try/catch and decides what a failed lookup means. --- .../Wolverine.RDBMS/DurabilityAgent.cs | 7 +- .../active_node_number_cache_tests.cs | 97 +++++++++++++++++++ .../Persistence/ActiveNodeNumberCache.cs | 95 ++++++++++++++++++ 3 files changed, 197 insertions(+), 2 deletions(-) create mode 100644 src/Testing/CoreTests/Persistence/active_node_number_cache_tests.cs create mode 100644 src/Wolverine/Persistence/ActiveNodeNumberCache.cs diff --git a/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs b/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs index 2bfb042e2..80425085d 100644 --- a/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs +++ b/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs @@ -99,8 +99,11 @@ public Task StartAsync(CancellationToken cancellationToken) { try { - var nodes = await _runtime.Storage.Nodes.LoadAllNodesAsync(_runtime.Cancellation); - activeNodeNumbers = nodes.Select(n => n.AssignedNodeNumber).ToList(); + // Node-wide, not per-database: LoadAllNodesAsync also selects the whole assignment + // table to populate ActiveAgents, which this caller never reads, and there is one + // durability agent per message database. See ActiveNodeNumberCache. + activeNodeNumbers = await ActiveNodeNumberCache.For(_runtime) + .FetchAsync(_runtime.Cancellation); } catch (Exception e) { diff --git a/src/Testing/CoreTests/Persistence/active_node_number_cache_tests.cs b/src/Testing/CoreTests/Persistence/active_node_number_cache_tests.cs new file mode 100644 index 000000000..bc8112429 --- /dev/null +++ b/src/Testing/CoreTests/Persistence/active_node_number_cache_tests.cs @@ -0,0 +1,97 @@ +using CoreTests.Runtime; +using JasperFx.Core; +using NSubstitute; +using Shouldly; +using Wolverine.Persistence; +using Wolverine.Runtime.Agents; +using Xunit; + +namespace CoreTests.Persistence; + +// The durability agent's recovery timer needs the active node numbers, but the call that yields +// them also selects the whole assignment table, and there is one durability agent per message +// database. On a sharded fleet that turned a per-node fact into a per-database query. These tests +// pin the collapsing: one fetch serves every database on the node until the polling interval is up. +public class active_node_number_cache_tests +{ + private readonly MockWolverineRuntime theRuntime = new(); + + private ActiveNodeNumberCache theCache => new(theRuntime); + + public active_node_number_cache_tests() + { + theRuntime.DurabilitySettings.ScheduledJobPollingTime = 200.Milliseconds(); + nodesAre(1, 2, 3); + } + + private void nodesAre(params int[] numbers) + { + theRuntime.Storage.Nodes + .LoadAllNodesAsync(Arg.Any()) + .Returns(numbers.Select(x => new WolverineNode { AssignedNodeNumber = x }).ToList()); + } + + private int fetchCount => theRuntime.Storage.Nodes.ReceivedCalls() + .Count(x => x.GetMethodInfo().Name == nameof(INodeAgentPersistence.LoadAllNodesAsync)); + + [Fact] + public async Task returns_the_assigned_node_numbers() + { + var numbers = await theCache.FetchAsync(CancellationToken.None); + numbers.ShouldBe([1, 2, 3]); + } + + [Fact] + public async Task one_fetch_serves_every_database_on_the_node() + { + var cache = theCache; + + // stand in for the node's message databases all polling on the same timer tick + var results = await Task.WhenAll(Enumerable.Range(0, 50) + .Select(_ => cache.FetchAsync(CancellationToken.None).AsTask())); + + fetchCount.ShouldBe(1); + results.ShouldAllBe(x => x.SequenceEqual(new[] { 1, 2, 3 })); + } + + [Fact] + public async Task refetches_once_the_polling_interval_has_passed() + { + var cache = theCache; + await cache.FetchAsync(CancellationToken.None); + + nodesAre(4, 5); + await Task.Delay(theRuntime.DurabilitySettings.ScheduledJobPollingTime + 100.Milliseconds(), + TestContext.Current.CancellationToken); + + var numbers = await cache.FetchAsync(CancellationToken.None); + + numbers.ShouldBe([4, 5]); + fetchCount.ShouldBe(2); + } + + [Fact] + public async Task a_failed_lookup_is_not_cached_and_reaches_the_caller() + { + var cache = theCache; + theRuntime.Storage.Nodes + .LoadAllNodesAsync(Arg.Any()) + .Returns>(_ => throw new TimeoutException("pool exhausted")); + + // the durability agent has its own try/catch and decides what a failed lookup means, so the + // cache must not swallow it — nor remember an empty result as if it were the truth + await Should.ThrowAsync(() => cache.FetchAsync(CancellationToken.None).AsTask()); + + nodesAre(7); + var numbers = await cache.FetchAsync(CancellationToken.None); + numbers.ShouldBe([7]); + } + + [Fact] + public void one_cache_per_runtime() + { + ActiveNodeNumberCache.For(theRuntime).ShouldBeSameAs(ActiveNodeNumberCache.For(theRuntime)); + ActiveNodeNumberCache.For(new MockWolverineRuntime()) + .ShouldNotBeSameAs(ActiveNodeNumberCache.For(theRuntime)); + } +} diff --git a/src/Wolverine/Persistence/ActiveNodeNumberCache.cs b/src/Wolverine/Persistence/ActiveNodeNumberCache.cs new file mode 100644 index 000000000..d0dc4aa02 --- /dev/null +++ b/src/Wolverine/Persistence/ActiveNodeNumberCache.cs @@ -0,0 +1,95 @@ +using System.Runtime.CompilerServices; +using JasperFx.Core; +using Wolverine.Runtime; + +namespace Wolverine.Persistence; + +/// +/// Node-wide cache of the active Wolverine node numbers, shared by every durability agent on the +/// node. +/// +/// +/// The durability agent's recovery timer needs the active node numbers to spot messages +/// orphaned by a departed node, but the only persistence call that yields them — +/// — also selects the entire +/// assignment table so it can populate WolverineNode.ActiveAgents, which this caller never +/// reads. With one durability agent per message database that turns a per-node fact into a +/// per-database query against the main store. +/// +/// On a sharded deployment the cost is quadratic in the fleet: 512 databases on a five-second +/// is ~100 calls a second, each dragging +/// back one row per assignment. Measured on a 512-database, ~10,000-agent cluster that was 76 +/// calls and 772,000 rows a second, and the main store spent its time writing those result sets +/// (Client:ClientWrite was 164 of 170 average active sessions) while each call held a pooled +/// connection — which exhausted the pool and made the heartbeat writes time out, so the leader +/// then declared healthy nodes stale and reassigned their agents, churning the very table being +/// read. +/// +/// Fetching it once per node per polling interval instead of once per database keeps the behaviour +/// (the caller already tolerates data up to one interval old, since that is how often it looks) +/// and drops the query count by the database count. Same shape, and the same reasoning, as +/// for the metrics polling in GH-3375. +/// +public class ActiveNodeNumberCache +{ + private static readonly ConditionalWeakTable _perRuntime = new(); + + public static ActiveNodeNumberCache For(IWolverineRuntime runtime) + { + return _perRuntime.GetValue(runtime, r => new ActiveNodeNumberCache(r)); + } + + private readonly IWolverineRuntime _runtime; + private readonly SemaphoreSlim _gate = new(1, 1); + private IReadOnlyList? _numbers; + private DateTimeOffset _staleAt = DateTimeOffset.MinValue; + + internal ActiveNodeNumberCache(IWolverineRuntime runtime) + { + _runtime = runtime; + } + + /// + /// The active node numbers, fetched at most once per + /// for the whole node. Exceptions are + /// deliberately not swallowed here: the caller decides what a failed lookup means, and a failure + /// leaves the previous value in place rather than caching an empty one. + /// + public async ValueTask> FetchAsync(CancellationToken token) + { + if (fresh(out var cached)) return cached; + + await _gate.WaitAsync(token).ConfigureAwait(false); + try + { + // a second check inside the gate: while this call waited, the database timers that + // queued up behind it have already been served by the winner's fetch + if (fresh(out cached)) return cached; + + var nodes = await _runtime.Storage.Nodes.LoadAllNodesAsync(token).ConfigureAwait(false); + var numbers = nodes.Select(x => x.AssignedNodeNumber).ToList(); + + _numbers = numbers; + _staleAt = DateTimeOffset.UtcNow.Add(_runtime.DurabilitySettings.ScheduledJobPollingTime); + + return numbers; + } + finally + { + _gate.Release(); + } + } + + private bool fresh(out IReadOnlyList numbers) + { + var current = _numbers; + if (current != null && DateTimeOffset.UtcNow < _staleAt) + { + numbers = current; + return true; + } + + numbers = Array.Empty(); + return false; + } +}