fix(local): buffered local queues no longer drop cascaded messages (GH-3287)

Directory.Packages.props

@@ -33,13 +33,13 @@
     <PackageVersion Include="Grpc.StatusProto" Version="2.76.0" />
     <PackageVersion Include="Grpc.Tools" Version="2.76.0" />
     <PackageVersion Include="HtmlTags" Version="9.0.0" />
-    <PackageVersion Include="JasperFx" Version="2.16.0" />
-    <PackageVersion Include="JasperFx.Events" Version="2.16.0" />
-    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.16.0" />
+    <PackageVersion Include="JasperFx" Version="2.19.1" />
+    <PackageVersion Include="JasperFx.Events" Version="2.19.1" />
+    <PackageVersion Include="JasperFx.Events.SourceGenerator" Version="2.19.1" />
     <!-- RuntimeCompiler is on its own 5.x line (the Roslyn compiler package) — not the 2.1.x
          family; it stays at 5.0.0. -->
     <PackageVersion Include="JasperFx.RuntimeCompiler" Version="5.0.0" />
-    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.16.0" />
+    <PackageVersion Include="JasperFx.SourceGenerator" Version="2.19.1" />
     <PackageVersion Include="Lamar.Microsoft.DependencyInjection" Version="16.0.0" />
     <PackageVersion Include="Marten" Version="9.11.0" />
     <PackageVersion Include="Microsoft.Data.SqlClient" Version="6.1.3" />

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;
+
+/// <summary>
+/// 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 <see cref="MessageCount"/> Step2 messages
+/// to a buffered local queue. Handler2 consumes them. Because the underlying JasperFx Block&lt;T&gt; 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.
+/// </summary>
+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<StartPipeline>()
+                    .Message<Step2Message>()
+                    .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<object> 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<int, byte> _seen = new();
+    private static int _count;
+    private static long _lastReceivedTicks;
+
+    public static void Reset()
+    {
+        _seen = new ConcurrentDictionary<int, byte>();
+        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;
+
+    /// <summary>
+    /// Waits until either the expected number of messages have been received, or the handler has been
+    /// idle (no new messages) for <paramref name="quietPeriod"/>, or the overall timeout elapses.
+    /// Returns the final received count.
+    /// </summary>
+    public static async Task<int> 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;
+    }
+}

src/Wolverine/Runtime/Partitioning/ShardedExecutionBlock.cs

@@ -11,14 +11,19 @@ internal class ShardedExecutionBlock : BlockBase<Envelope>
     private readonly Block<Envelope>[] _slots;
 
     public ShardedExecutionBlock(int numberOfSlots, MessagePartitioningRules rules, Func<Envelope, CancellationToken, Task> processAsync)
+        : this(numberOfSlots, rules, Block<Envelope>.DefaultBoundedCapacity, processAsync)
+    {
+    }
+
+    public ShardedExecutionBlock(int numberOfSlots, MessagePartitioningRules rules, int boundedCapacity, Func<Envelope, CancellationToken, Task> processAsync)
     {
         _numberOfSlots = numberOfSlots;
         _rules = rules;
 
         _slots = new Block<Envelope>[_numberOfSlots];
         for (int i = 0; i < _numberOfSlots; i++)
         {
-            _slots[i] = new Block<Envelope>(processAsync);
+            _slots[i] = new Block<Envelope>(1, boundedCapacity, processAsync);
         }
     }
 

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<Envelope>.DefaultBoundedCapacity)
+    {
+    }
+
+    /// <param name="boundedCapacity">
+    /// 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 <see cref="Block{T}.Unbounded"/> — a bounded buffer would deadlock a handler that
+    /// enqueues to its own queue faster than it drains (GH-3287).
+    /// </param>
+    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<Envelope>(endpoint.MaxDegreeOfParallelism, executeAsync)
-            : new ShardedExecutionBlock((int)endpoint.GroupShardingSlotNumber, runtime.Options.MessagePartitioning, executeAsync).DeserializeFirst(pipeline, runtime, this);
+        _receivingBlock = endpoint.GroupShardingSlotNumber == null
+            ? new Block<Envelope>(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))
         {

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<Envelope>.Unbounded)
     {
         _messageTracker = runtime.MessageTracking;
         _runtime = runtime;