fix(blocks): Post no longer silently drops items on a full bounded channel

Directory.Build.props

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
 <Project>
     <PropertyGroup>
-        <JasperFxVersion>2.19.0</JasperFxVersion>
+        <JasperFxVersion>2.19.1</JasperFxVersion>
         <LangVersion>13</LangVersion>
         <NoWarn>1570;1571;1572;1573;1574;1587;1591;1701;1702;1711;1735;0618</NoWarn>
         <Authors>Jeremy D. Miller;Jaedyn Tonee</Authors>

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<int>(1, Block<int>.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<int>.DefaultBoundedCapacity && DateTime.UtcNow < deadline)
+        {
+            await Task.Delay(10);
+        }
+
+        queue.Count.ShouldBeGreaterThanOrEqualTo((uint)Block<int>.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<int>(1, Block<int>.Unbounded, (_, _) =>
+        {
+            Interlocked.Increment(ref processed);
+            return Task.CompletedTask;
+        });
+
+        for (var i = 0; i < count; i++)
+        {
+            queue.Post(i);
+        }
+
+        await queue.WaitForCompletionAsync();
+
+        processed.ShouldBe(count);
+    }
 }

src/JasperFx/Blocks/Block.cs

@@ -6,6 +6,19 @@ namespace JasperFx.Blocks;
 
 public class Block<T> : BlockBase<T>
 {
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public const int DefaultBoundedCapacity = 10000;
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public const int Unbounded = -1;
+
     private readonly Func<T, CancellationToken, Task> _action;
     private readonly Channel<T> _channel;
     private readonly CancellationTokenSource _cancellation = new();
@@ -23,23 +36,44 @@ public Block(Action<T> action) : this(1, (item, _) =>
         action(item);
         return Task.CompletedTask;
     }){}
-    
+
     /// <summary>
     /// A CancellationToken for the overarching process
     /// </summary>
     public CancellationToken Cancellation { get; set;  } = CancellationToken.None;
 
     public Block(int parallelCount, Func<T, CancellationToken, Task> action)
+        : this(parallelCount, DefaultBoundedCapacity, action)
+    {
+    }
+
+    /// <param name="parallelCount">The number of parallel readers processing the channel</param>
+    /// <param name="boundedCapacity">
+    /// The maximum number of buffered items before writers are throttled with back pressure. Pass
+    /// <see cref="Unbounded"/> for an unbounded channel that never blocks or drops on write.
+    /// </param>
+    /// <param name="action">The processing action applied to each item</param>
+    public Block(int parallelCount, int boundedCapacity, Func<T, CancellationToken, Task> action)
     {
         _action = action;
-
-        // TODO -- what about on dropping????????
-        _channel = Channel.CreateBounded<T>(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<T>(new UnboundedChannelOptions
+            {
+                SingleReader = parallelCount == 1,
+                SingleWriter = false,
+                AllowSynchronousContinuations = true
+            })
+            : Channel.CreateBounded<T>(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!");
+            }
         }
     }
 

src/JasperFx/Blocks/RetryBlock.cs

@@ -41,7 +41,10 @@ public RetryBlock(IItemHandler<T> handler, ILogger logger, CancellationToken can
         _logger = logger;
         _cancellationToken = cancellationToken;
 
-        _block = new Block<Item>(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<Item>(1, Block<Item>.Unbounded, executeAsync);
     }
 
     public int MaximumAttempts { get; set; } = 3;