diff --git a/src/Persistence/SqlServerTests/release_orphaned_ancillary_high_water_mark_3850.cs b/src/Persistence/SqlServerTests/release_orphaned_ancillary_high_water_mark_3850.cs new file mode 100644 index 000000000..4da2aacb4 --- /dev/null +++ b/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; + +/// +/// 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 live +/// node's rows to owner_id = 0 hands its in-flight work to somebody else, so the release is +/// bounded by the highest node number the cache has ever seen. +/// +public class release_orphaned_ancillary_high_water_mark_3850 +{ + private static string sqlFor(IReadOnlyList activeNodeNumbers, int highWaterMark) + { + var database = Substitute.For(); + 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); + } +} diff --git a/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesForAncillaryOperation.cs b/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesForAncillaryOperation.cs index 6bc606345..f7e659fd5 100644 --- a/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesForAncillaryOperation.cs +++ b/src/Persistence/Wolverine.RDBMS/Durability/ReleaseOrphanedMessagesForAncillaryOperation.cs @@ -16,11 +16,19 @@ internal class ReleaseOrphanedMessagesForAncillaryOperation : IDatabaseOperation { private readonly IMessageDatabase _database; private readonly IReadOnlyList _activeNodeNumbers; + private readonly int _highWaterMark; - public ReleaseOrphanedMessagesForAncillaryOperation(IMessageDatabase database, IReadOnlyList activeNodeNumbers) + /// + /// 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. + /// + public ReleaseOrphanedMessagesForAncillaryOperation(IMessageDatabase database, + IReadOnlyList 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 exceptions, CancellationToken token) diff --git a/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs b/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs index 80425085d..3d80dd950 100644 --- a/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs +++ b/src/Persistence/Wolverine.RDBMS/DurabilityAgent.cs @@ -95,6 +95,7 @@ public Task StartAsync(CancellationToken cancellationToken) _recoveryTimer = new Timer(async _ => { IReadOnlyList? activeNodeNumbers = null; + var nodeNumberHighWaterMark = 0; if (_settings.Mode != DurabilityMode.Solo && _database.Settings.Role != MessageStoreRole.Main) { try @@ -102,8 +103,12 @@ public Task StartAsync(CancellationToken cancellationToken) // 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) { @@ -111,7 +116,7 @@ public Task StartAsync(CancellationToken cancellationToken) } } - 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? activeNodeNumbers = null) + internal IDatabaseOperation[] buildOperationBatch(IReadOnlyList? 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? activeNode } else if (activeNodeNumbers is { Count: > 0 }) { - ops.Add(new ReleaseOrphanedMessagesForAncillaryOperation(_database, activeNodeNumbers)); + ops.Add(new ReleaseOrphanedMessagesForAncillaryOperation(_database, activeNodeNumbers, + nodeNumberHighWaterMark)); } } diff --git a/src/Testing/CoreTests/Persistence/active_node_number_cache_tests.cs b/src/Testing/CoreTests/Persistence/active_node_number_cache_tests.cs index bc8112429..db2aeb403 100644 --- a/src/Testing/CoreTests/Persistence/active_node_number_cache_tests.cs +++ b/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); + } } diff --git a/src/Wolverine/Persistence/ActiveNodeNumberCache.cs b/src/Wolverine/Persistence/ActiveNodeNumberCache.cs index d0dc4aa02..4966f40dc 100644 --- a/src/Wolverine/Persistence/ActiveNodeNumberCache.cs +++ b/src/Wolverine/Persistence/ActiveNodeNumberCache.cs @@ -43,17 +43,47 @@ public static ActiveNodeNumberCache For(IWolverineRuntime runtime) private readonly SemaphoreSlim _gate = new(1, 1); private IReadOnlyList? _numbers; private DateTimeOffset _staleAt = DateTimeOffset.MinValue; + private int _highWaterMark; internal ActiveNodeNumberCache(IWolverineRuntime runtime) { _runtime = runtime; } + /// + /// The highest node number this cache has ever observed — monotonic across fetches, not + /// recomputed from the current list. + /// + /// + /// 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 live node resets that + /// node's in-flight rows to owner_id = 0 and lets another node claim work it is already + /// doing. Node numbers are database-generated and monotonic (SERIAL on Postgres, + /// AutoNumber() on SQL Server), so anything above this mark registered after the cache + /// was taken and must not be judged against it. + /// + /// Deliberately not max(active), 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. + /// + /// 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. + /// + public int HighWaterMark => _highWaterMark; + /// /// 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. + /// + /// Note that the gate serializes every database's timer callback behind one fetch, so a + /// hung LoadAllNodesAsync 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. /// public async ValueTask> FetchAsync(CancellationToken token) { @@ -70,6 +100,14 @@ public async ValueTask> 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;