From 7905edbaae42f5c0d76efdc16884dbf4ef316968 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Mon, 6 Jul 2026 06:55:37 -0500 Subject: [PATCH] fix(blocks): Post no longer silently drops items on a full bounded channel Block.Post used a non-blocking TryWrite against a channel bounded at 10,000 and silently dropped every item once the channel filled -- only a Debug.WriteLine marked the loss. Callers that post from outside the reader (Wolverine's buffered local queues, Marten's ProjectionUpdateBatch) lost data under load with no error. Reported as wolverine GH-3287, where a handler cascading 20k messages to a local queue only delivered ~10k. - Post now block-waits for capacity via WriteAsync (BoundedChannelFullMode.Wait) instead of dropping, so writers are back-pressured rather than losing items. - Add a configurable boundedCapacity plus an Unbounded option for blocks that must never block or drop on write. - RetryBlock uses an unbounded internal block: executeAsync re-enqueues failed items onto its own block from within its processing action, which would deadlock against back pressure on a full bounded channel. Bumps to 2.19.1. Co-Authored-By: Claude Opus 4.8 (1M context) --- Directory.Build.props | 2 +- src/CoreTests/Blocks/InMemoryQueueTests.cs | 68 +++++++++++++++++- src/JasperFx/Blocks/Block.cs | 81 ++++++++++++++++++---- src/JasperFx/Blocks/RetryBlock.cs | 5 +- 4 files changed, 141 insertions(+), 15 deletions(-) diff --git a/Directory.Build.props b/Directory.Build.props index 120750c5..caa2bc34 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -1,7 +1,7 @@ - 2.19.0 + 2.19.1 13 1570;1571;1572;1573;1574;1587;1591;1701;1702;1711;1735;0618 Jeremy D. Miller;Jaedyn Tonee diff --git a/src/CoreTests/Blocks/InMemoryQueueTests.cs b/src/CoreTests/Blocks/InMemoryQueueTests.cs index 2e1c5367..0b52784a 100644 --- a/src/CoreTests/Blocks/InMemoryQueueTests.cs +++ b/src/CoreTests/Blocks/InMemoryQueueTests.cs @@ -120,7 +120,73 @@ await Task.WhenAll([ var actual = list.OrderBy(x => x).ToArray(); actual.Length.ShouldBe(all.Length); - + //actual.ShouldBe(all); } + + [Fact] + public async Task synchronous_Post_does_not_drop_items_when_the_bounded_channel_fills() + { + // Regression for GH-3287: previously Post() used a non-blocking TryWrite and silently dropped + // every item once more than the bounded capacity (10k) were queued faster than the reader drained. + // Here the reader is held closed until the channel has saturated and the producer is blocked in + // Post() waiting for capacity -- proving the item is back-pressured rather than dropped. + const int count = 25_000; + var processed = 0; + using var gate = new ManualResetEventSlim(false); + + await using var queue = new Block(1, Block.DefaultBoundedCapacity, (_, token) => + { + gate.Wait(token); + Interlocked.Increment(ref processed); + return Task.CompletedTask; + }); + + var producer = Task.Run(() => + { + for (var i = 0; i < count; i++) + { + queue.Post(i); + } + }); + + // Wait until the channel is saturated (producer is now blocked inside Post waiting for room). + var deadline = DateTime.UtcNow.AddSeconds(10); + while (queue.Count < Block.DefaultBoundedCapacity && DateTime.UtcNow < deadline) + { + await Task.Delay(10); + } + + queue.Count.ShouldBeGreaterThanOrEqualTo((uint)Block.DefaultBoundedCapacity); + producer.IsCompleted.ShouldBeFalse(); // still blocked -> the overflow item was NOT dropped + + gate.Set(); + + await producer; + await queue.WaitForCompletionAsync(); + + processed.ShouldBe(count); + } + + [Fact] + public async Task unbounded_block_never_blocks_or_drops_on_Post() + { + const int count = 25_000; + var processed = 0; + + await using var queue = new Block(1, Block.Unbounded, (_, _) => + { + Interlocked.Increment(ref processed); + return Task.CompletedTask; + }); + + for (var i = 0; i < count; i++) + { + queue.Post(i); + } + + await queue.WaitForCompletionAsync(); + + processed.ShouldBe(count); + } } diff --git a/src/JasperFx/Blocks/Block.cs b/src/JasperFx/Blocks/Block.cs index 297f5e55..0b7c4a10 100644 --- a/src/JasperFx/Blocks/Block.cs +++ b/src/JasperFx/Blocks/Block.cs @@ -6,6 +6,19 @@ namespace JasperFx.Blocks; public class Block : BlockBase { + /// + /// The default bounded capacity for a Block's underlying channel. Writers that outrun the readers + /// are throttled (back pressure) rather than dropped once this many items are buffered. + /// + public const int DefaultBoundedCapacity = 10000; + + /// + /// Pass as the boundedCapacity to build a Block backed by an unbounded channel. Use this for blocks + /// that may legitimately re-enqueue onto themselves from within their own processing action (which + /// would deadlock against back pressure) or where callers must never block when posting. + /// + public const int Unbounded = -1; + private readonly Func _action; private readonly Channel _channel; private readonly CancellationTokenSource _cancellation = new(); @@ -23,23 +36,44 @@ public Block(Action action) : this(1, (item, _) => action(item); return Task.CompletedTask; }){} - + /// /// A CancellationToken for the overarching process /// public CancellationToken Cancellation { get; set; } = CancellationToken.None; public Block(int parallelCount, Func action) + : this(parallelCount, DefaultBoundedCapacity, action) + { + } + + /// The number of parallel readers processing the channel + /// + /// The maximum number of buffered items before writers are throttled with back pressure. Pass + /// for an unbounded channel that never blocks or drops on write. + /// + /// The processing action applied to each item + public Block(int parallelCount, int boundedCapacity, Func action) { _action = action; - - // TODO -- what about on dropping???????? - _channel = Channel.CreateBounded(new BoundedChannelOptions(10000) - { - SingleReader = parallelCount == 1, - SingleWriter = false, - AllowSynchronousContinuations = true - }); + + // A bounded channel using BoundedChannelFullMode.Wait gives us back pressure: writers wait for + // capacity rather than silently dropping items (GH-3287). Post() honors this by blocking-waiting + // for room when TryWrite fails. Unbounded channels never block or drop, but grow with memory. + _channel = boundedCapacity == Unbounded + ? Channel.CreateUnbounded(new UnboundedChannelOptions + { + SingleReader = parallelCount == 1, + SingleWriter = false, + AllowSynchronousContinuations = true + }) + : Channel.CreateBounded(new BoundedChannelOptions(boundedCapacity) + { + SingleReader = parallelCount == 1, + SingleWriter = false, + AllowSynchronousContinuations = true, + FullMode = BoundedChannelFullMode.Wait + }); _tasks = new Task[parallelCount]; for (int i = 0; i < parallelCount; i++) @@ -137,11 +171,34 @@ private async Task processAsync() public override void Post(T item) { + if (_latched) return; + Interlocked.Increment(ref _count); - if (!_channel.Writer.TryWrite(item)) + + if (_channel.Writer.TryWrite(item)) + { + return; + } + + // The channel is bounded and currently full. Rather than silently dropping the item (the old + // behavior that lost messages once >10k were queued, GH-3287), block-wait for capacity. This is + // safe as long as Post() is not called from within this same Block's own processing action on a + // saturated channel -- blocks that re-enqueue onto themselves must use the Unbounded capacity. + try { - // TODO -- REALLY NEED TO DEAL WITH THIS - Debug.WriteLine($"Was not able to write {item} to the queue synchronously!"); + _channel.Writer.WriteAsync(item, _cancellation.Token).AsTask().GetAwaiter().GetResult(); + } + catch (Exception e) + { + Interlocked.Decrement(ref _count); + try + { + _onError(item, e); + } + catch (Exception) + { + Debug.WriteLine($"Was not able to write {item} to the queue synchronously!"); + } } } diff --git a/src/JasperFx/Blocks/RetryBlock.cs b/src/JasperFx/Blocks/RetryBlock.cs index fe759db3..c9e768f4 100644 --- a/src/JasperFx/Blocks/RetryBlock.cs +++ b/src/JasperFx/Blocks/RetryBlock.cs @@ -41,7 +41,10 @@ public RetryBlock(IItemHandler handler, ILogger logger, CancellationToken can _logger = logger; _cancellationToken = cancellationToken; - _block = new Block(executeAsync); + // Unbounded: executeAsync re-posts failed items back onto this same block from within its own + // processing action. With a bounded, back-pressuring block that self-re-enqueue would deadlock + // against a full channel (GH-3287), so retries must never block on write. + _block = new Block(1, Block.Unbounded, executeAsync); } public int MaximumAttempts { get; set; } = 3;