From 2c529d1be03937930b0df2fe005fa3a3e26a5e05 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Wed, 29 Jul 2026 20:06:50 -0500 Subject: [PATCH] test(rabbitmq): seed the departed node's inbox rows already owned (GH-3729) `rows_released_after_the_listener_is_already_running_are_still_recovered` failed about 40% of the time, and the flake was entirely in its own setup. The rows were stored claimable (`OwnerId = TransportConstants.AnyNode`) and only reassigned to the out-of-cluster node a database round-trip later: var seeded = await seedDormantMessagesAsync(listener, 5); await store.ReassignIncomingAsync(departedNode, seeded); The listener's recovery loop polls every `ScheduledJobPollingTime`, which this test sets to 250ms. A sweep landing between those two statements claimed the rows -- perfectly correct behaviour, they were owned by AnyNode at that instant. But the very next assertion exists to prove the listener leaves other nodes' rows alone: tracking.Count.ShouldBe(0, "The listener must not touch inbox rows that are still owned by another node"); So the test failed on rows it had itself published as claimable, not on any product behaviour. Worth being explicit that the ownership check was never at fault. `seedDormantMessagesAsync` now takes the owner to persist the rows WITH -- `StoreIncomingAsync` writes `envelope.OwnerId` straight through (DatabasePersistence.cs:78) -- so the rows are owned by the departed node from the moment they exist and the window is gone. The parameter is required rather than defaulted because `TransportConstants.AnyNode` is a static readonly field, not a constant, so it cannot be a default parameter value; both call sites now say which owner they mean. Verified: 12 consecutive green runs of the class (~12s each). At the previous ~40% rate that is p ~ 0.002. `dotnet build wolverine.slnx -c Release` clean with 0 warnings. Co-Authored-By: Claude Opus 5 (1M context) --- .../multi_node_exclusive_listener_recovery.cs | 24 +++++++++++++------ 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/multi_node_exclusive_listener_recovery.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/multi_node_exclusive_listener_recovery.cs index 3270a9579..4e455211a 100644 --- a/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/multi_node_exclusive_listener_recovery.cs +++ b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/multi_node_exclusive_listener_recovery.cs @@ -200,7 +200,12 @@ private static async Task pinListenerToAsync(IHost leader, int nodeNumber) } - private static async Task seedDormantMessagesAsync(IHost host, int count) + /// + /// Seeds inbox rows directly. is the owner they are persisted WITH, which + /// matters: staging rows as claimable and reassigning them afterwards leaves a window in which the + /// listener's recovery loop can legitimately claim them. See GH-3729. + /// + private static async Task seedDormantMessagesAsync(IHost host, int count, int ownerId) { var runtime = host.GetRuntime(); var store = host.Services.GetRequiredService(); @@ -215,9 +220,7 @@ private static async Task seedDormantMessagesAsync(IHost host, int c Destination = queueUri(), Status = EnvelopeStatus.Incoming, - // Exactly how an ungraceful shutdown leaves them, and how the durability agent releases - // the rows a dead node owned. - OwnerId = TransportConstants.AnyNode, + OwnerId = ownerId, ContentType = serializer.ContentType, MessageType = typeof(NodeTrackedMessage).ToMessageTypeName(), SentAt = DateTimeOffset.UtcNow @@ -268,7 +271,9 @@ public async Task the_listening_node_recovers_the_inbox_when_the_durability_agen // these two agents are on different nodes. listenerNode.ShouldNotBe(agentNode); - var seeded = await seedDormantMessagesAsync(listener, 5); + // Dormant and claimable from the outset -- exactly how an ungraceful shutdown leaves them once the + // durability agent has released the rows a dead node owned. + var seeded = await seedDormantMessagesAsync(listener, 5, TransportConstants.AnyNode); var expected = seeded.Select(x => x.Id).ToArray(); var recovered = await waitForAsync(() => expected.All(tracking.Contains), 60.Seconds()); @@ -315,12 +320,17 @@ public async Task rows_released_after_the_listener_is_already_running_are_still_ var listenerNode = nodeNumberOf(listener); var store = listener.Services.GetRequiredService(); - var seeded = await seedDormantMessagesAsync(listener, 5); // Owned by a node that is not part of this cluster -- the state a dead node's in-flight batch is in // before the durability agent gets around to releasing it. + // + // GH-3729: these have to be seeded owned by the departed node, not stored as claimable and reassigned + // a round-trip later. The listener's recovery loop polls every ScheduledJobPollingTime (250ms here), so + // a sweep landing in that window claimed the rows perfectly correctly -- and then the assertion below, + // which exists to prove the listener leaves other nodes' rows alone, failed on the test's own setup + // rather than on any product behaviour. That was roughly 40% of runs. const int departedNode = 987654; - await store.ReassignIncomingAsync(departedNode, seeded); + var seeded = await seedDormantMessagesAsync(listener, 5, departedNode); (await store.LoadPageOfGloballyOwnedIncomingAsync(queueUri(), 100)) .ShouldBeEmpty("Nothing should be dormant yet -- the rows are still owned by the departed node");