Bound the cached node-number release by a high-water mark (GH-3850)

src/Persistence/SqlServerTests/release_orphaned_ancillary_high_water_mark_3850.cs

@@ -0,0 +1,76 @@
+using Microsoft.Data.SqlClient;
+using NSubstitute;
+using Shouldly;
+using Weasel.Core;
+using Wolverine.RDBMS;
+using Wolverine.RDBMS.Durability;
+
+namespace SqlServerTests;
+
+/// <summary>
+/// GH-3850. The active node numbers are cached per node for up to one polling interval (GH-3846),
+/// so the list cannot describe a node that registered after it was taken. Releasing a <i>live</i>
+/// node's rows to <c>owner_id = 0</c> hands its in-flight work to somebody else, so the release is
+/// bounded by the highest node number the cache has ever seen.
+/// </summary>
+public class release_orphaned_ancillary_high_water_mark_3850
+{
+    private static string sqlFor(IReadOnlyList<int> activeNodeNumbers, int highWaterMark)
+    {
+        var database = Substitute.For<IMessageDatabase>();
+        database.SchemaName.Returns("ancillary");
+
+        var operation = new ReleaseOrphanedMessagesForAncillaryOperation(database, activeNodeNumbers,
+            highWaterMark);
+
+        var builder = new DbCommandBuilder(new SqlCommand());
+        operation.ConfigureCommand(builder);
+
+        return builder.Compile().CommandText;
+    }
+
+    [Fact]
+    public void the_release_is_bounded_by_the_high_water_mark()
+    {
+        var sql = sqlFor([1, 2, 3], highWaterMark: 3);
+
+        // Node 4 registered after the cached list was taken. Without this bound the update would
+        // reset its rows -- it is absent from the list purely because the list predates it.
+        sql.ShouldContain("owner_id <= 3");
+    }
+
+    [Fact]
+    public void a_departed_high_numbered_node_is_still_reclaimable()
+    {
+        // Node 3 was the highest and has died, so the active list is [1, 2] -- but the mark stays at
+        // 3 because the cache saw it. Bounding by max(active) instead would put node 3's orphaned
+        // messages permanently out of reach, which is why the mark is monotonic.
+        var sql = sqlFor([1, 2], highWaterMark: 3);
+
+        sql.ShouldContain("owner_id <= 3");
+        sql.ShouldContain("owner_id not in (1, 2)");
+    }
+
+    [Fact]
+    public void the_guard_is_omitted_when_no_mark_is_supplied()
+    {
+        // 0 means "no mark", which restores the un-bounded behaviour rather than releasing nothing.
+        var sql = sqlFor([1, 2, 3], highWaterMark: 0);
+
+        sql.ShouldNotContain("owner_id <=");
+        sql.ShouldContain("owner_id not in (1, 2, 3)");
+    }
+
+    [Fact]
+    public void both_the_inbox_and_the_outbox_are_bounded()
+    {
+        var sql = sqlFor([1, 2], highWaterMark: 7);
+
+        sql.ShouldContain("ancillary.wolverine_incoming");
+        sql.ShouldContain("ancillary.wolverine_outgoing");
+
+        // one bound per statement -- an unbounded outbox release is the same defect, and the outbox
+        // is where a newcomer owns rows first (MessageRoute stamps OwnerId on persist)
+        System.Text.RegularExpressions.Regex.Matches(sql, "owner_id <= 7").Count.ShouldBe(2);
+    }
+}

src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesForAncillaryOperation.cs

@@ -16,11 +16,19 @@ internal class ReleaseOrphanedMessagesForAncillaryOperation : IDatabaseOperation
 {
     private readonly IMessageDatabase _database;
     private readonly IReadOnlyList<int> _activeNodeNumbers;
+    private readonly int _highWaterMark;
 
-    public ReleaseOrphanedMessagesForAncillaryOperation(IMessageDatabase database, IReadOnlyList<int> activeNodeNumbers)
+    /// <param name="highWaterMark">
+    /// The highest node number the active list has ever been known to cover. Owners above it
+    /// registered after that list was taken, so it says nothing about them and they are left alone
+    /// (GH-3850). Pass 0 to disable the guard, which restores the un-bounded behaviour.
+    /// </param>
+    public ReleaseOrphanedMessagesForAncillaryOperation(IMessageDatabase database,
+        IReadOnlyList<int> activeNodeNumbers, int highWaterMark = 0)
     {
         _database = database;
         _activeNodeNumbers = activeNodeNumbers;
+        _highWaterMark = highWaterMark;
     }
 
     public string Description => "Release inbox/outbox messages owned by nodes that no longer exist (ancillary database)";
@@ -34,15 +42,23 @@ public void ConfigureCommand(DbCommandBuilder builder)
         var outgoingTable = new DbObjectName(schemaName, DatabaseConstants.OutgoingTable);
         var nodeList = string.Join(", ", _activeNodeNumbers);
 
+        // GH-3850. The list is cached per node for up to one polling interval, so it cannot describe
+        // a node that registered after it was taken -- and releasing a LIVE node's rows hands its
+        // in-flight work to somebody else. Node numbers are monotonic, so anything above the mark is
+        // newer than the list and is not ours to judge.
+        var ceiling = _highWaterMark > 0
+            ? $" and {DatabaseConstants.OwnerId} <= {_highWaterMark}"
+            : string.Empty;
+
         builder.Append(
-            $"update {incomingTable} set {DatabaseConstants.OwnerId} = 0 where {DatabaseConstants.OwnerId} != 0 and {DatabaseConstants.OwnerId} not in ({nodeList});");
+            $"update {incomingTable} set {DatabaseConstants.OwnerId} = 0 where {DatabaseConstants.OwnerId} != 0 and {DatabaseConstants.OwnerId} not in ({nodeList}){ceiling};");
 
         // Two statements in one operation, so the boundary has to be explicit for the providers
         // that cannot execute several statements from one command
         builder.StartNewCommand();
 
         builder.Append(
-            $"update {outgoingTable} set {DatabaseConstants.OwnerId} = 0 where {DatabaseConstants.OwnerId} != 0 and {DatabaseConstants.OwnerId} not in ({nodeList});");
+            $"update {outgoingTable} set {DatabaseConstants.OwnerId} = 0 where {DatabaseConstants.OwnerId} != 0 and {DatabaseConstants.OwnerId} not in ({nodeList}){ceiling};");
     }
 
     public Task ReadResultsAsync(DbDataReader reader, IList<Exception> exceptions, CancellationToken token)

src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs

@@ -95,23 +95,28 @@ public Task StartAsync(CancellationToken cancellationToken)
         _recoveryTimer = new Timer(async _ =>
         {
             IReadOnlyList<int>? activeNodeNumbers = null;
+            var nodeNumberHighWaterMark = 0;
             if (_settings.Mode != DurabilityMode.Solo && _database.Settings.Role != MessageStoreRole.Main)
             {
                 try
                 {
                     // 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);
+                    var cache = ActiveNodeNumberCache.For(_runtime);
+                    activeNodeNumbers = await cache.FetchAsync(_runtime.Cancellation);
+
+                    // GH-3850: the list is up to one polling interval old, so it cannot speak for a
+                    // node that registered after it was taken. The mark bounds who it may judge.
+                    nodeNumberHighWaterMark = cache.HighWaterMark;
                 }
                 catch (Exception e)
                 {
                     _logger.LogDebug(e, "Failed to load active nodes for orphaned message detection");
                 }
             }
 
-            var operations = buildOperationBatch(activeNodeNumbers);
+            var operations = buildOperationBatch(activeNodeNumbers, nodeNumberHighWaterMark);
 
             var batch = new DatabaseOperationBatch(_database, operations);
             _runningBlock.Post(batch);
@@ -243,7 +248,8 @@ internal void PruneNodeRecords()
         }
     }
 
-    internal IDatabaseOperation[] buildOperationBatch(IReadOnlyList<int>? activeNodeNumbers = null)
+    internal IDatabaseOperation[] buildOperationBatch(IReadOnlyList<int>? activeNodeNumbers = null,
+        int nodeNumberHighWaterMark = 0)
     {
         var incomingTable = new DbObjectName(_database.SchemaName, DatabaseConstants.IncomingTable);
         var now = DateTimeOffset.UtcNow;
@@ -265,7 +271,8 @@ internal IDatabaseOperation[] buildOperationBatch(IReadOnlyList<int>? activeNode
             }
             else if (activeNodeNumbers is { Count: > 0 })
             {
-                ops.Add(new ReleaseOrphanedMessagesForAncillaryOperation(_database, activeNodeNumbers));
+                ops.Add(new ReleaseOrphanedMessagesForAncillaryOperation(_database, activeNodeNumbers,
+                    nodeNumberHighWaterMark));
             }
         }
 

src/Testing/CoreTests/Persistence/active_node_number_cache_tests.cs

@@ -94,4 +94,63 @@ public void one_cache_per_runtime()
         ActiveNodeNumberCache.For(new MockWolverineRuntime())
             .ShouldNotBeSameAs(ActiveNodeNumberCache.For(theRuntime));
     }
+
+    // ─── GH-3850: the staleness window, stated rather than implied ───────────────────────────
+
+    [Fact]
+    public async Task the_high_water_mark_is_the_highest_number_seen()
+    {
+        var cache = theCache;
+        await cache.FetchAsync(CancellationToken.None);
+
+        cache.HighWaterMark.ShouldBe(3);
+    }
+
+    [Fact]
+    public async Task the_high_water_mark_does_not_drop_when_the_highest_node_departs()
+    {
+        var cache = theCache;
+        await cache.FetchAsync(CancellationToken.None);
+
+        // node 3 -- the highest -- dies
+        nodesAre(1, 2);
+        await Task.Delay(theRuntime.DurabilitySettings.ScheduledJobPollingTime + 100.Milliseconds(),
+            TestContext.Current.CancellationToken);
+        await cache.FetchAsync(CancellationToken.None);
+
+        // max(active) would now be 2, which would put node 3's orphaned messages permanently out of
+        // reach of the release. The mark is monotonic precisely so that cannot happen.
+        cache.HighWaterMark.ShouldBe(3);
+    }
+
+    [Fact]
+    public async Task the_high_water_mark_rises_for_a_node_that_joins()
+    {
+        var cache = theCache;
+        await cache.FetchAsync(CancellationToken.None);
+
+        nodesAre(1, 2, 3, 9);
+        await Task.Delay(theRuntime.DurabilitySettings.ScheduledJobPollingTime + 100.Milliseconds(),
+            TestContext.Current.CancellationToken);
+        await cache.FetchAsync(CancellationToken.None);
+
+        cache.HighWaterMark.ShouldBe(9);
+    }
+
+    [Fact]
+    public async Task a_node_that_joins_mid_interval_is_above_the_mark_and_so_is_not_judged()
+    {
+        var cache = theCache;
+        await cache.FetchAsync(CancellationToken.None);
+
+        // Node 4 registers after the fetch. It is absent from the cached list, so the release would
+        // reset its in-flight rows to owner 0 and let another node claim work it is already doing --
+        // the one direction of staleness that is not benign. Its number is above the mark, which is
+        // what keeps the release away from it until the next fetch sees it.
+        const int joinedAfterTheFetch = 4;
+
+        var numbers = await cache.FetchAsync(CancellationToken.None);
+        numbers.ShouldNotContain(joinedAfterTheFetch);
+        joinedAfterTheFetch.ShouldBeGreaterThan(cache.HighWaterMark);
+    }
 }

src/Wolverine/Persistence/ActiveNodeNumberCache.cs

@@ -43,17 +43,47 @@ public static ActiveNodeNumberCache For(IWolverineRuntime runtime)
     private readonly SemaphoreSlim _gate = new(1, 1);
     private IReadOnlyList<int>? _numbers;
     private DateTimeOffset _staleAt = DateTimeOffset.MinValue;
+    private int _highWaterMark;
 
     internal ActiveNodeNumberCache(IWolverineRuntime runtime)
     {
         _runtime = runtime;
     }
 
+    /// <summary>
+    /// The highest node number this cache has <i>ever</i> observed — monotonic across fetches, not
+    /// recomputed from the current list.
+    /// </summary>
+    /// <remarks>
+    /// GH-3850. A cached list cannot describe a node that registered after it was taken, and the
+    /// release of orphaned messages is not symmetric about that: a list still naming a dead node
+    /// merely delays recovery by one interval, but a list missing a <i>live</i> node resets that
+    /// node's in-flight rows to <c>owner_id = 0</c> and lets another node claim work it is already
+    /// doing. Node numbers are database-generated and monotonic (<c>SERIAL</c> on Postgres,
+    /// <c>AutoNumber()</c> on SQL Server), so anything above this mark registered after the cache
+    /// was taken and must not be judged against it.
+    ///
+    /// <para>Deliberately <b>not</b> <c>max(active)</c>, which looks equivalent and is not: when the
+    /// highest-numbered node dies its number leaves the active list, the max drops below it, and its
+    /// orphaned messages become permanently unreclaimable. Keeping the mark monotonic means a node
+    /// that was ever seen stays reclaimable after it departs, while one never seen stays protected.</para>
+    ///
+    /// <para>The remaining gap is a node that both registers and departs between two fetches: never
+    /// observed, so its rows stay owned until the mark rises. That parks messages rather than
+    /// double-processing them — the safe direction — and the next registration clears it.</para>
+    /// </remarks>
+    public int HighWaterMark => _highWaterMark;
+
     /// <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.
+    ///
+    /// <para>Note that the gate serializes every database's timer callback behind one fetch, so a
+    /// <i>hung</i> <c>LoadAllNodesAsync</c> parks them all rather than letting each fail on its own
+    /// timeout. That is the intended trade — one slow query beats one per database, which is the
+    /// pool exhaustion this class exists to prevent — but it is worth knowing when reading a stall.</para>
     /// </summary>
     public async ValueTask<IReadOnlyList<int>> FetchAsync(CancellationToken token)
     {
@@ -70,6 +100,14 @@ public async ValueTask<IReadOnlyList<int>> FetchAsync(CancellationToken token)
             var numbers = nodes.Select(x => x.AssignedNodeNumber).ToList();
 
             _numbers = numbers;
+
+            // Monotonic on purpose -- see HighWaterMark. Raised, never lowered, so a departed node
+            // stays reclaimable while a node newer than this cache stays protected.
+            foreach (var number in numbers)
+            {
+                if (number > _highWaterMark) _highWaterMark = number;
+            }
+
             _staleAt = DateTimeOffset.UtcNow.Add(_runtime.DurabilitySettings.ScheduledJobPollingTime);
 
             return numbers;