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; + } +}