diff --git a/src/Persistence/PersistenceTests/Durability/single_node_listener_recovery_exclusion.cs b/src/Persistence/PersistenceTests/Durability/single_node_listener_recovery_exclusion.cs index 05dc6107c..fe47c44d8 100644 --- a/src/Persistence/PersistenceTests/Durability/single_node_listener_recovery_exclusion.cs +++ b/src/Persistence/PersistenceTests/Durability/single_node_listener_recovery_exclusion.cs @@ -3,6 +3,7 @@ using NSubstitute; using Shouldly; using Wolverine; +using Wolverine.ComplianceTests.ExclusiveListeners; using Wolverine.Configuration; using Wolverine.Persistence.Durability; using Wolverine.Runtime.Agents; @@ -57,12 +58,19 @@ private async Task commandsFor(params Uri[] destinations) private IListenerCircuit acceptingCircuitFor(Uri uri, ListenerScope scope) { - var endpoint = new LocalQueue(uri.Segments.Last()) + // Deliberately NOT a LocalQueue: a local queue is never a single node listener no matter what its + // ListenerScope says, which is the whole point of GH-3856 below. + var endpoint = new SingleNodeListenerEndpoint(uri.Segments.Last()) { ListenerScope = scope, BufferingLimits = new BufferingLimits(500, 100) }; + return acceptingCircuitFor(uri, endpoint); + } + + private IListenerCircuit acceptingCircuitFor(Uri uri, Endpoint endpoint) + { var circuit = Substitute.For(); circuit.Endpoint.Returns(endpoint); circuit.Status.Returns(ListeningStatus.Accepting); @@ -144,4 +152,60 @@ public void determine_page_size_is_unchanged_for_competing_consumers() command.DeterminePageSize(circuit, new IncomingCount(theCompetingUri, 50), theSettings).ShouldBe(50); } + + private readonly Uri theLocalUri = new("local://activiteiten3"); + + private LocalQueue exclusiveLocalQueue() + { + // Exactly what PartitionedMessageTopology produces for PublishToPartitionedLocalMessaging(): a durable + // local queue forced to ListenerScope.Exclusive. + return new LocalQueue("activiteiten3") + { + ListenerScope = ListenerScope.Exclusive, + Mode = EndpointMode.Durable, + BufferingLimits = new BufferingLimits(500, 100) + }; + } + + /// + /// GH-3856. A local queue never gets a ListeningAgent -- LocalQueue.BuildListenerAsync() throws and + /// StartListenersAsync() filters local queues out -- so nothing ever starts the ListenerInboxRecoveryLoop + /// that the GH-3590 carve-out hands ownership to. If the durability agent skips it too, its dormant inbox + /// rows are recovered by nobody and sit at owner_id = 0 forever. + /// + [Fact] + public void a_local_queue_is_never_a_single_node_listener_whatever_its_scope() + { + exclusiveLocalQueue().IsSingleNodeListener.ShouldBeFalse(); + + new LocalQueue("pinned") { ListenerScope = ListenerScope.PinnedToLeader } + .IsSingleNodeListener.ShouldBeFalse(); + } + + [Fact] + public async Task still_recovers_for_an_exclusive_local_queue() + { + acceptingCircuitFor(theLocalUri, exclusiveLocalQueue()); + theEndpoints.IsSingleNodeListener(theLocalUri).Returns(false); + + var commands = await commandsFor(theLocalUri); + + commands.Single().ShouldBeOfType(); + } + + /// + /// GH-3856. The second, independent guard. Getting past the CheckRecoverableIncomingMessagesOperation skip + /// is not enough on its own -- DeterminePageSize() used to test the raw ListenerScope, so an exclusive local + /// queue got a command issued and then recovered zero rows on every single pass. + /// + [Fact] + public void determine_page_size_is_unchanged_for_an_exclusive_local_queue() + { + var circuit = acceptingCircuitFor(theLocalUri, exclusiveLocalQueue()); + + var command = new RecoverIncomingMessagesCommand(theDatabase, new IncomingCount(theLocalUri, 50), + circuit, theSettings, NullLogger.Instance); + + command.DeterminePageSize(circuit, new IncomingCount(theLocalUri, 50), theSettings).ShouldBe(50); + } } diff --git a/src/Persistence/PostgresqlTests/Durability/partitioned_local_queue_inbox_recovery.cs b/src/Persistence/PostgresqlTests/Durability/partitioned_local_queue_inbox_recovery.cs new file mode 100644 index 000000000..63f13fc71 --- /dev/null +++ b/src/Persistence/PostgresqlTests/Durability/partitioned_local_queue_inbox_recovery.cs @@ -0,0 +1,135 @@ +using IntegrationTests; +using JasperFx.Core; +using JasperFx.Resources; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine; +using Wolverine.ComplianceTests.ExclusiveListeners; +using Wolverine.Persistence.Durability; +using Wolverine.Postgresql; +using Wolverine.Runtime; +using Wolverine.Tracking; +using Wolverine.Transports; +using Wolverine.Util; + +namespace PostgresqlTests.Durability; + +/// +/// GH-3856. PublishToPartitionedLocalMessaging() stamps ListenerScope.Exclusive onto every one of its durable +/// local queues, and the GH-3590 carve-out then handed inbox recovery for those queues to a +/// ListenerInboxRecoveryLoop that is never constructed for a local queue — LocalQueue.BuildListenerAsync() +/// throws and StartListenersAsync() filters local queues out, so they never get a ListeningAgent at all. +/// +/// The result in the field was thousands of envelopes sitting at status = 'Incoming', owner_id = 0 for hours, +/// surviving rolling deploys, because neither recovery path would claim them. This test reproduces exactly +/// that state and asserts that the durability agent — which IS a valid owner here, since a local queue exists +/// on every node — drains it. +/// +public class partitioned_local_queue_inbox_recovery : IAsyncLifetime +{ + private IHost _host = null!; + + public async ValueTask InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Durability.Mode = DurabilityMode.Solo; + + // Keep the polling tight so the test doesn't wait out the 5 second default + opts.Durability.ScheduledJobPollingTime = 250.Milliseconds(); + + opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, "partitioned_local_recovery"); + + opts.Discovery.DisableConventionalDiscovery().IncludeType(); + + opts.MessagePartitioning.ByMessage(x => x.Id.ToString()); + + opts.MessagePartitioning.PublishToPartitionedLocalMessaging("activiteiten", 4, topology => + { + topology.Message(); + topology.ConfigureQueues(q => q.UseDurableInbox()); + }); + }).StartAsync(); + + await _host.ResetResourceState(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + [Fact] + public async Task durability_agent_recovers_dormant_rows_for_a_partitioned_local_queue() + { + using var tracking = RecoveredMessages.Track(); + + var runtime = _host.GetRuntime(); + var store = _host.Services.GetRequiredService(); + + var queue = runtime.Endpoints.EndpointByName("activiteiten3")!; + + // Nothing ever builds a ListeningAgent for a local queue, which is precisely why the durability agent + // has to be the one to claim these rows. + runtime.Endpoints.FindListeningAgent(queue.Uri).ShouldBeNull(); + + var seeded = await seedDormantMessagesAsync(store, runtime, queue.Uri, 5); + var expected = seeded.Select(x => x.Id).ToArray(); + + var succeeded = await waitForAsync(() => expected.All(tracking.Contains), 30.Seconds()); + + succeeded.ShouldBeTrue( + $"Expected the durability agent to recover all {seeded.Length} dormant inbox rows for the " + + $"partitioned local queue {queue.Uri}, but only saw {tracking.Count}"); + + (await store.LoadPageOfGloballyOwnedIncomingAsync(queue.Uri, 100)).ShouldBeEmpty(); + } + + private static async Task seedDormantMessagesAsync(IMessageStore store, IWolverineRuntime runtime, + Uri destination, int count) + { + var serializer = runtime.Options.DefaultSerializer!; + + var envelopes = Enumerable.Range(0, count).Select(i => + { + var id = Guid.NewGuid(); + var envelope = new Envelope(new RecoveredMessage(id, i)) + { + Id = id, + Destination = destination, + Status = EnvelopeStatus.Incoming, + OwnerId = TransportConstants.AnyNode, + ContentType = serializer.ContentType, + MessageType = typeof(RecoveredMessage).ToMessageTypeName(), + SentAt = DateTimeOffset.UtcNow + }; + + envelope.Data = serializer.Write(envelope); + + return envelope; + }).ToArray(); + + await store.Inbox.StoreIncomingAsync(envelopes); + + return envelopes; + } + + private static async Task waitForAsync(Func condition, TimeSpan timeout) + { + var stopwatch = System.Diagnostics.Stopwatch.StartNew(); + while (stopwatch.Elapsed < timeout) + { + if (condition()) + { + return true; + } + + await Task.Delay(100.Milliseconds()); + } + + return condition(); + } +} diff --git a/src/Testing/CoreTests/Configuration/is_single_node_listener.cs b/src/Testing/CoreTests/Configuration/is_single_node_listener.cs index 8d457d508..52dead456 100644 --- a/src/Testing/CoreTests/Configuration/is_single_node_listener.cs +++ b/src/Testing/CoreTests/Configuration/is_single_node_listener.cs @@ -5,6 +5,7 @@ using Wolverine; using Wolverine.Configuration; using Wolverine.Tracking; +using Wolverine.Transports.Local; using Wolverine.Transports.Tcp; using Wolverine.Util; using Xunit; @@ -66,4 +67,51 @@ public Task unknown_address_is_not_a_single_node_listener() return withHostAsync(host => host.GetRuntime().Endpoints.IsSingleNodeListener(new Uri("tcp://localhost:65001")).ShouldBeFalse()); } + + /// + /// GH-3856. PartitionedMessageTopology forces ListenerScope.Exclusive onto every slot, local queues + /// included, but a local queue never gets a ListeningAgent and so never starts the + /// ListenerInboxRecoveryLoop that the GH-3590 carve-out hands recovery to. Answering "true" here left the + /// dormant inbox rows for these queues owned by nobody at all. + /// + [Fact] + public async Task partitioned_local_queues_are_not_single_node_listeners() + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.MessagePartitioning.PublishToPartitionedLocalMessaging("activiteiten", 4, topology => + { + topology.MessagesImplementing(); + topology.ConfigureQueues(q => q.UseDurableInbox()); + }); + + opts.Durability.Mode = DurabilityMode.Solo; + }).StartAsync(TestContext.Current.CancellationToken); + + var endpoints = host.GetRuntime().Endpoints; + + foreach (var name in new[] { "activiteiten1", "activiteiten2", "activiteiten3", "activiteiten4" }) + { + var queue = (LocalQueue)endpoints.EndpointByName(name)!; + + // The topology really does stamp Exclusive onto the local queue... + queue.ListenerScope.ShouldBe(ListenerScope.Exclusive); + + // ...and the durability agent must claim its inbox rows anyway. + queue.IsSingleNodeListener.ShouldBeFalse(); + endpoints.IsSingleNodeListener(queue.Uri).ShouldBeFalse(); + } + } +} + +public interface IPartitionedLocalMessage; + +public record PartitionedLocalOne(Guid Id) : IPartitionedLocalMessage; + +public static class PartitionedLocalMessageHandler +{ + public static void Handle(PartitionedLocalOne message) + { + } } diff --git a/src/Wolverine/Configuration/Endpoint.cs b/src/Wolverine/Configuration/Endpoint.cs index f48ee2ce7..bcd3cc4f6 100644 --- a/src/Wolverine/Configuration/Endpoint.cs +++ b/src/Wolverine/Configuration/Endpoint.cs @@ -234,6 +234,15 @@ protected Endpoint(Uri uri, EndpointRole role) /// public ListenerScope ListenerScope { get; set; } = ListenerScope.CompetingConsumers; + /// + /// GH-3590. Is this endpoint's listener only ever active on ONE node of the cluster? Inbox recovery for + /// such an endpoint is owned by the node hosting the listener () rather + /// than by the per-database durability agent, which is assigned per database and routinely lands on a + /// different node. Every guard that implements that hand-off asks *this* question, so that the two sides + /// can never disagree and strand messages in between. + /// + internal virtual bool IsSingleNodeListener => ListenerScope != ListenerScope.CompetingConsumers; + /// /// Is OpenTelemetry enabled for this endpoint? /// diff --git a/src/Wolverine/Configuration/EndpointCollection.cs b/src/Wolverine/Configuration/EndpointCollection.cs index 1bd0372e1..ed971bacf 100644 --- a/src/Wolverine/Configuration/EndpointCollection.cs +++ b/src/Wolverine/Configuration/EndpointCollection.cs @@ -57,7 +57,7 @@ public interface IEndpointCollection : IAsyncDisposable /// bool IsSingleNodeListener(Uri address) { - return EndpointFor(address) is { ListenerScope: not ListenerScope.CompetingConsumers }; + return EndpointFor(address) is { IsSingleNodeListener: true }; } } @@ -381,7 +381,7 @@ public bool IsSingleNodeListener(Uri address) return isSingleNode; } - isSingleNode = EndpointFor(address) is { ListenerScope: not ListenerScope.CompetingConsumers }; + isSingleNode = EndpointFor(address) is { IsSingleNodeListener: true }; _singleNodeListeners = _singleNodeListeners.AddOrUpdate(address, isSingleNode); return isSingleNode; diff --git a/src/Wolverine/Persistence/Durability/RecoverIncomingMessagesCommand.cs b/src/Wolverine/Persistence/Durability/RecoverIncomingMessagesCommand.cs index 65a8e0dc0..52cf82a99 100644 --- a/src/Wolverine/Persistence/Durability/RecoverIncomingMessagesCommand.cs +++ b/src/Wolverine/Persistence/Durability/RecoverIncomingMessagesCommand.cs @@ -110,7 +110,10 @@ public virtual int DeterminePageSize(IListenerCircuit listener, IncomingCount co // GH-3590 defense in depth. Inbox recovery for single node listeners (Exclusive / PinnedToLeader) is // owned by the node that is actually hosting the listener, never by the per-database durability agent. - if (listener.Endpoint.ListenerScope != ListenerScope.CompetingConsumers) + // GH-3856: this MUST ask the same question as IEndpointCollection.IsSingleNodeListener(). Testing the + // raw ListenerScope here meant a local queue carrying ListenerScope.Exclusive got a recovery command + // issued and then silently recovered nothing, forever. + if (listener.Endpoint.IsSingleNodeListener) { return 0; } diff --git a/src/Wolverine/Transports/ListeningAgent.cs b/src/Wolverine/Transports/ListeningAgent.cs index d9549cc83..3787ed20f 100644 --- a/src/Wolverine/Transports/ListeningAgent.cs +++ b/src/Wolverine/Transports/ListeningAgent.cs @@ -403,7 +403,7 @@ public async ValueTask StartAsync() private void startInboxRecoveryIfNecessary() { if (Endpoint.Mode != EndpointMode.Durable) return; - if (Endpoint.ListenerScope == ListenerScope.CompetingConsumers) return; + if (!Endpoint.IsSingleNodeListener) return; if (!_runtime.Options.Durability.DurabilityAgentEnabled) return; if (_runtime.Storage is NullMessageStore) return; diff --git a/src/Wolverine/Transports/Local/LocalQueue.cs b/src/Wolverine/Transports/Local/LocalQueue.cs index ee8f2791e..2161a9891 100644 --- a/src/Wolverine/Transports/Local/LocalQueue.cs +++ b/src/Wolverine/Transports/Local/LocalQueue.cs @@ -18,6 +18,16 @@ public LocalQueue(string name) : base($"local://{name}".ToUri(), EndpointRole.Ap internal List HandledMessageTypes { get; } = new(); public int MessageCount => Agent?.As().QueueCount ?? 0; + /// + /// GH-3856. A local queue is NEVER a single node listener regardless of its . + /// It exists on every node, it never gets a (BuildListenerAsync throws), and + /// EndpointCollection.ExclusiveListeners() excludes it, so nothing ever starts the + /// ListenerInboxRecoveryLoop that would otherwise own its inbox recovery. The per-database durability + /// agent is the only recovery path a local queue has, and it is a perfectly good owner precisely because + /// the queue lives on whichever node that agent happens to run on. + /// + internal override bool IsSingleNodeListener => false; + public override bool ShouldEnforceBackPressure() { return false;