Recover dormant inbox rows for durable local queues (GH-3856)

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<IAgentCommand[]> 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<IListeningAgent, IListenerCircuit>();
         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)
+        };
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    [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<RecoverIncomingMessagesCommand>();
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    [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);
+    }
 }

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+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<RecoveredMessageHandler>();
+
+                opts.MessagePartitioning.ByMessage<RecoveredMessage>(x => x.Id.ToString());
+
+                opts.MessagePartitioning.PublishToPartitionedLocalMessaging("activiteiten", 4, topology =>
+                {
+                    topology.Message<RecoveredMessage>();
+                    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<IMessageStore>();
+
+        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<Envelope[]> 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<bool> waitForAsync(Func<bool> condition, TimeSpan timeout)
+    {
+        var stopwatch = System.Diagnostics.Stopwatch.StartNew();
+        while (stopwatch.Elapsed < timeout)
+        {
+            if (condition())
+            {
+                return true;
+            }
+
+            await Task.Delay(100.Milliseconds());
+        }
+
+        return condition();
+    }
+}

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());
     }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    [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<IPartitionedLocalMessage>();
+                    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)
+    {
+    }
 }

src/Wolverine/Configuration/Endpoint.cs

@@ -234,6 +234,15 @@ protected Endpoint(Uri uri, EndpointRole role)
     /// </summary>
     public ListenerScope ListenerScope { get; set; } = ListenerScope.CompetingConsumers;
 
+    /// <summary>
+    /// 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 (<see cref="ListenerInboxRecovery"/>) 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.
+    /// </summary>
+    internal virtual bool IsSingleNodeListener => ListenerScope != ListenerScope.CompetingConsumers;
+
     /// <summary>
     /// Is OpenTelemetry enabled for this endpoint?
     /// </summary>

src/Wolverine/Configuration/EndpointCollection.cs

@@ -57,7 +57,7 @@ public interface IEndpointCollection : IAsyncDisposable
     /// </summary>
     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;

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

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;
 

src/Wolverine/Transports/Local/LocalQueue.cs

@@ -18,6 +18,16 @@ public LocalQueue(string name) : base($"local://{name}".ToUri(), EndpointRole.Ap
     internal List<Type> HandledMessageTypes { get; } = new();
     public int MessageCount => Agent?.As<ILocalQueue>().QueueCount ?? 0;
 
+    /// <summary>
+    /// GH-3856. A local queue is NEVER a single node listener regardless of its <see cref="ListenerScope"/>.
+    /// It exists on every node, it never gets a <see cref="ListeningAgent"/> (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.
+    /// </summary>
+    internal override bool IsSingleNodeListener => false;
+
     public override bool ShouldEnforceBackPressure()
     {
         return false;