diff --git a/Directory.Packages.props b/Directory.Packages.props index 9cd783e70..2f028445c 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -33,13 +33,13 @@ - - - + + + - + diff --git a/src/Testing/SlowTests/dropped_messages_on_full_local_queue.cs b/src/Testing/SlowTests/dropped_messages_on_full_local_queue.cs new file mode 100644 index 000000000..a27ae071f --- /dev/null +++ b/src/Testing/SlowTests/dropped_messages_on_full_local_queue.cs @@ -0,0 +1,150 @@ +using System.Collections.Concurrent; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine; +using Wolverine.Attributes; +using Xunit; +using Xunit.Abstractions; + +namespace SlowTests; + +/// +/// Reproduction harness for GH-3287: Wolverine silently drops cascaded messages once more than +/// 10,000 are queued into a buffered (in-memory) local queue faster than the handlers can drain it. +/// +/// Handler1 handles a single StartPipeline message and cascades Step2 messages +/// to a buffered local queue. Handler2 consumes them. Because the underlying JasperFx Block<T> uses a +/// bounded channel of 10,000 and the synchronous enqueue path (BufferedReceiver.Enqueue -> Block.Post -> +/// Channel.Writer.TryWrite) silently fails when the channel is full, everything past ~10,000 is lost. +/// +public class dropped_messages_on_full_local_queue +{ + public const int MessageCount = 20_000; + + private readonly ITestOutputHelper _output; + + public dropped_messages_on_full_local_queue(ITestOutputHelper output) => _output = output; + + // parallelism 5 mirrors the customer's config; parallelism 1 is the harder case — the single reader + // that runs Handler1 is the only thing that can drain the queue, so naive back pressure (await for + // room on a full bounded channel) would deadlock. The unbounded buffered local queue must handle both. + [Theory] + [InlineData(5)] + [InlineData(1)] + public async Task all_cascaded_messages_reach_the_second_handler(int maxParallel) + { + DropTestTracker.Reset(); + + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.IncludeAssembly(typeof(dropped_messages_on_full_local_queue).Assembly); + + // Route both the trigger and the cascaded messages to one buffered local queue, + // throttled to a small worker count so the producer easily outruns the consumer + // and the 10k channel fills — exactly the customer's configuration. + opts.Publish(x => x + .Message() + .Message() + .ToLocalQueue("pipeline") + .MaximumParallelMessages(maxParallel)); + + opts.Policies.DisableConventionalLocalRouting(); + }) + .StartAsync(); + + var bus = host.MessageBus(); + await bus.PublishAsync(new StartPipeline()); + + // Wait until Handler2 goes quiet (no new messages for a stretch) or an overall timeout. + // On the buggy code the count plateaus at ~10k and stays there; a correct implementation + // reaches MessageCount. + var received = await DropTestTracker.WaitForQuiescenceAsync( + expected: MessageCount, + quietPeriod: TimeSpan.FromSeconds(5), + overallTimeout: TimeSpan.FromMinutes(2)); + + _output.WriteLine($"Handler2 received {received:N0} of {MessageCount:N0} messages " + + $"({DropTestTracker.UniqueCount:N0} unique ids)."); + + DropTestTracker.UniqueCount.ShouldBe(MessageCount); + received.ShouldBe(MessageCount); + } +} + +public record StartPipeline; + +public record Step2Message(int Id); + +[WolverineHandler] +public class DropTestHandler1 +{ + public IEnumerable Handle(StartPipeline message) + { + for (var id = 0; id < dropped_messages_on_full_local_queue.MessageCount; id++) + { + yield return new Step2Message(id); + } + } +} + +[WolverineHandler] +public class DropTestHandler2 +{ + public void Handle(Step2Message message) + { + DropTestTracker.Record(message.Id); + } +} + +public static class DropTestTracker +{ + private static ConcurrentDictionary _seen = new(); + private static int _count; + private static long _lastReceivedTicks; + + public static void Reset() + { + _seen = new ConcurrentDictionary(); + Interlocked.Exchange(ref _count, 0); + Interlocked.Exchange(ref _lastReceivedTicks, DateTime.UtcNow.Ticks); + } + + public static void Record(int id) + { + _seen.TryAdd(id, 0); + Interlocked.Increment(ref _count); + Interlocked.Exchange(ref _lastReceivedTicks, DateTime.UtcNow.Ticks); + } + + public static int Count => Volatile.Read(ref _count); + + public static int UniqueCount => _seen.Count; + + /// + /// Waits until either the expected number of messages have been received, or the handler has been + /// idle (no new messages) for , or the overall timeout elapses. + /// Returns the final received count. + /// + public static async Task WaitForQuiescenceAsync(int expected, TimeSpan quietPeriod, TimeSpan overallTimeout) + { + var deadline = DateTime.UtcNow + overallTimeout; + while (DateTime.UtcNow < deadline) + { + if (Count >= expected) + { + return Count; + } + + var idle = DateTime.UtcNow - new DateTime(Volatile.Read(ref _lastReceivedTicks), DateTimeKind.Utc); + if (idle > quietPeriod && Count > 0) + { + return Count; + } + + await Task.Delay(250); + } + + return Count; + } +} diff --git a/src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs b/src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs index 8c92d601b..91859e6bd 100644 --- a/src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs +++ b/src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs @@ -11,6 +11,11 @@ internal class ShardedExecutionBlock : BlockBase private readonly Block[] _slots; public ShardedExecutionBlock(int numberOfSlots, MessagePartitioningRules rules, Func processAsync) + : this(numberOfSlots, rules, Block.DefaultBoundedCapacity, processAsync) + { + } + + public ShardedExecutionBlock(int numberOfSlots, MessagePartitioningRules rules, int boundedCapacity, Func processAsync) { _numberOfSlots = numberOfSlots; _rules = rules; @@ -18,7 +23,7 @@ public ShardedExecutionBlock(int numberOfSlots, MessagePartitioningRules rules, _slots = new Block[_numberOfSlots]; for (int i = 0; i < _numberOfSlots; i++) { - _slots[i] = new Block(processAsync); + _slots[i] = new Block(1, boundedCapacity, processAsync); } } diff --git a/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs index 90ad891fc..3ebb4d9f1 100644 --- a/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs +++ b/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs @@ -28,6 +28,17 @@ internal class BufferedReceiver : ILocalQueue, IChannelCallback, ISupportNativeS private bool _latched; public BufferedReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandlerPipeline pipeline) + : this(endpoint, runtime, pipeline, Block.DefaultBoundedCapacity) + { + } + + /// + /// The in-memory buffer size before writers are back-pressured. Broker-backed buffered receivers keep + /// the bounded default so a fast broker can't flood memory. Local queues that may cascade onto + /// themselves pass — a bounded buffer would deadlock a handler that + /// enqueues to its own queue faster than it drains (GH-3287). + /// + protected BufferedReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandlerPipeline pipeline, int boundedCapacity) { _endpoint = endpoint; _runtime = runtime; @@ -48,9 +59,9 @@ public BufferedReceiver(Endpoint endpoint, IWolverineRuntime runtime, IHandlerPi (env, _) => env.Listener is { } l ? l.CompleteAsync(env).AsTask() : Task.CompletedTask, runtime.Logger, runtime.Cancellation); - _receivingBlock = endpoint.GroupShardingSlotNumber == null - ? new Block(endpoint.MaxDegreeOfParallelism, executeAsync) - : new ShardedExecutionBlock((int)endpoint.GroupShardingSlotNumber, runtime.Options.MessagePartitioning, executeAsync).DeserializeFirst(pipeline, runtime, this); + _receivingBlock = endpoint.GroupShardingSlotNumber == null + ? new Block(endpoint.MaxDegreeOfParallelism, boundedCapacity, executeAsync) + : new ShardedExecutionBlock((int)endpoint.GroupShardingSlotNumber, runtime.Options.MessagePartitioning, boundedCapacity, executeAsync).DeserializeFirst(pipeline, runtime, this); if (endpoint.TryBuildDeadLetterSender(runtime, out var dlq)) { diff --git a/src/Wolverine/Transports/Local/BufferedLocalQueue.cs b/src/Wolverine/Transports/Local/BufferedLocalQueue.cs index 59e5c06dd..8d909be7c 100644 --- a/src/Wolverine/Transports/Local/BufferedLocalQueue.cs +++ b/src/Wolverine/Transports/Local/BufferedLocalQueue.cs @@ -1,3 +1,4 @@ +using JasperFx.Blocks; using Wolverine.Configuration; using Wolverine.Logging; using Wolverine.Persistence.Durability; @@ -12,7 +13,12 @@ internal class BufferedLocalQueue : BufferedReceiver, ISendingAgent, IListenerCi private readonly IMessageTracker _messageTracker; private readonly IWolverineRuntime _runtime; - public BufferedLocalQueue(Endpoint endpoint, IWolverineRuntime runtime) : base(endpoint, runtime, new HandlerPipeline((WolverineRuntime)runtime, (IExecutorFactory)runtime, endpoint)) + // Unbounded (GH-3287): a buffered local queue is routinely a handler's own cascade target — Handler A + // running on this queue enqueues thousands of messages back onto it. A bounded, back-pressuring buffer + // would either drop those (the original bug) or deadlock (the handler blocks on a full queue that only + // it can drain). Local queues are in-process, so we favor unbounded growth over losing messages. + public BufferedLocalQueue(Endpoint endpoint, IWolverineRuntime runtime) : base(endpoint, runtime, + new HandlerPipeline((WolverineRuntime)runtime, (IExecutorFactory)runtime, endpoint), Block.Unbounded) { _messageTracker = runtime.MessageTracking; _runtime = runtime;