Ask for the active node numbers once per node, not once per database (GH-3846)

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)
                 {

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<CancellationToken>())
+            .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<CancellationToken>())
+            .Returns<IReadOnlyList<WolverineNode>>(_ => 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<TimeoutException>(() => 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));
+    }
+}

src/Wolverine/Persistence/ActiveNodeNumberCache.cs

@@ -0,0 +1,95 @@
+using System.Runtime.CompilerServices;
+using JasperFx.Core;
+using Wolverine.Runtime;
+
+namespace Wolverine.Persistence;
+
+/// <summary>
+/// Node-wide cache of the active Wolverine node numbers, shared by every durability agent on the
+/// node.
+/// </summary>
+/// <remarks>
+/// The durability agent's recovery timer needs the active node <i>numbers</i> to spot messages
+/// orphaned by a departed node, but the only persistence call that yields them —
+/// <see cref="Runtime.Agents.INodeAgentPersistence.LoadAllNodesAsync" /> — also selects the entire
+/// assignment table so it can populate <c>WolverineNode.ActiveAgents</c>, 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
+/// <see cref="DurabilitySettings.ScheduledJobPollingTime" /> 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
+/// (<c>Client:ClientWrite</c> 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
+/// <see cref="PersistenceMetricsSweeper" /> for the metrics polling in GH-3375.
+/// </remarks>
+public class ActiveNodeNumberCache
+{
+    private static readonly ConditionalWeakTable<IWolverineRuntime, ActiveNodeNumberCache> _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<int>? _numbers;
+    private DateTimeOffset _staleAt = DateTimeOffset.MinValue;
+
+    internal ActiveNodeNumberCache(IWolverineRuntime runtime)
+    {
+        _runtime = runtime;
+    }
+
+    /// <summary>
+    /// The active node numbers, fetched at most once per
+    /// <see cref="DurabilitySettings.ScheduledJobPollingTime" /> 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.
+    /// </summary>
+    public async ValueTask<IReadOnlyList<int>> 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<int> numbers)
+    {
+        var current = _numbers;
+        if (current != null && DateTimeOffset.UtcNow < _staleAt)
+        {
+            numbers = current;
+            return true;
+        }
+
+        numbers = Array.Empty<int>();
+        return false;
+    }
+}