feat(batching): ApplyItemException to isolate poison items from a batch (GH-3289 Phase 3a)

docs/guide/handlers/batching.md

@@ -296,6 +296,54 @@ public static void Handle(Item[] items, IBatchContext batch)
 settled. When combined with `CoalesceBy`, `Members` still lists **every** original member message (all the
 ones that settle with the batch), even though the `items` array the handler sees was de-duplicated.
 
+## Isolating poison items with `ApplyItemException`
+
+By default a failed batch is retried and dead-lettered **as a unit** — one poison message takes every other
+message in the batch to the dead-letter queue with it. When your batch handler already knows *which* item is
+bad (a validation failure, a specific row a bulk API rejected), it can throw `ApplyItemException` to isolate
+just that item instead. Wolverine dead-letters only the named item(s) and dispositions the survivors, so the
+healthy messages are not collateral damage:
+
+```csharp
+public static void Handle(Order[] orders)
+{
+    foreach (var order in orders)
+    {
+        if (!IsValid(order))
+        {
+            // Dead-letter this one order, re-run the batch handler over the rest
+            throw ApplyItemException.DeadLetterAndReplayOthers(order);
+        }
+        // ... process order ...
+    }
+}
+```
+
+The static factories make the intent explicit at the call site:
+
+```csharp
+throw ApplyItemException.DeadLetterAndReplayOthers(badOrder);            // DLQ it, re-run the rest
+throw ApplyItemException.DeadLetterAndAckOthers(bad1, bad2);             // DLQ them, ack the rest as-is
+throw ApplyItemException.DeadLetter(poison: bads, ackItems: committed);  // DLQ bads, ack what I committed, replay the remainder
+```
+
+- **`DeadLetterAndReplayOthers`** — dead-letter the poison item(s), then re-run the batch handler over the
+  remaining items as a fresh, reduced batch. Use it when the handler had not yet committed anything.
+- **`DeadLetterAndAckOthers`** — dead-letter the poison item(s) and acknowledge every other item as-is (no
+  re-run). Use it when the handler already committed the good items in the same transaction.
+- **`DeadLetter(poison, ackItems)`** — dead-letter the poison item(s), acknowledge the items you explicitly
+  committed, and replay the remainder.
+
+Throwing the exception *is* the opt-in — there is no configuration to enable. The items are matched back to
+their original messages by reference identity, so throw the factory with the actual object(s) handed to your
+handler.
+
+::: info
+This isolates messages the handler can *name*. For opaque failures where the handler cannot tell which item
+is bad, a bounded "probe each item individually" fallback and exception-type-driven isolation
+(`IsolateBatchMembers()`) are planned as a follow-up (see [GH-3289](https://github.com/JasperFx/wolverine/issues/3289)).
+:::
+
 ## Custom Batching Strategies
 
 ::: info

src/Testing/CoreTests/Acceptance/batch_item_isolation.cs

@@ -0,0 +1,227 @@
+using JasperFx.Core;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Wolverine;
+using Wolverine.Persistence.Durability;
+using Xunit;
+
+namespace CoreTests.Acceptance;
+
+public class ApplyItemExceptionTests
+{
+    [Fact]
+    public void deadletter_and_replay_others()
+    {
+        var a = new object();
+        var ex = ApplyItemException.DeadLetterAndReplayOthers(a);
+        ex.PoisonItems.ShouldHaveSingleItem().ShouldBeSameAs(a);
+        ex.Disposition.ShouldBe(NonPoisonItems.Replay);
+        ex.AckItems.ShouldBeEmpty();
+    }
+
+    [Fact]
+    public void deadletter_and_replay_others_with_inner()
+    {
+        var a = new object();
+        var inner = new InvalidOperationException("boom");
+        var ex = ApplyItemException.DeadLetterAndReplayOthers(a, inner);
+        ex.PoisonItems.ShouldHaveSingleItem().ShouldBeSameAs(a);
+        ex.Disposition.ShouldBe(NonPoisonItems.Replay);
+        ex.InnerException.ShouldBeSameAs(inner);
+    }
+
+    [Fact]
+    public void deadletter_and_ack_others()
+    {
+        var a = new object();
+        var b = new object();
+        var ex = ApplyItemException.DeadLetterAndAckOthers(a, b);
+        ex.PoisonItems.ShouldBe(new[] { a, b });
+        ex.Disposition.ShouldBe(NonPoisonItems.AckAll);
+    }
+
+    [Fact]
+    public void deadletter_with_selected_acks()
+    {
+        var poison = new object();
+        var ack = new object();
+        var ex = ApplyItemException.DeadLetter(new[] { poison }, new[] { ack });
+        ex.PoisonItems.ShouldHaveSingleItem().ShouldBeSameAs(poison);
+        ex.AckItems.ShouldHaveSingleItem().ShouldBeSameAs(ack);
+        ex.Disposition.ShouldBe(NonPoisonItems.AckSelected);
+    }
+
+    [Fact]
+    public void rejects_empty_or_null_poison()
+    {
+        Should.Throw<ArgumentException>(() => ApplyItemException.DeadLetterAndReplayOthers());
+        Should.Throw<ArgumentException>(() => ApplyItemException.DeadLetterAndAckOthers(null!));
+    }
+}
+
+public class batch_item_isolation : IAsyncLifetime
+{
+    private IHost _host = null!;
+    private readonly CapturingDeadLetterInterceptor _deadLetters = new();
+
+    public async Task InitializeAsync()
+    {
+        IsoItemBatchHandler.Reset();
+
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Discovery.DisableConventionalDiscovery();
+                opts.Discovery.IncludeType<IsoItemBatchHandler>();
+
+                opts.Services.AddSingleton<IDeadLetterInterceptor>(_deadLetters);
+
+                opts.BatchMessagesOf<IsoItem>(b =>
+                {
+                    b.BatchSize = 500;
+                    b.TriggerTime = 1.Seconds();
+                }).Sequential();
+            }).StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await _host.StopAsync();
+        _host.Dispose();
+    }
+
+    private async Task publishAsync(params IsoItem[] items)
+    {
+        var bus = _host.MessageBus();
+        foreach (var item in items)
+        {
+            await bus.PublishAsync(item);
+        }
+    }
+
+    [Fact]
+    public async Task deadletter_and_replay_others_isolates_the_poison_item()
+    {
+        IsoItemBatchHandler.Throw = items => ApplyItemException.DeadLetterAndReplayOthers(
+            items.Single(x => x.Poison));
+
+        await publishAsync(new IsoItem("a", false), new IsoItem("bad", true), new IsoItem("c", false));
+
+        // The reduced batch (survivors only) is re-run to success.
+        await IsoItemBatchHandler.SuccessSignal.Task.WaitAsync(10.Seconds());
+
+        var successful = IsoItemBatchHandler.SuccessfulRuns.ShouldHaveSingleItem();
+        successful.Select(x => x.Id).OrderBy(x => x).ShouldBe(new[] { "a", "c" });
+        successful.ShouldNotContain(x => x.Poison);
+
+        // Only the poison item was dead-lettered.
+        _deadLetters.DeadLettered.OfType<IsoItem>().Select(x => x.Id).ShouldBe(new[] { "bad" });
+    }
+
+    [Fact]
+    public async Task deadletter_and_ack_others_does_not_replay()
+    {
+        IsoItemBatchHandler.Throw = items => ApplyItemException.DeadLetterAndAckOthers(
+            items.Single(x => x.Poison));
+
+        await publishAsync(new IsoItem("a", false), new IsoItem("bad", true), new IsoItem("c", false));
+
+        // Wait for the poison item to be dead-lettered, then confirm no replay happened.
+        await _deadLetters.Signal.Task.WaitAsync(10.Seconds());
+        await Task.Delay(500.Milliseconds());
+
+        _deadLetters.DeadLettered.OfType<IsoItem>().Select(x => x.Id).ShouldBe(new[] { "bad" });
+
+        // AckAll -> survivors acked as-is, handler NOT re-invoked, so no successful run recorded.
+        IsoItemBatchHandler.SuccessfulRuns.ShouldBeEmpty();
+        IsoItemBatchHandler.Invocations.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task deadletter_with_selected_acks_replays_the_remainder()
+    {
+        IsoItemBatchHandler.Throw = items =>
+        {
+            var poison = items.Single(x => x.Poison);
+            var ack = items.Single(x => x.Id == "ackme");
+            return ApplyItemException.DeadLetter(new object[] { poison }, new object[] { ack });
+        };
+
+        await publishAsync(
+            new IsoItem("ackme", false),
+            new IsoItem("bad", true),
+            new IsoItem("replay1", false),
+            new IsoItem("replay2", false));
+
+        await IsoItemBatchHandler.SuccessSignal.Task.WaitAsync(10.Seconds());
+
+        // Only the two non-acked survivors are replayed; "ackme" was settled without a re-run.
+        var successful = IsoItemBatchHandler.SuccessfulRuns.ShouldHaveSingleItem();
+        successful.Select(x => x.Id).OrderBy(x => x).ShouldBe(new[] { "replay1", "replay2" });
+
+        _deadLetters.DeadLettered.OfType<IsoItem>().Select(x => x.Id).ShouldBe(new[] { "bad" });
+    }
+}
+
+public record IsoItem(string Id, bool Poison);
+
+public class IsoItemBatchHandler
+{
+    private static readonly object _locker = new();
+
+    public static Func<IsoItem[], ApplyItemException>? Throw;
+    public static List<IsoItem[]> SuccessfulRuns { get; } = new();
+    public static int Invocations;
+    public static TaskCompletionSource SuccessSignal = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+    public static void Reset()
+    {
+        Throw = null;
+        lock (_locker)
+        {
+            SuccessfulRuns.Clear();
+        }
+
+        Invocations = 0;
+        SuccessSignal = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+    }
+
+    public void Handle(IsoItem[] items)
+    {
+        Interlocked.Increment(ref Invocations);
+
+        // Only the first (poison-carrying) batch throws; the replayed survivor batch has no poison.
+        if (Throw != null && items.Any(x => x.Poison))
+        {
+            throw Throw(items);
+        }
+
+        lock (_locker)
+        {
+            SuccessfulRuns.Add(items);
+        }
+
+        SuccessSignal.TrySetResult();
+    }
+}
+
+public class CapturingDeadLetterInterceptor : IDeadLetterInterceptor
+{
+    public List<object> DeadLettered { get; } = new();
+    public TaskCompletionSource Signal { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+    public ValueTask<Exception?> BeforeStoreAsync(Envelope envelope, Exception? exception,
+        CancellationToken cancellation)
+    {
+        lock (DeadLettered)
+        {
+            if (envelope.Message != null)
+            {
+                DeadLettered.Add(envelope.Message);
+            }
+        }
+
+        Signal.TrySetResult();
+        return new ValueTask<Exception?>(exception);
+    }
+}

src/Wolverine/ApplyItemException.cs

@@ -0,0 +1,116 @@
+namespace Wolverine;
+
+/// <summary>
+/// Disposition for the non-poison ("surviving") members of a batch when a batch handler throws an
+/// <see cref="ApplyItemException"/> to isolate poison items.
+/// </summary>
+public enum NonPoisonItems
+{
+    /// <summary>Acknowledge every surviving item as-is; do not re-run any of them.</summary>
+    AckAll,
+
+    /// <summary>Re-run the batch handler over every surviving item as a fresh, reduced batch.</summary>
+    Replay,
+
+    /// <summary>
+    /// Acknowledge the explicitly listed <see cref="ApplyItemException.AckItems"/> and replay the
+    /// remaining survivors.
+    /// </summary>
+    AckSelected
+}
+
+/// <summary>
+/// Thrown from a batch message handler that already knows which item(s) in the batch are "poison"
+/// (deterministically bad) to isolate them from the healthy members. Wolverine dead-letters only the
+/// poison items and dispositions the survivors per <see cref="Disposition"/>, so a single bad message
+/// no longer dead-letters the whole batch. This mirrors JasperFx.Events' <c>ApplyEventException</c> /
+/// <c>SkipApplyErrors</c>, where the daemon skips the offending event, dead-letters it, and rebuilds the
+/// batch without it so the good events still commit. Construct via the static factories so the intent
+/// reads clearly at the call site; there is no other opt-in — throwing the exception IS the opt-in.
+/// </summary>
+public sealed class ApplyItemException : Exception
+{
+    private ApplyItemException(IReadOnlyList<object> poisonItems, NonPoisonItems disposition,
+        IReadOnlyList<object> ackItems, Exception? inner)
+        : base(buildMessage(poisonItems, disposition), inner)
+    {
+        PoisonItems = poisonItems;
+        Disposition = disposition;
+        AckItems = ackItems;
+    }
+
+    /// <summary>The item(s) to dead-letter. Never empty.</summary>
+    public IReadOnlyList<object> PoisonItems { get; }
+
+    /// <summary>What to do with the surviving (non-poison) items.</summary>
+    public NonPoisonItems Disposition { get; }
+
+    /// <summary>
+    /// The survivors to acknowledge as-is. Consulted only when <see cref="Disposition"/> is
+    /// <see cref="NonPoisonItems.AckSelected"/>; the remaining survivors are replayed.
+    /// </summary>
+    public IReadOnlyList<object> AckItems { get; }
+
+    /// <summary>
+    /// Dead-letter the poison item(s) and re-run the batch handler over all the remaining items.
+    /// </summary>
+    public static ApplyItemException DeadLetterAndReplayOthers(params object[] poison)
+    {
+        return new ApplyItemException(requireNonEmpty(poison), NonPoisonItems.Replay,
+            Array.Empty<object>(), null);
+    }
+
+    /// <summary>
+    /// Dead-letter a single poison item (recording the underlying exception) and re-run the batch
+    /// handler over all the remaining items. Mirrors <c>ApplyEventException(@event, inner)</c>.
+    /// </summary>
+    public static ApplyItemException DeadLetterAndReplayOthers(object poison, Exception inner)
+    {
+        return new ApplyItemException(requireNonEmpty(new[] { poison }), NonPoisonItems.Replay,
+            Array.Empty<object>(), inner);
+    }
+
+    /// <summary>
+    /// Dead-letter the poison item(s) and acknowledge all the remaining items as-is (no re-run). Use
+    /// this when the batch handler already committed the good items in the same transaction.
+    /// </summary>
+    public static ApplyItemException DeadLetterAndAckOthers(params object[] poison)
+    {
+        return new ApplyItemException(requireNonEmpty(poison), NonPoisonItems.AckAll,
+            Array.Empty<object>(), null);
+    }
+
+    /// <summary>
+    /// Dead-letter the poison item(s), acknowledge the explicitly listed <paramref name="ackItems"/>
+    /// as-is, and replay the remaining survivors.
+    /// </summary>
+    public static ApplyItemException DeadLetter(object[] poison, object[] ackItems)
+    {
+        return new ApplyItemException(requireNonEmpty(poison), NonPoisonItems.AckSelected,
+            ackItems ?? Array.Empty<object>(), null);
+    }
+
+    private static object[] requireNonEmpty(object[] poison)
+    {
+        if (poison == null || poison.Length == 0)
+        {
+            throw new ArgumentException("At least one poison item is required", nameof(poison));
+        }
+
+        foreach (var item in poison)
+        {
+            if (item is null)
+            {
+                throw new ArgumentException("Poison items cannot be null", nameof(poison));
+            }
+        }
+
+        return poison;
+    }
+
+    private static string buildMessage(IReadOnlyList<object> poisonItems, NonPoisonItems disposition)
+    {
+        var count = poisonItems?.Count ?? 0;
+        return $"Isolating {count} poison item(s) from the batch (surviving items: {disposition}).";
+    }
+}