diff --git a/docs/guide/messaging/partitioning.md b/docs/guide/messaging/partitioning.md index a45602a63..6229e53e5 100644 --- a/docs/guide/messaging/partitioning.md +++ b/docs/guide/messaging/partitioning.md @@ -382,6 +382,10 @@ The example above uses the listener's default mode. Note what each mode costs yo The ordering guarantee is the same in every mode that supports it: messages sharing a group id never execute concurrently. Strict processing in original delivery order is *not* promised under failure or redelivery in any non-durable mode. + +Everything above is per *listener*. The cluster-wide equivalent -- one consumer per slot across every node, +with the same database-free ack behaviour -- is +[native ack global partitioning](#native-ack-global-partitioning). ::: ## Exempting Message Types from Partitioned Processing @@ -618,7 +622,8 @@ When you configure global partitioning, Wolverine: 3. **Support for modular monoliths** -- You can configure multiple global partitioning topologies for the same message type in different modules. Each module can have its own set of sharded queues and routing rules, allowing independent sequential processing pipelines within a single application. ::: tip -In single-node mode, global partitioning automatically shortcuts all messages to the companion local queues since the current node owns all listeners. +In single-node mode, global partitioning automatically shortcuts all messages to the companion local queues since the current node owns all listeners. The shortcut is disabled under +[`ProcessInParallelWithNativeAcks()`](#native-ack-global-partitioning), which has no companion local queues and where the broker delivery *is* the durability story -- every send goes through the broker even when this node owns the slot. ::: ### Configuration @@ -654,7 +659,7 @@ A couple of transport-specific notes: * **Kafka** -- all nodes listening to the sharded topics share a single Kafka consumer group named after the base name so that Kafka assigns each topic's partitions exclusively to one consumer at a time. Wolverine stamps that consumer group id onto the `GroupId` of incoming envelopes by default, which you can turn off per listener with `DisableConsumerGroupIdStamping()` when the consumer group name is not meaningful as envelope metadata (e.g. when combined with `PropagateGroupIdToPartitionKey()`). * **Azure Service Bus** -- the broker's native [session identifiers](/guide/messaging/transports/azureservicebus/session-identifiers) provide strictly ordered, per-session processing with a single queue and may be a simpler alternative if you are exclusively on Azure Service Bus. Global partitioning is the transport-agnostic option that behaves the same way across every broker in the table above. -* **PostgreSQL / Sql Server** -- the database queues need no extra infrastructure at all; each shard is just another pair of tables in the database you already have. They are inherently durable, which suits global partitioning since the topology forces `EndpointMode.Durable` on every slot anyway. The Sql Server shard queues additionally opt into the [`seq`-clustered high-throughput table layout](/guide/durability/sqlserver#optimizing-queue-throughput) by default. +* **PostgreSQL / Sql Server** -- the database queues need no extra infrastructure at all; each shard is just another pair of tables in the database you already have. They are inherently durable, which suits global partitioning since the topology defaults every slot to `EndpointMode.Durable`. The Sql Server shard queues additionally opt into the [`seq`-clustered high-throughput table layout](/guide/durability/sqlserver#optimizing-queue-throughput) by default. ### Example with RabbitMQ @@ -688,6 +693,86 @@ using var host = await Host.CreateDefaultBuilder() snippet source | anchor +### Native Ack Global Partitioning + +By default every slot in a global partitioned topology is `EndpointMode.Durable`, and the external listener +is bridged into a companion local queue that does the actual partitioned execution. That is a database write +per message on the way in (inbox insert) and another on the way out (mark handled), and the broker is +acknowledged as soon as the inbox row lands -- long before the handler runs. + +`ProcessInParallelWithNativeAcks()` removes both the database and the bridge: + +```csharp +opts.MessagePartitioning + .ByMessage(x => x.EntityId) + .GlobalPartitioned(topology => + { + topology.UseShardedRabbitQueues("webhooks", 5); + topology.Message(); + + // Each slot listener settles its own broker deliveries and shards + // into sequential lanes by group id in memory. No companion local + // queues, no bridge, no inbox. + topology.ProcessInParallelWithNativeAcks(); + }); +``` + +Each slot listener now holds its broker delivery **unacknowledged** until the handler reaches a terminal, and +shards incoming messages into sequential lanes by group id inside its own receiver. The transport has to opt +in to `EndpointMode.NativeAck`; RabbitMQ does, and a transport that does not will fail fast at bootstrap +naming the endpoint. See [Native Ack Endpoints](/guide/messaging/listeners#native-ack-endpoints). + +#### The guarantee + +**No two messages sharing a group id execute concurrently.** Within a node the sequential lane inside the +slot's receiver enforces it; across the cluster the exclusive slot listener enforces it, because exactly one +node consumes a given slot. + +Delivery is **at-least-once**, owned by the broker rather than by the inbox: anything received but not yet +settled is redelivered if the node dies or the channel drops, so nothing is lost, and a handler may see a +duplicate. Ordering is **per-slot best effort, not per-group guaranteed** -- redelivery or requeue may +reorder. And the caveat from the durable topology still applies unchanged: the ordering unit is the **slot**, +not the group, so two group ids that hash to the same slot serialize against each other. + +#### When to choose it over the durable default + +Reach for it when the traffic is a flood that the database cannot absorb -- webhook storms, telemetry, +event fan-in -- and the handler's own work is what actually needs to be durable, not the transport hop. +Concretely you are trading: + +| | Durable slots (default) | `ProcessInParallelWithNativeAcks()` | +|---|---|---| +| Database per message | inbox insert + mark handled | none | +| Broker ack timing | at inbox insert, before the handler | at handler completion | +| Loss on node death | none | none (redelivered) | +| Duplicates | suppressed by inbox dedup | possible; handlers must tolerate them | +| Outbox atomicity with handler side effects | yes | no | +| Recovery of stranded messages | inbox recovery | broker redelivery | +| Back pressure | in-process listener circuit | the broker's prefetch window | + +Because there is no inbox, there is also no inbox dedup and no outbox atomicity between a handler's database +work and the messages it publishes. Handlers must be idempotent. + +#### Failover caveat + +Slot failover is the one place where the cluster-wide half of the guarantee is actually load bearing. When a +slot moves nodes, the outgoing node stops **and drains** its listener -- it finishes the handlers already +running and settles them -- before the leader is allowed to start the slot anywhere else. Only then does the +incoming node begin pulling. That drain is what keeps a group from running in two places at once across a +handoff, and it is why a redelivery after failover lands on the new owner rather than alongside the old one. + +The price is that a slot is briefly unconsumed during the handoff, and that in-flight messages the drain +could not finish inside `DurabilitySettings.DrainTimeout` are redelivered rather than completed. Both are +consistent with at-least-once; neither is consistent with "exactly once". + +::: warning Slot ownership still needs somewhere to coordinate +The *messages* touch no database in this mode, but Wolverine's dynamic one-consumer-per-slot assignment is +done by the node agent framework, which needs a message store to persist node records and agent assignments +-- that is, `DurabilityMode.Balanced`. A host with no message store at all runs in `DurabilityMode.Solo`, +where *every* node starts *every* listener, so a multi-node storage-free deployment has to assign slots to +nodes itself (each node listening only to the slots it owns). Single-node deployments are unaffected. +::: + ### Excluding Message Types `Except()` carves a message type -- or a whole family, when given an interface or base class -- @@ -736,6 +821,11 @@ Wolverine validates global partitioning configuration at startup. It will throw - No external transport topology is configured - The external and local topologies have different shard counts +The last rule does not apply to a [native ack topology](#native-ack-global-partitioning), which deliberately +has no companion local topology at all. Calling `LocalQueues()` and `ProcessInParallelWithNativeAcks()` +together throws, as does `Mode(EndpointMode.NativeAck)` -- that would set the mode while leaving the bridge in +place, and a local queue has no broker delivery to settle. + ### Native Per-Transport Alternatives Global partitioning is the *portable* answer: it behaves identically on all ten transports because diff --git a/src/Testing/CoreTests/Runtime/WorkerQueues/latched_receiver_contract_3709.cs b/src/Testing/CoreTests/Runtime/WorkerQueues/latched_receiver_contract_3709.cs new file mode 100644 index 000000000..21aaf58ce --- /dev/null +++ b/src/Testing/CoreTests/Runtime/WorkerQueues/latched_receiver_contract_3709.cs @@ -0,0 +1,56 @@ +using JasperFx.Core; +using JasperFx.Core.Reflection; +using Shouldly; +using Wolverine.Runtime.WorkerQueues; +using Wolverine.Transports; +using Xunit; + +namespace CoreTests.Runtime.WorkerQueues; + +/// +/// GH-3709. ListeningAgent.LatchReceiver() used to be an if/else chain naming +/// DurableReceiver, BufferedReceiver and InlineReceiver one at a time. When GH-3708 +/// added the chain silently skipped it, and the consequence was not a +/// compile error or an exception but a stop-and-drain that stopped waiting: an unlatched receiver's +/// DrainAsync returns immediately, so the transport channel closed underneath running handlers, every +/// unsettled delivery went back to the broker, and on an exclusive listener handoff the incoming node re-ran +/// those messages concurrently with the outgoing one -- precisely the intra-group concurrency that +/// partitioned processing exists to prevent. +/// +public class latched_receiver_contract_3709 +{ + /// + /// A receiver that knows how to latch has to say so through ILatchedReceiver, because that is the + /// only thing LatchReceiver() looks at now. A Latch() method that no caller can see is the + /// exact shape of the original bug. + /// + [Fact] + public void every_receiver_that_can_latch_declares_ILatchedReceiver() + { + var offenders = typeof(IReceiver).Assembly + .GetTypes() + .Where(x => x is { IsClass: true, IsAbstract: false } && x.CanBeCastTo()) + .Where(x => x.GetMethod("Latch", Type.EmptyTypes) != null) + .Where(x => !x.CanBeCastTo()) + .Select(x => x.FullNameInCode()) + .ToArray(); + + offenders.ShouldBeEmpty( + "These receivers have a Latch() method that ListeningAgent.LatchReceiver() cannot see: " + + offenders.Join(", ")); + } + + /// + /// And the four that exist today are all wired up, so the guard above is asserting against a non-empty + /// population rather than passing vacuously. + /// + [Theory] + [InlineData(typeof(DurableReceiver))] + [InlineData(typeof(BufferedReceiver))] + [InlineData(typeof(InlineReceiver))] + [InlineData(typeof(NativeAckReceiver))] + public void the_known_receivers_are_latchable(Type receiverType) + { + receiverType.CanBeCastTo().ShouldBeTrue(); + } +} diff --git a/src/Testing/CoreTests/Runtime/WorkerQueues/native_ack_receiver.cs b/src/Testing/CoreTests/Runtime/WorkerQueues/native_ack_receiver.cs index e1565c001..9544f9bad 100644 --- a/src/Testing/CoreTests/Runtime/WorkerQueues/native_ack_receiver.cs +++ b/src/Testing/CoreTests/Runtime/WorkerQueues/native_ack_receiver.cs @@ -25,8 +25,14 @@ public class NativeAckPingHandler /// Gate so a test can observe the receiver mid-flight without Task.Delay. public static TaskCompletionSource? Gate; + /// Signals that the handler has actually started, so a test can latch mid-flight rather than + /// racing the block. + public static TaskCompletionSource? Entered; + public async Task Handle(NativeAckPing message) { + Entered?.TrySetResult(); + if (Gate != null) { await Gate.Task; @@ -56,11 +62,13 @@ public async ValueTask InitializeAsync() theRuntime = _host.Services.GetRequiredService(); NativeAckPingHandler.Handled.Clear(); NativeAckPingHandler.Gate = null; + NativeAckPingHandler.Entered = null; } public async ValueTask DisposeAsync() { NativeAckPingHandler.Gate = null; + NativeAckPingHandler.Entered = null; await _host.StopAsync(); _host.Dispose(); } @@ -137,6 +145,42 @@ public async Task a_latched_receiver_hands_the_delivery_back_to_the_broker() NativeAckPingHandler.Handled.ShouldBeEmpty(); } + /// + /// GH-3709. The half of the drain contract that an exclusive listener handoff depends on: once the + /// receiver has been latched, DrainAsync must not report done while a handler is still running. + /// If it returns early the listener is disposed underneath live work, the still-unsettled deliveries go + /// back to the broker, and the node taking the listener over runs them concurrently with the node that + /// is still finishing them. + /// + [Fact] + public async Task draining_a_latched_receiver_waits_for_the_in_flight_handler() + { + var receiver = receiverFor(); + var listener = new RecordingListener(); + + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + NativeAckPingHandler.Gate = gate; + NativeAckPingHandler.Entered = entered; + + await receiver.ReceivedAsync(listener, pingEnvelope("in-flight")); + await entered.Task.WaitAsync(30.Seconds(), TestContext.Current.CancellationToken); + + receiver.Latch(); + var drain = receiver.DrainAsync().AsTask(); + + var raced = await Task.WhenAny(drain, Task.Delay(250.Milliseconds(), TestContext.Current.CancellationToken)); + raced.ShouldNotBeSameAs(drain); + + gate.SetResult(); + + await drain.WaitAsync(30.Seconds(), TestContext.Current.CancellationToken); + + listener.Completed.Count.ShouldBe(1); + listener.Deferred.Count.ShouldBe(0); + NativeAckPingHandler.Handled.ShouldContain("in-flight"); + } + /// /// The reason GH-4013 added the per-envelope channel-source overload. With ListenerCount > 1 the receiver /// is shared across listeners, so a single bound IChannelCallback would settle the wrong delivery. diff --git a/src/Testing/CoreTests/Runtime/pooled_outgoing_envelope_metrics_race_3709.cs b/src/Testing/CoreTests/Runtime/pooled_outgoing_envelope_metrics_race_3709.cs new file mode 100644 index 000000000..80ec1ea82 --- /dev/null +++ b/src/Testing/CoreTests/Runtime/pooled_outgoing_envelope_metrics_race_3709.cs @@ -0,0 +1,187 @@ +using JasperFx.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using Shouldly; +using Wolverine; +using Wolverine.Logging; +using Wolverine.Runtime; +using Wolverine.Transports.Sending; +using Wolverine.Transports.Stub; +using Xunit; + +namespace CoreTests.Runtime; + +/// +/// GH-3709. hands out POOLED outgoing envelopes +/// (WolverineRuntime.AcquireOutgoingEnvelope, wolverine#2955) and its storeAndForwardAsync does +/// nothing but post the envelope to an in-memory block. The block's consumer then sends it, succeeds, and +/// returns it to the pool -- Envelope.Reset(), which nulls Destination and MessageType +/// and clears FromPool -- so any read of that envelope after the post races a thread that is actively +/// blanking it. +/// +/// +/// The read that existed was _messageLogger.Sent(envelope) at the end of +/// SendingAgent.StoreAndForwardAsync, and the symptom was an intermittent +/// out of Envelope.ToMetricsHeaders(): Destination passed +/// its own null guard and was null one line later at Destination.ToString(). +/// +/// It was found through , which mapped to +/// at the time. GH-4061 has since moved NativeAck onto +/// , so that particular trigger is gone -- but nothing about the race was +/// ever native-ack specific. The remaining exposure is any BufferedInMemory endpoint, which is what both +/// tests below drive. +/// +public class pooled_outgoing_envelope_metrics_race_3709 : IAsyncLifetime +{ + private const int Messages = 2000; + + private IHost _host = null!; + private WolverineRuntime theRuntime = null!; + + public async ValueTask InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => opts.Discovery.DisableConventionalDiscovery()) + .StartAsync(TestContext.Current.CancellationToken); + + theRuntime = (WolverineRuntime)_host.Services.GetRequiredService(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + private static StubEndpoint bufferedEndpoint() => new("pooled-metrics-3709", new StubTransport()); + + private BufferedSendingAgent agentFor(StubEndpoint endpoint, ISender sender, IMessageTracker tracker) + { + return new BufferedSendingAgent(NullLogger.Instance, tracker, sender, theRuntime.DurabilitySettings, + endpoint, theRuntime, null); + } + + private Envelope pooledEnvelopeFor(BufferedSendingAgent agent, Uri destination, int number) + { + var envelope = theRuntime.AcquireOutgoingEnvelope(agent); + + // Only a pooled envelope can be recycled out from under the caller, so a run where the pool gate + // declined would prove nothing. + envelope.FromPool.ShouldBeTrue(); + + envelope.Message = new PooledMetricsProbe(number); + envelope.Destination = destination; + envelope.Sender = agent; + + return envelope; + } + + /// + /// The deterministic statement of the fix: for a pooled envelope the metrics read happens BEFORE the + /// envelope is handed to the sending block, so the recycle can never get there first. + /// + /// + /// The gate is what makes this deterministic rather than a race the test hopes to lose. The stand-in + /// tracker spins inside Sent() until the envelope has been recycled -- FromPool is cleared + /// by Envelope.Reset(), so it is the recycle flag -- or a short deadline passes. With the fix + /// nothing has been posted yet when Sent() runs, so no recycle is possible, the spin times out and + /// the envelope is read intact. Without it, Sent() runs after the post and the spin waits for + /// exactly the recycle it is racing, so Destination is reliably null by the time it is read. + /// + [Fact] + public async Task a_pooled_envelope_is_read_for_metrics_before_it_is_handed_to_the_sending_block() + { + var endpoint = bufferedEndpoint(); + var tracker = Substitute.For(); + + Uri? observedDestination = null; + var observedAtAll = false; + + tracker.When(x => x.Sent(Arg.Any())).Do(call => + { + var envelope = call.Arg(); + + var deadline = DateTimeOffset.UtcNow.Add(2.Seconds()); + while (envelope.FromPool && DateTimeOffset.UtcNow < deadline) + { + Thread.Sleep(1); + } + + observedDestination = envelope.Destination; + observedAtAll = true; + }); + + var agent = agentFor(endpoint, new ImmediateSender(endpoint.Uri), tracker); + + await agent.StoreAndForwardAsync(pooledEnvelopeFor(agent, endpoint.Uri, 0)); + + observedAtAll.ShouldBeTrue("The message tracker was never called at all"); + observedDestination.ShouldBe(endpoint.Uri, + "The metrics hook observed a recycled envelope -- the pooled read must happen before the handoff"); + } + + /// + /// The same invariant against the REAL message tracker under concurrency, + /// because Sent() is what actually walks the envelope to build its metric tags. This is the shape + /// the bug was originally caught in, so it stays -- but it is a probabilistic reproduction that depends on + /// machine load, which is why the deterministic gate above is the real guard. + /// + [Fact] + public async Task metrics_never_observe_a_recycled_envelope_under_concurrent_buffered_sends() + { + var endpoint = bufferedEndpoint(); + var sender = new ImmediateSender(endpoint.Uri); + var agent = agentFor(endpoint, sender, theRuntime.MessageTracking); + + await Parallel.ForEachAsync(Enumerable.Range(0, Messages), + new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount }, + async (i, _) => await agent.StoreAndForwardAsync(pooledEnvelopeFor(agent, endpoint.Uri, i))); + + // Nothing to assert beyond "it did not throw" -- the failure mode is a NullReferenceException raised + // inside the send, which Parallel.ForEachAsync surfaces straight out of the await above. Confirm the + // sends really happened so a silently latched agent cannot pass this vacuously. + await sender.WaitForAtLeastAsync(Messages, 30.Seconds()); + + sender.Count.ShouldBeGreaterThanOrEqualTo(Messages); + } + + public record PooledMetricsProbe(int Number); + + /// + /// Sends synchronously and does nothing else, so the block's consumer gets back to + /// sendWithExplicitHandlingAsync -- and therefore to the pool release -- as fast as possible. + /// + private class ImmediateSender : ISender + { + private int _count; + + public ImmediateSender(Uri destination) + { + Destination = destination; + } + + public int Count => Volatile.Read(ref _count); + + public bool SupportsNativeScheduledSend => false; + public Uri Destination { get; } + + public Task PingAsync() => Task.FromResult(true); + + public ValueTask SendAsync(Envelope envelope) + { + Interlocked.Increment(ref _count); + return ValueTask.CompletedTask; + } + + public async Task WaitForAtLeastAsync(int count, TimeSpan timeout) + { + var deadline = DateTimeOffset.UtcNow.Add(timeout); + while (Count < count && DateTimeOffset.UtcNow < deadline) + { + await Task.Delay(25); + } + } + } +} diff --git a/src/Testing/Wolverine.ComplianceTests/Partitioning/NativeAckPartitionedProcessing.cs b/src/Testing/Wolverine.ComplianceTests/Partitioning/NativeAckPartitionedProcessing.cs new file mode 100644 index 000000000..eaaa732ab --- /dev/null +++ b/src/Testing/Wolverine.ComplianceTests/Partitioning/NativeAckPartitionedProcessing.cs @@ -0,0 +1,248 @@ +using System.Collections.Concurrent; +using JasperFx.Core; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; + +namespace Wolverine.ComplianceTests.Partitioning; + +/// +/// GH-3709. The transport-agnostic harness for ProcessInParallelWithNativeAcks() on a global +/// partitioned topology, in the same spirit as (GH-3467): a transport +/// suite supplies only its own UseSharded*() call and the cluster shape, and everything else -- +/// the message type, the handler, the group ledger, the publishing burst, the assertions -- lives here. +/// +/// +/// The guarantee under test, stated exactly: no two messages sharing a group id execute +/// concurrently. Within a node the sequential lane inside the slot's own receiver enforces it; across the +/// cluster the exclusive slot listener enforces it, because exactly one node consumes a given slot. +/// +/// Ordering is per-slot best effort, not per-group guaranteed: redelivery or requeue may +/// reorder, and the ordering unit is the slot, not the group -- two groups hashing to the same slot +/// serialize against each other. Nothing here asserts ordering. +/// +/// Delivery is at-least-once and owned by the broker rather than by the inbox, so the completeness +/// assertion is -- every published letter seen at least +/// once. Duplicates are legal. +/// +public static class NativeAckPartitionedProcessing +{ + /// + /// The shared, cluster-wide ledger. Every host in a multi-node test runs in this same process, which is + /// exactly what makes a genuinely cluster-wide concurrency assertion possible. + /// + public static GroupConcurrencyLedger Ledger { get; } = new(); + + /// + /// How long each handler holds its group. This is the width of the window a concurrency violation has to + /// land in, so a test that wants to *catch* overlap needs it comfortably longer than the broker round trip. + /// + public static TimeSpan Dwell { get; set; } = 50.Milliseconds(); + + /// + /// Set up partitioning rules and handler discovery for , then call the + /// transport's own UseSharded*() plus ProcessInParallelWithNativeAcks() inside + /// . + /// + public static void UseNativeAckLetters(this WolverineOptions opts, string nodeName, + Action configureTopology) + { + opts.ServiceName = nodeName; + + opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(NativeAckLetterHandler)); + opts.Services.AddSingleton(new NativeAckNodeMarker(nodeName)); + + opts.MessagePartitioning.ByMessage(x => x.GroupId); + + opts.MessagePartitioning.GlobalPartitioned(topology => + { + configureTopology(topology); + topology.Message(); + }); + } + + /// + /// Publish letters for each of group + /// ids, spreading the publishing across every host so no single node is the sole producer. + /// + /// + /// One factory per host, not one bus per host: an is scoped and is not built to + /// be shared across concurrent invocations, so each parallel publisher below resolves its own. + /// + public static async Task> PumpOutLettersAsync( + IReadOnlyList> busSources, int groupCount, int messagesPerGroup) + { + var published = new ConcurrentQueue<(string, int)>(); + + var groups = Enumerable.Range(0, groupCount).Select(_ => Guid.NewGuid().ToString()).ToArray(); + + await Parallel.ForEachAsync(groups, async (groupId, _) => + { + var buses = busSources.Select(x => x()).ToArray(); + + for (var i = 0; i < messagesPerGroup; i++) + { + // Round robin the producer so the send path -- not just the receive path -- is exercised + // from more than one node. + var bus = buses[Math.Abs(groupId.GetHashCode() + i) % buses.Length]; + await bus.PublishAsync(new NativeAckLetter(groupId, i)); + published.Enqueue((groupId, i)); + } + }); + + return published.ToArray(); + } + + /// + /// Poll until every published letter has been handled at least once, or the timeout expires. Returns + /// true if everything landed. Polling rather than a tracked session because the multi-node scenarios + /// deliberately stop a host mid-stream. + /// + public static async Task WaitForCompletionAsync( + IReadOnlyCollection<(string GroupId, int Sequence)> published, TimeSpan timeout) + { + var deadline = DateTimeOffset.UtcNow.Add(timeout); + while (DateTimeOffset.UtcNow < deadline) + { + if (Ledger.OutstandingFrom(published).Count == 0) + { + return true; + } + + await Task.Delay(100.Milliseconds()); + } + + return Ledger.OutstandingFrom(published).Count == 0; + } + + /// + /// The heart of it: assert that no group id was ever executing in two places at once, anywhere in the + /// cluster. See for how overlap is detected. + /// + public static void AssertNoIntraGroupConcurrency() + { + Ledger.Handled.Count.ShouldBeGreaterThan(0, "Nothing was handled at all, so the invariant is untested"); + + Ledger.Violations.ShouldBeEmpty( + "Two messages sharing a group id executed concurrently: " + Ledger.Violations.Join(" | ")); + } + + /// + /// At-least-once completeness. Duplicates are expected and legal in this mode. + /// + public static void AssertEveryLetterWasHandled(IReadOnlyCollection<(string GroupId, int Sequence)> published) + { + var outstanding = Ledger.OutstandingFrom(published); + + outstanding.ShouldBeEmpty( + $"{outstanding.Count} of {published.Count} published letters were never handled: " + + outstanding.Take(10).Select(x => $"{x.GroupId}#{x.Sequence}").Join(", ")); + } + + /// + /// Every slot in the topology must have actually executed something, otherwise a "no concurrency" + /// result would be trivially satisfied by everything landing on one slot. + /// + public static void AssertEverySlotWasUsed(int numberOfSlots) + { + var destinations = Ledger.Handled.Select(x => x.Destination).Distinct().ToArray(); + + destinations.Length.ShouldBe(numberOfSlots, + $"Expected all {numberOfSlots} slots to be used. Saw: {destinations.Select(x => x?.ToString() ?? "null").Join(", ")}"); + } + + /// + /// A group must never straddle two slots -- that is the routing half of the guarantee, and it is what + /// makes the single exclusive consumer per slot sufficient for the cluster-wide half. + /// + public static void AssertGroupsNeverStraddleSlots() + { + foreach (var group in Ledger.Handled.GroupBy(x => x.GroupId)) + { + group.Select(x => x.Destination).Distinct().Count() + .ShouldBe(1, $"Group id {group.Key} was handled on more than one slot"); + } + } +} + +public record NativeAckLetter(string GroupId, int Sequence); + +/// +/// Injected per host so the ledger can name which node executed a message. All hosts in a multi-node test +/// share one process, so the node name has to come from configuration rather than from the environment. +/// +public class NativeAckNodeMarker(string nodeName) +{ + public string NodeName { get; } = nodeName; +} + +public static class NativeAckLetterHandler +{ + public static Task Handle(NativeAckLetter letter, Envelope envelope, NativeAckNodeMarker node) + { + return NativeAckPartitionedProcessing.Ledger.ExecuteAsync(letter, envelope.Destination, node.NodeName, + NativeAckPartitionedProcessing.Dwell); + } +} + +/// +/// Detects overlapping execution of one group id across the whole cluster. A handler claims its group id on +/// entry and releases it on exit; a claim that finds the group already held is recorded as a violation. +/// +/// +/// The release is a compare-and-remove on the claim token, not a blind remove, so a losing claimant can +/// never evict the rightful holder's entry and cascade one real violation into a string of phantom ones. +/// +public sealed class GroupConcurrencyLedger +{ + private readonly ConcurrentDictionary _inFlight = new(); + private readonly ConcurrentQueue _violations = new(); + private readonly ConcurrentQueue _handled = new(); + + public IReadOnlyList Violations => _violations.ToArray(); + public IReadOnlyList Handled => _handled.ToArray(); + + public void Clear() + { + _inFlight.Clear(); + _violations.Clear(); + _handled.Clear(); + } + + public async Task ExecuteAsync(NativeAckLetter letter, Uri? destination, string nodeName, TimeSpan dwell) + { + var claim = $"{nodeName}/{letter.Sequence}/{Guid.NewGuid():N}"; + + if (!_inFlight.TryAdd(letter.GroupId, claim)) + { + _inFlight.TryGetValue(letter.GroupId, out var holder); + _violations.Enqueue( + $"group {letter.GroupId} was held by {holder ?? "(released)"} when {claim} began executing on {destination}"); + } + + try + { + if (dwell > TimeSpan.Zero) + { + await Task.Delay(dwell); + } + + _handled.Enqueue(new HandledLetter(letter.GroupId, letter.Sequence, nodeName, destination)); + } + finally + { + _inFlight.TryRemove(new KeyValuePair(letter.GroupId, claim)); + } + } + + /// + /// The published letters that have not been handled even once yet. + /// + public IReadOnlyList<(string GroupId, int Sequence)> OutstandingFrom( + IReadOnlyCollection<(string GroupId, int Sequence)> published) + { + var seen = _handled.Select(x => (x.GroupId, x.Sequence)).ToHashSet(); + return published.Where(x => !seen.Contains(x)).ToArray(); + } + + public record HandledLetter(string GroupId, int Sequence, string NodeName, Uri? Destination); +} diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/native_ack_global_partitioning_cluster.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/native_ack_global_partitioning_cluster.cs new file mode 100644 index 000000000..ef640ddff --- /dev/null +++ b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/native_ack_global_partitioning_cluster.cs @@ -0,0 +1,231 @@ +using JasperFx.Core; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine.ComplianceTests.Partitioning; +using Wolverine.Configuration; +using Wolverine.Runtime; +using Wolverine.Tracking; +using Wolverine.Transports; +using Xunit; + +namespace Wolverine.RabbitMQ.Tests; + +/// +/// GH-3709. ProcessInParallelWithNativeAcks() on a global partitioned topology, exercised end to end +/// against a real broker on hosts with no message store at all. +/// +/// +/// The guarantee: no two messages sharing a group id execute concurrently. Within a node the +/// sequential lane inside the slot's own receiver enforces it; across the cluster the exclusive slot listener +/// enforces it, because exactly one node consumes a given slot. Ordering is per-slot best effort, not +/// per-group guaranteed -- redelivery or requeue may reorder, and the ordering unit is the slot rather than +/// the group, so two groups hashing to the same slot serialize against each other. +/// +/// Why slot ownership is assigned statically here. Wolverine's dynamic one-consumer-per-slot +/// assignment is ExclusiveListenerFamily, which runs under NodeAgentController -- and +/// WolverineRuntime.startAgentsAsync returns early when Storage is NullMessageStore, so a host +/// with no message store never builds a node agent controller and never assigns an exclusive listener. On a +/// storeless host the durability mode therefore has to be Solo, where Endpoint.ShouldAutoStartAsListener +/// starts every listener on every node. Slot ownership in a storage-free cluster is consequently +/// a deployment decision rather than something Wolverine negotiates, and this fixture makes it by stopping the +/// unowned slot listeners on each node. The dynamic-assignment and failover half of the story needs a store +/// for node coordination and lives in . +/// +public class native_ack_global_partitioning_cluster : IAsyncLifetime +{ + private readonly List _hosts = []; + private readonly List<(IHost Host, int[] Owned)> _ownership = []; + + public ValueTask InitializeAsync() + { + NativeAckPartitionedProcessing.Ledger.Clear(); + NativeAckPartitionedProcessing.Dwell = 50.Milliseconds(); + return ValueTask.CompletedTask; + } + + public async ValueTask DisposeAsync() + { + foreach (var host in _hosts.ToArray()) + { + try + { + await host.StopAsync(); + host.Dispose(); + } + catch (Exception) + { + // Nothing useful to do about a host that will not shut down cleanly during teardown + } + } + + _hosts.Clear(); + _ownership.Clear(); + NativeAckPartitionedProcessing.Ledger.Clear(); + } + + /// + /// A host with no message store whatsoever -- no Marten, no Postgres, no inbox. + /// are the zero-based slot indexes this node consumes; every other slot is send-only here. + /// + /// + /// Ownership is applied by stopping the unowned listeners after startup rather than by clearing + /// IsListener in configuration, for two reasons. It has to be after Compile: every + /// ListenerConfiguration carries a delayed e.IsListener = true that runs during + /// Endpoint.Compile(), after endpoint policies, so an IsListener = false set while + /// configuring is simply overwritten. And a stopped-and-drained listening agent is exactly the state + /// ExclusiveListenerAgent leaves behind on a node that is not assigned the slot -- so this + /// reproduces real slot ownership rather than approximating it. + /// + private async Task startStorelessHostAsync(string nodeName, string baseName, int slotCount, + params int[] ownedSlots) + { + var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + // Storeless hosts have no cluster coordination available, so Solo is the only workable mode. + opts.Durability.Mode = DurabilityMode.Solo; + + opts.UseRabbitMq("host=localhost;port=5672").AutoProvision().AutoPurgeOnStartup(); + + opts.UseNativeAckLetters(nodeName, + topology => + { + topology.ProcessInParallelWithNativeAcks(); + topology.UseShardedRabbitQueues(baseName, slotCount); + }); + }).StartAsync(); + + _hosts.Add(host); + _ownership.Add((host, ownedSlots)); + + var runtime = host.GetRuntime(); + for (var i = 0; i < slotCount; i++) + { + if (ownedSlots.Contains(i)) continue; + + var endpoint = runtime.Endpoints.EndpointFor(slotUri(baseName, i))!; + await runtime.Endpoints.StopListenerAsync(endpoint, CancellationToken.None); + } + + assertOwnsExactly(host, baseName, slotCount, ownedSlots); + + return host; + } + + private static Uri slotUri(string baseName, int slotIndex) => new($"rabbitmq://queue/{baseName}{slotIndex + 1}"); + + /// + /// One consumer per slot is the cluster-wide half of the guarantee, so it is asserted rather than assumed -- + /// both right after ownership is applied and again at the end of the run. + /// + private void assertOwnershipStillHolds(string baseName, int slotCount) + { + foreach (var (host, owned) in _ownership) + { + assertOwnsExactly(host, baseName, slotCount, owned); + } + } + + private static void assertOwnsExactly(IHost host, string baseName, int slotCount, int[] ownedSlots) + { + var runtime = host.GetRuntime(); + + for (var i = 0; i < slotCount; i++) + { + var status = runtime.Endpoints.FindListenerCircuit(slotUri(baseName, i))?.Status; + var expected = ownedSlots.Contains(i) ? ListeningStatus.Accepting : ListeningStatus.Stopped; + + status.ShouldBe(expected, + $"{runtime.Options.ServiceName} had slot {baseName}{i + 1} in status {status}, expected {expected}"); + } + } + + private static void assertSlotsAreNativeAckWithNoCompanionQueue(IHost host, string baseName, int slotCount) + { + var runtime = host.GetRuntime(); + + for (var i = 1; i <= slotCount; i++) + { + var endpoint = runtime.Endpoints.EndpointFor(new Uri($"rabbitmq://queue/{baseName}{i}")) + .ShouldNotBeNull($"No endpoint was built for slot {baseName}{i}"); + + endpoint.Mode.ShouldBe(EndpointMode.NativeAck); + endpoint.ListenerScope.ShouldBe(ListenerScope.Exclusive); + } + + // No companion local queues means no bridge and no durable receiver on the path. + runtime.Endpoints.ActiveSendingAgents() + .Select(x => x.Destination) + .Any(x => x.Scheme == "local" && x.Host.StartsWith($"global-{baseName}")) + .ShouldBeFalse("A native-ack topology must not create companion local queues"); + } + + /// + /// The cluster-wide statement of the guarantee: three storage-free nodes, six slots split between them, + /// and no group id ever executing in two places at once anywhere in the cluster. + /// + [Fact] + public async Task no_two_messages_of_a_group_execute_concurrently_across_a_storage_free_cluster() + { + const string baseName = "naclust"; + const int slotCount = 6; + + var node1 = await startStorelessHostAsync("NativeAckNode1", baseName, slotCount, 0, 1); + var node2 = await startStorelessHostAsync("NativeAckNode2", baseName, slotCount, 2, 3); + var node3 = await startStorelessHostAsync("NativeAckNode3", baseName, slotCount, 4, 5); + + assertSlotsAreNativeAckWithNoCompanionQueue(node1, baseName, slotCount); + + var published = await NativeAckPartitionedProcessing.PumpOutLettersAsync( + [node1.MessageBus, node2.MessageBus, node3.MessageBus], groupCount: 24, messagesPerGroup: 4); + + (await NativeAckPartitionedProcessing.WaitForCompletionAsync(published, 90.Seconds())) + .ShouldBeTrue("Not every published letter was handled inside the timeout"); + + // Nothing restarted a slot listener behind our back, so "exactly one consumer per slot" really did + // hold for the whole run rather than just at the start of it. + assertOwnershipStillHolds(baseName, slotCount); + + NativeAckPartitionedProcessing.AssertNoIntraGroupConcurrency(); + NativeAckPartitionedProcessing.AssertEveryLetterWasHandled(published); + NativeAckPartitionedProcessing.AssertGroupsNeverStraddleSlots(); + NativeAckPartitionedProcessing.AssertEverySlotWasUsed(slotCount); + + // Every node has to have done real work, otherwise "cluster-wide" is a claim about one node. + NativeAckPartitionedProcessing.Ledger.Handled.Select(x => x.NodeName).Distinct().OrderBy(x => x) + .ShouldBe(["NativeAckNode1", "NativeAckNode2", "NativeAckNode3"]); + } + + /// + /// The local shortcut hands a message straight to the companion local queue when the publishing node + /// already owns the target slot. A native-ack topology has no companion queue, and the broker delivery + /// is the durability story, so the shortcut is disabled: every send goes through the broker even + /// when this very node is the exclusive consumer of the slot it hashes to. + /// + [Fact] + public async Task sends_go_through_the_broker_even_when_this_node_owns_every_slot() + { + const string baseName = "nashortcut"; + const int slotCount = 3; + + var host = await startStorelessHostAsync("NativeAckSoleNode", baseName, slotCount, 0, 1, 2); + + var published = await NativeAckPartitionedProcessing.PumpOutLettersAsync( + [host.MessageBus], groupCount: 12, messagesPerGroup: 3); + + (await NativeAckPartitionedProcessing.WaitForCompletionAsync(published, 60.Seconds())) + .ShouldBeTrue("Not every published letter was handled inside the timeout"); + + var destinations = NativeAckPartitionedProcessing.Ledger.Handled + .Select(x => x.Destination).Distinct().ToArray(); + + destinations.ShouldAllBe(x => x!.Scheme == "rabbitmq"); + + // The durable topology's tell-tale is local://global-N/ destinations. There must be none. + destinations.Any(x => x!.Scheme == "local") + .ShouldBeFalse("The local shortcut fired: " + destinations.Select(x => x!.ToString()).Join(", ")); + + NativeAckPartitionedProcessing.AssertNoIntraGroupConcurrency(); + NativeAckPartitionedProcessing.AssertEveryLetterWasHandled(published); + } +} diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/native_ack_global_partitioning_failover.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/native_ack_global_partitioning_failover.cs new file mode 100644 index 000000000..c409fddfd --- /dev/null +++ b/src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/native_ack_global_partitioning_failover.cs @@ -0,0 +1,392 @@ +using System.Collections.Concurrent; +using IntegrationTests; +using JasperFx.Core; +using JasperFx.Resources; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Npgsql; +using Shouldly; +using Weasel.Postgresql; +using Wolverine.ComplianceTests.Partitioning; +using Wolverine.Configuration; +using Wolverine.Postgresql; +using Wolverine.Runtime; +using Wolverine.Runtime.Agents; +using Wolverine.Tracking; +using Xunit; + +namespace Wolverine.RabbitMQ.Tests; + +/// +/// GH-3709. Slot failover is the one true cross-node concurrency hazard for a native-ack global partitioned +/// topology: when a slot moves nodes, the new owner may start pulling while the old owner still has an +/// in-flight handler for a group in that slot. ExclusiveListenerAgent is supposed to stop and +/// drain the listener before releasing it. This suite verifies that claim under the new mode rather +/// than assuming it. +/// +/// +/// The guarantee: no two messages sharing a group id execute concurrently -- within a node the +/// sequential lane enforces it, across the cluster the exclusive slot listener does. Ordering is per-slot +/// best effort, not per-group guaranteed; redelivery or requeue may reorder, and two groups hashing to the +/// same slot serialize against each other. +/// +/// Why there is a database here at all. The slots themselves stay storage-free -- they are +/// , so no envelope touches an inbox and no message ever hits the +/// database. Postgres is present only as the cluster's node/agent coordination store, because dynamic +/// one-consumer-per-slot assignment runs through NodeAgentController, which +/// WolverineRuntime.startAgentsAsync skips entirely when there is no message store. The genuinely +/// store-free deployment shape -- static slot ownership per node -- is covered by +/// . +/// +// Deliberately NOT in its own [Collection]. The assembly runs CollectionPerAssembly, so an explicit +// collection attribute would put this class in a SEPARATE collection -- which xUnit then runs in PARALLEL +// with the assembly collection, and native_ack_global_partitioning_cluster writes the same static +// NativeAckPartitionedProcessing.Ledger. That combination produced exactly the cross-class contamination +// you would expect. +public class native_ack_global_partitioning_failover : IAsyncLifetime +{ + private const string SchemaName = "native_ack_gp_failover"; + private const string BaseName = "nafail"; + private const int SlotCount = 4; + + private readonly List _hosts = []; + private readonly ITestOutputHelper _output; + + public native_ack_global_partitioning_failover(ITestOutputHelper output) + { + _output = output; + } + + public async ValueTask InitializeAsync() + { + NativeAckPartitionedProcessing.Ledger.Clear(); + + // Wide enough that a handler is genuinely still in flight when its slot is handed off. + NativeAckPartitionedProcessing.Dwell = 250.Milliseconds(); + + await using var conn = new NpgsqlConnection(Servers.PostgresConnectionString); + await conn.OpenAsync(); + await conn.DropSchemaAsync(SchemaName); + await conn.CloseAsync(); + } + + public async ValueTask DisposeAsync() + { + _hosts.Reverse(); + foreach (var host in _hosts.ToArray()) + { + try + { + await shutdownHostAsync(host); + } + catch (Exception) + { + // Nothing useful to do about a host that will not shut down cleanly during teardown + } + } + + _hosts.Clear(); + NativeAckPartitionedProcessing.Ledger.Clear(); + NativeAckPartitionedProcessing.Dwell = 50.Milliseconds(); + } + + private async Task startHostAsync(string nodeName) + { + var host = await Host.CreateDefaultBuilder().UseWolverine(opts => + { + opts.Durability.Mode = DurabilityMode.Balanced; + opts.Durability.HealthCheckPollingTime = 1.Seconds(); + opts.Durability.NodeReassignmentPollingTime = 1.Seconds(); + opts.Durability.CheckAssignmentPeriod = 1.Seconds(); + opts.Durability.StaleNodeTimeout = 3.Seconds(); + + opts.UseRabbitMq("host=localhost;port=5672").EnableWolverineControlQueues().AutoProvision(); + + // Node/agent coordination only -- see the class remarks. The slots never touch it. + opts.PersistMessagesWithPostgresql(Servers.PostgresConnectionString, SchemaName); + + opts.UseNativeAckLetters(nodeName, topology => + { + topology.ProcessInParallelWithNativeAcks(); + topology.UseShardedRabbitQueues(BaseName, SlotCount); + }); + + opts.Services.AddResourceSetupOnStartup(); + }).StartAsync(); + + _hosts.Add(host); + + return host; + } + + private async Task shutdownHostAsync(IHost host) + { + host.GetRuntime().Agents.DisableHealthChecks(); + await host.StopAsync(); + host.Dispose(); + _hosts.Remove(host); + } + + private static Uri slotAgentUri(int slotNumber) => + new($"{ExclusiveListenerFamily.SchemeName}://rabbitmq/{BaseName}{slotNumber}"); + + private IReadOnlyList slotAgentsOn(IHost host) + { + var running = host.RunningAgents().ToHashSet(); + return Enumerable.Range(1, SlotCount).Select(slotAgentUri).Where(running.Contains).ToArray(); + } + + /// + /// Every slot must be consumed by exactly one node -- never zero, never two. + /// + private bool everySlotOwnedExactlyOnce() + { + return Enumerable.Range(1, SlotCount) + .All(slot => _hosts.Count(h => slotAgentsOn(h).Contains(slotAgentUri(slot))) == 1); + } + + private async Task waitForFullSlotOwnershipAsync(TimeSpan timeout) + { + var stopwatch = System.Diagnostics.Stopwatch.StartNew(); + while (stopwatch.Elapsed < timeout) + { + if (everySlotOwnedExactlyOnce()) + { + // Has to hold across a health-check cycle to count as settled rather than mid-flap. + await Task.Delay(1500.Milliseconds()); + if (everySlotOwnedExactlyOnce()) return; + } + + await Task.Delay(250.Milliseconds()); + } + + var report = _hosts.Select(h => + $"{h.GetRuntime().Options.ServiceName}=[{slotAgentsOn(h).Select(x => x.ToString()).Join(", ")}]").Join("; "); + + throw new TimeoutException($"The {SlotCount} slot listeners never settled one-per-node. Saw: {report}"); + } + + private void assertSlotsAreNativeAck(IHost host) + { + var runtime = host.GetRuntime(); + + for (var i = 1; i <= SlotCount; i++) + { + var endpoint = runtime.Endpoints.EndpointFor(new Uri($"rabbitmq://queue/{BaseName}{i}")) + .ShouldNotBeNull(); + + endpoint.Mode.ShouldBe(EndpointMode.NativeAck); + endpoint.ListenerScope.ShouldBe(ListenerScope.Exclusive); + } + + _output.WriteLine($"{runtime.Options.ServiceName}: all {SlotCount} slots are NativeAck + Exclusive"); + } + + /// + /// Publish a steady stream of grouped letters until cancelled, so the cluster is genuinely mid-flight + /// when a node is taken away. + /// + private static Task<(Task Pump, ConcurrentQueue<(string GroupId, int Sequence)> Published)> startPumpAsync( + IMessageBus bus, int groupCount, CancellationToken token) + { + var published = new ConcurrentQueue<(string, int)>(); + var groups = Enumerable.Range(0, groupCount).Select(_ => Guid.NewGuid().ToString()).ToArray(); + + var pump = Task.Run(async () => + { + var sequence = 0; + while (!token.IsCancellationRequested) + { + foreach (var groupId in groups) + { + if (token.IsCancellationRequested) return; + + await bus.PublishAsync(new NativeAckLetter(groupId, sequence)); + published.Enqueue((groupId, sequence)); + } + + sequence++; + await Task.Delay(100.Milliseconds(), CancellationToken.None); + } + }, CancellationToken.None); + + return Task.FromResult((pump, published)); + } + + /// + /// The test the issue calls out as the one that matters: take the node owning a slot away mid-stream and + /// assert that the slot is reassigned, that processing continues, that no two messages sharing a group id + /// ever executed concurrently across the handoff, and that nothing was lost. + /// + [Fact] + public async Task no_intra_group_concurrency_when_a_slot_owner_leaves_mid_stream() + { + var leader = await startHostAsync("FailoverNode1"); + await startHostAsync("FailoverNode2"); + await startHostAsync("FailoverNode3"); + + (await leader.WaitUntilAssumesLeadershipAsync(30.Seconds())) + .ShouldBeTrue("The first host never assumed leadership"); + + assertSlotsAreNativeAck(leader); + + await waitForFullSlotOwnershipAsync(60.Seconds()); + + using var cts = new CancellationTokenSource(); + var (pump, published) = await startPumpAsync(leader.MessageBus(), groupCount: 16, cts.Token); + + // Let real work get under way before pulling a node out from under it. + await Task.Delay(3.Seconds(), TestContext.Current.CancellationToken); + + // Take away a non-leader node that is actually consuming slots, so the reassignment is a genuine + // slot handoff rather than a leadership election. + var candidates = _hosts.Skip(1).Where(h => slotAgentsOn(h).Any()).ToArray(); + candidates.ShouldNotBeEmpty("No non-leader node was consuming a slot, so there is no handoff to test"); + + var victim = candidates[0]; + var orphanedSlots = slotAgentsOn(victim); + var victimName = victim.GetRuntime().Options.ServiceName; + + _output.WriteLine($"Stopping {victimName}, which owns {orphanedSlots.Select(x => x.ToString()).Join(", ")}"); + orphanedSlots.ShouldNotBeEmpty(); + + await shutdownHostAsync(victim); + + await waitForFullSlotOwnershipAsync(90.Seconds()); + + foreach (var slot in orphanedSlots) + { + _hosts.Count(h => slotAgentsOn(h).Contains(slot)) + .ShouldBe(1, $"Slot {slot} did not land on exactly one survivor"); + } + + // Processing has to keep going on the survivors, not merely resume owning the slots. + var handledBeforeSettling = NativeAckPartitionedProcessing.Ledger.Handled.Count; + await Task.Delay(3.Seconds(), TestContext.Current.CancellationToken); + NativeAckPartitionedProcessing.Ledger.Handled.Count.ShouldBeGreaterThan(handledBeforeSettling, + "Nothing was processed after the slot handoff"); + + await cts.CancelAsync(); + await pump; + + var everything = published.ToArray(); + (await NativeAckPartitionedProcessing.WaitForCompletionAsync(everything, 120.Seconds())) + .ShouldBeTrue("Not every published letter was handled after the failover"); + + _output.WriteLine( + $"{everything.Length} published, {NativeAckPartitionedProcessing.Ledger.Handled.Count} handled " + + $"(duplicates are legal in this mode)"); + + // The claim under test: exclusive-listener handoff drains in-flight work before releasing the slot, + // so the new owner never overlaps the old owner on a shared group id. + NativeAckPartitionedProcessing.AssertNoIntraGroupConcurrency(); + + // At-least-once completeness -- the broker, not an inbox, is what makes this true. + NativeAckPartitionedProcessing.AssertEveryLetterWasHandled(everything); + + // A group hashes to one slot and stays there; a handoff changes which node consumes that slot, never + // which slot the group belongs to. + NativeAckPartitionedProcessing.AssertGroupsNeverStraddleSlots(); + + // The handoff really did move work between nodes, otherwise nothing above was tested. + var nodes = NativeAckPartitionedProcessing.Ledger.Handled.Select(x => x.NodeName).Distinct().ToArray(); + nodes.ShouldContain(victimName, "The node that was stopped never handled anything before it left"); + nodes.Length.ShouldBeGreaterThan(1); + + // And specifically: the orphaned slots kept being drained, on a survivor. A survivor can only ever + // handle one of those slots' messages after the handoff, because before it the victim owned them + // exclusively -- so this is the assertion that the reassigned slots did real work post-failover. + var orphanedQueues = orphanedSlots.Select(x => x.Segments.Last()).ToHashSet(); + NativeAckPartitionedProcessing.Ledger.Handled + .Where(x => x.NodeName != victimName && orphanedQueues.Contains(x.Destination!.Segments.Last())) + .ShouldNotBeEmpty("No survivor ever processed a message from one of the reassigned slots"); + } + + /// + /// The sharper version of the same hazard: move a slot between two nodes that are both still alive, + /// mid-stream. Nothing here is helped along by a dying process -- the old owner keeps running, so the only + /// thing standing between the new owner's first pull and the old owner's in-flight handler is + /// ExclusiveListenerAgent.StopAsync draining the listener before it releases the slot. + /// + [Fact] + public async Task no_intra_group_concurrency_when_a_live_slot_handoff_moves_the_slot() + { + var leader = await startHostAsync("HandoffNode1"); + await startHostAsync("HandoffNode2"); + await startHostAsync("HandoffNode3"); + + (await leader.WaitUntilAssumesLeadershipAsync(30.Seconds())) + .ShouldBeTrue("The first host never assumed leadership"); + + await waitForFullSlotOwnershipAsync(60.Seconds()); + + using var cts = new CancellationTokenSource(); + var (pump, published) = await startPumpAsync(leader.MessageBus(), groupCount: 16, cts.Token); + + await Task.Delay(3.Seconds(), TestContext.Current.CancellationToken); + + var movedSlot = slotAgentUri(1); + + var owners = _hosts.Where(h => slotAgentsOn(h).Contains(movedSlot)).ToArray(); + owners.Length.ShouldBe(1, + $"{movedSlot} was owned by {owners.Length} nodes rather than exactly one when the handoff started"); + + var originalOwner = owners[0]; + var target = _hosts.First(h => !ReferenceEquals(h, originalOwner)); + + var originalOwnerName = originalOwner.GetRuntime().Options.ServiceName; + var targetNumber = target.GetRuntime().DurabilitySettings.AssignedNodeNumber; + + _output.WriteLine( + $"Moving {movedSlot} from {originalOwnerName} to {target.GetRuntime().Options.ServiceName} " + + $"(node {targetNumber}) while both are live"); + + var restrictions = new AgentRestrictions(); + restrictions.PinAgent(movedSlot, targetNumber); + await leader.GetRuntime().Agents.ApplyRestrictionsAsync(restrictions, CancellationToken.None); + + await waitForSlotOwnerAsync(movedSlot, target, 60.Seconds()); + await waitForFullSlotOwnershipAsync(60.Seconds()); + + // Keep the stream running across the handoff so the new owner has real work waiting for it. + await Task.Delay(3.Seconds(), TestContext.Current.CancellationToken); + + await cts.CancelAsync(); + await pump; + + var everything = published.ToArray(); + (await NativeAckPartitionedProcessing.WaitForCompletionAsync(everything, 120.Seconds())) + .ShouldBeTrue("Not every published letter was handled after the live handoff"); + + _output.WriteLine( + $"{everything.Length} published, {NativeAckPartitionedProcessing.Ledger.Handled.Count} handled"); + + NativeAckPartitionedProcessing.AssertNoIntraGroupConcurrency(); + NativeAckPartitionedProcessing.AssertEveryLetterWasHandled(everything); + NativeAckPartitionedProcessing.AssertGroupsNeverStraddleSlots(); + + // The moved slot has to have been worked by both nodes, or the handoff window was never entered. + var workersOnMovedSlot = NativeAckPartitionedProcessing.Ledger.Handled + .Where(x => x.Destination!.Segments.Last() == movedSlot.Segments.Last()) + .Select(x => x.NodeName) + .Distinct() + .ToArray(); + + workersOnMovedSlot.ShouldContain(originalOwnerName); + workersOnMovedSlot.Length.ShouldBeGreaterThan(1, + $"Slot {movedSlot} was only ever processed by {workersOnMovedSlot.Join(", ")}, so no handoff happened"); + } + + private async Task waitForSlotOwnerAsync(Uri slotAgent, IHost expected, TimeSpan timeout) + { + var stopwatch = System.Diagnostics.Stopwatch.StartNew(); + while (stopwatch.Elapsed < timeout) + { + if (slotAgentsOn(expected).Contains(slotAgent)) return; + await Task.Delay(250.Milliseconds()); + } + + throw new TimeoutException( + $"{slotAgent} never moved to {expected.GetRuntime().Options.ServiceName}"); + } +} diff --git a/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs index 903af3102..ab8b1216e 100644 --- a/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs +++ b/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs @@ -13,7 +13,7 @@ namespace Wolverine.Runtime.WorkerQueues; internal class BufferedReceiver : ILocalQueue, IChannelCallback, ISupportNativeScheduling, ISupportDeadLetterQueue, - IFaultTrackingReceiver + IFaultTrackingReceiver, ILatchedReceiver { private readonly RetryBlock _completeBlock; diff --git a/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs index ad846db81..ecf743482 100644 --- a/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs +++ b/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs @@ -13,7 +13,7 @@ namespace Wolverine.Runtime.WorkerQueues; public class DurableReceiver : ILocalQueue, IChannelCallback, ISupportNativeScheduling, ISupportDeadLetterQueue, - IAsyncDisposable, IFaultTrackingReceiver + IAsyncDisposable, IFaultTrackingReceiver, ILatchedReceiver { private readonly RetryBlock _completeBlock; diff --git a/src/Wolverine/Runtime/WorkerQueues/ILatchedReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/ILatchedReceiver.cs new file mode 100644 index 000000000..2e6fbaacd --- /dev/null +++ b/src/Wolverine/Runtime/WorkerQueues/ILatchedReceiver.cs @@ -0,0 +1,24 @@ +namespace Wolverine.Runtime.WorkerQueues; + +/// +/// A receiver that can be latched -- told to stop executing anything further -- ahead of +/// . +/// +/// +/// GH-3709. This exists so ListeningAgent.LatchReceiver() is a single type test rather than an +/// if/else chain naming each receiver implementation. The chain had already silently missed +/// when GH-3708 added it, and the failure was invisible rather than loud: +/// an unlatched receiver's DrainAsync returns immediately instead of waiting for in-flight handlers, +/// so a stop-and-drain closed the transport channel underneath still-running work. Every unsettled delivery +/// was then requeued and redelivered while the original was still executing -- which on an exclusive +/// listener handoff means the new owner runs a message concurrently with the old owner, breaking the +/// no-two-messages-of-a-group-at-once guarantee that partitioned processing exists to provide. +/// +internal interface ILatchedReceiver +{ + /// + /// Stop executing further messages. Does not wait for in-flight work -- that is DrainAsync's job, + /// and it only waits when the receiver has been latched first. + /// + void Latch(); +} diff --git a/src/Wolverine/Runtime/WorkerQueues/InlineReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/InlineReceiver.cs index f2076e610..52fb0b217 100644 --- a/src/Wolverine/Runtime/WorkerQueues/InlineReceiver.cs +++ b/src/Wolverine/Runtime/WorkerQueues/InlineReceiver.cs @@ -7,7 +7,7 @@ namespace Wolverine.Runtime.WorkerQueues; -internal class InlineReceiver : IReceiver +internal class InlineReceiver : IReceiver, ILatchedReceiver { private readonly ILogger _logger; private readonly Endpoint _endpoint; diff --git a/src/Wolverine/Runtime/WorkerQueues/NativeAckReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/NativeAckReceiver.cs index b4ab2c902..c1410a938 100644 --- a/src/Wolverine/Runtime/WorkerQueues/NativeAckReceiver.cs +++ b/src/Wolverine/Runtime/WorkerQueues/NativeAckReceiver.cs @@ -34,7 +34,7 @@ namespace Wolverine.Runtime.WorkerQueues; /// this mode exists to provide. /// /// -internal class NativeAckReceiver : IReceiver, IFaultTrackingReceiver +internal class NativeAckReceiver : IReceiver, IFaultTrackingReceiver, ILatchedReceiver { private readonly RetryBlock _completeBlock; private readonly RetryBlock _deferBlock; diff --git a/src/Wolverine/Transports/ListeningAgent.cs b/src/Wolverine/Transports/ListeningAgent.cs index 8d8146e93..88ac8d3de 100644 --- a/src/Wolverine/Transports/ListeningAgent.cs +++ b/src/Wolverine/Transports/ListeningAgent.cs @@ -247,18 +247,16 @@ public async Task EnqueueDirectlyAsync(IEnumerable envelopes) /// public void LatchReceiver() { + // GH-3709. Deliberately a single ILatchedReceiver test rather than an if/else chain naming each + // receiver type. As a chain this silently missed NativeAckReceiver when GH-3708 added it, and an + // unlatched receiver's DrainAsync returns immediately instead of waiting for in-flight handlers -- + // so a stop-and-drain closed the transport channel underneath running work, the unsettled deliveries + // were requeued, and on an exclusive listener handoff the new owner re-ran them concurrently with + // the old owner. That is exactly the intra-group concurrency the partitioned modes forbid. var actual = _receiver is ReceiverWithRules rwr ? rwr.Inner : _receiver; - if (actual is DurableReceiver dr) + if (actual is ILatchedReceiver latched) { - dr.Latch(); - } - else if (actual is BufferedReceiver br) - { - br.Latch(); - } - else if (actual is InlineReceiver ir) - { - ir.Latch(); + latched.Latch(); } } diff --git a/src/Wolverine/Transports/Sending/SendingAgent.cs b/src/Wolverine/Transports/Sending/SendingAgent.cs index 82f28990a..10c95d076 100644 --- a/src/Wolverine/Transports/Sending/SendingAgent.cs +++ b/src/Wolverine/Transports/Sending/SendingAgent.cs @@ -149,19 +149,36 @@ async Task ISenderCircuit.ResumeAsync(CancellationToken cancellationToken) public async ValueTask EnqueueOutgoingAsync(Envelope envelope) { setDefaults(envelope); + + var pooled = envelope.FromPool; + if (pooled) _messageLogger.Sent(envelope); + await _sending.PostAsync(envelope); + _lastMessageSentAt = DateTimeOffset.UtcNow; - _messageLogger.Sent(envelope); + if (!pooled) _messageLogger.Sent(envelope); } public async ValueTask StoreAndForwardAsync(Envelope envelope) { setDefaults(envelope); + // GH-3709. Handing a POOLED envelope to the sending block publishes it to another thread that may + // send it, succeed, and return it to the pool -- Envelope.Reset(), which nulls Destination and + // MessageType -- before this frame resumes. Reading it afterwards for metrics is therefore a data + // race, and it surfaced as an intermittent NullReferenceException out of Envelope.ToMetricsHeaders() + // (Destination non-null at its own guard, null one line later) under concurrent publishing. It was + // found through EndpointMode.NativeAck, which mapped to BufferedSendingAgent at the time; GH-4061 has + // since moved NativeAck onto the inline agent, so the remaining exposure is any BufferedInMemory + // endpoint. Non-pooled envelopes -- every durable send, and anything published inside a tracking + // session -- keep the original ordering, so a store that throws still reports no send. + var pooled = envelope.FromPool; + if (pooled) _messageLogger.Sent(envelope); + await storeAndForwardAsync(envelope); _lastMessageSentAt = DateTimeOffset.UtcNow; - _messageLogger.Sent(envelope); + if (!pooled) _messageLogger.Sent(envelope); } public bool SupportsNativeScheduledSend => _sender.SupportsNativeScheduledSend;