feat(batching): IsolateBatchMembers error policy for opaque batch failures (GH-3289 Phase 3b)

docs/guide/handlers/batching.md

@@ -346,10 +346,33 @@ Throwing the exception *is* the opt-in — there is no configuration to enable.
 their original messages by reference identity, so throw the factory with the actual object(s) handed to your
 handler.
 
+`ApplyItemException` is for failures the handler can *name*. For **opaque** failures — where the handler
+throws but cannot tell which item was the culprit — use the `IsolateBatchMembers()` error policy below.
+
+## Isolating an opaque batch failure with `IsolateBatchMembers`
+
+When a batch handler throws an exception it can't attribute to a specific item, you can still avoid
+dead-lettering the whole batch by isolating the failing member. The `IsolateBatchMembers()` error policy,
+keyed on an exception type, re-runs each member of the failed batch as its own size-1 batch — so only the
+member that actually reproduces the failure is dead-lettered, and every healthy member succeeds:
+
+```csharp
+// A deterministic error isolates the offending member...
+opts.Policies.OnException<ValidationException>().IsolateBatchMembers();
+
+// ...while a transient error still retries the whole batch (the two policies compose by exception type).
+opts.Policies.OnException<SqlException>().RetryWithCooldown(100.Milliseconds(), 1.Seconds());
+```
+
+Because it is matched by exception type, it composes with the ordinary retry verbs: a transient
+`SqlException` retries the whole batch with a cooldown, while a deterministic `ValidationException` isolates
+the bad member. The isolation is bounded and one-time — a member that has already been reduced to a size-1
+batch is simply dead-lettered rather than probed again. On a message type that is **not** batched,
+`IsolateBatchMembers()` behaves like a plain move-to-error-queue (there is nothing to isolate).
+
 ::: 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)).
+A count-based variant — `ProbeIndividuallyAfter(attempts: N)`, which probes individually only after N
+whole-batch failures — is a planned follow-up (see [GH-3289](https://github.com/JasperFx/wolverine/issues/3289)).
 :::
 
 ## Custom Batching Strategies

src/Testing/CoreTests/Acceptance/batch_isolate_members.cs

@@ -0,0 +1,155 @@
+using JasperFx.Core;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Wolverine;
+using Wolverine.ErrorHandling;
+using Wolverine.Persistence.Durability;
+using Xunit;
+
+namespace CoreTests.Acceptance;
+
+public class batch_isolate_members : IAsyncLifetime
+{
+    private IHost _host = null!;
+    private readonly CapturingDeadLetterInterceptor _deadLetters = new();
+
+    public async Task InitializeAsync()
+    {
+        ProbeItemBatchHandler.Reset();
+
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Discovery.DisableConventionalDiscovery();
+                opts.Discovery.IncludeType<ProbeItemBatchHandler>();
+
+                opts.Services.AddSingleton<IDeadLetterInterceptor>(_deadLetters);
+
+                // The flagship composition from the plan: a deterministic error type isolates the bad
+                // member, while transient errors (not exercised here) would retry the whole batch.
+                opts.Policies.OnException<ProbeFailure>().IsolateBatchMembers();
+
+                opts.BatchMessagesOf<ProbeItem>(b =>
+                {
+                    b.BatchSize = 500;
+                    b.TriggerTime = 1.Seconds();
+                }).Sequential();
+            }).StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await _host.StopAsync();
+        _host.Dispose();
+    }
+
+    [Fact]
+    public async Task isolates_the_failing_member_by_probing_individually()
+    {
+        var bus = _host.MessageBus();
+        await bus.PublishAsync(new ProbeItem("good1", false));
+        await bus.PublishAsync(new ProbeItem("bad", true));
+        await bus.PublishAsync(new ProbeItem("good2", false));
+
+        // The whole batch throws the opaque ProbeFailure; IsolateBatchMembers re-runs each member as its
+        // own size-1 batch. The two healthy singletons succeed; the poison singleton dead-letters.
+        await ProbeItemBatchHandler.BothGoodsSucceeded.Task.WaitAsync(10.Seconds());
+        await _deadLetters.Signal.Task.WaitAsync(10.Seconds());
+
+        ProbeItemBatchHandler.SucceededIds.OrderBy(x => x).ShouldBe(new[] { "good1", "good2" });
+
+        // Only the poison item was dead-lettered - the healthy members were not collateral damage.
+        _deadLetters.DeadLettered.OfType<ProbeItem>().Select(x => x.Id).ShouldBe(new[] { "bad" });
+    }
+}
+
+public class isolate_batch_members_on_a_non_batched_message : IAsyncLifetime
+{
+    private IHost _host = null!;
+    private readonly CapturingDeadLetterInterceptor _deadLetters = new();
+
+    public async Task InitializeAsync()
+    {
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Discovery.DisableConventionalDiscovery();
+                opts.Discovery.IncludeType<SoloProbeHandler>();
+                opts.Services.AddSingleton<IDeadLetterInterceptor>(_deadLetters);
+
+                // IsolateBatchMembers on a message type that is NOT batched must behave like a plain
+                // move-to-error-queue: there is nothing to isolate.
+                opts.Policies.OnException<ProbeFailure>().IsolateBatchMembers();
+            }).StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await _host.StopAsync();
+        _host.Dispose();
+    }
+
+    [Fact]
+    public async Task falls_back_to_dead_lettering_the_single_message()
+    {
+        await _host.MessageBus().PublishAsync(new SoloProbe("only"));
+
+        await _deadLetters.Signal.Task.WaitAsync(10.Seconds());
+
+        _deadLetters.DeadLettered.OfType<SoloProbe>().Select(x => x.Id).ShouldBe(new[] { "only" });
+    }
+}
+
+public record SoloProbe(string Id);
+
+public class SoloProbeHandler
+{
+    public void Handle(SoloProbe message)
+    {
+        throw new ProbeFailure();
+    }
+}
+
+public record ProbeItem(string Id, bool Poison);
+
+// Opaque failure: the handler cannot name which item is bad, it just throws.
+public class ProbeFailure : Exception;
+
+public class ProbeItemBatchHandler
+{
+    private static readonly object _locker = new();
+
+    public static List<string> SucceededIds { get; } = new();
+    public static TaskCompletionSource BothGoodsSucceeded = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+    public static void Reset()
+    {
+        lock (_locker)
+        {
+            SucceededIds.Clear();
+        }
+
+        BothGoodsSucceeded = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+    }
+
+    public void Handle(ProbeItem[] items)
+    {
+        if (items.Any(x => x.Poison))
+        {
+            throw new ProbeFailure();
+        }
+
+        lock (_locker)
+        {
+            foreach (var item in items)
+            {
+                SucceededIds.Add(item.Id);
+            }
+
+            if (SucceededIds.Count >= 2)
+            {
+                BothGoodsSucceeded.TrySetResult();
+            }
+        }
+    }
+}

src/Wolverine/ErrorHandling/PolicyExpression.cs

@@ -306,6 +306,11 @@ public IAdditionalActions Discard()
         return this;
     }
 
+    public IAdditionalActions IsolateBatchMembers()
+    {
+        return ContinueWith(new Runtime.Batching.ProbeIndividuallyContinuationSource());
+    }
+
     public IAdditionalActions PauseSending(TimeSpan pauseTime)
     {
         var slot = _rule.AddSlot(new PauseSendingContinuation(pauseTime));
@@ -524,6 +529,13 @@ public interface IFailureActions
     /// </summary>
     IAdditionalActions Discard();
 
+    /// <summary>
+    ///     For a batched message handler, isolate the failing member(s): re-run each member of the batch
+    ///     as its own size-1 batch so only the message that actually causes this exception is
+    ///     dead-lettered, while the healthy members succeed. No effect on non-batched messages. GH-3289.
+    /// </summary>
+    IAdditionalActions IsolateBatchMembers();
+
 
     /// <summary>
     ///     Schedule the message for additional attempts with a delay. Use this
@@ -671,6 +683,18 @@ public IAdditionalActions Discard()
         return new FailureActions(_match, _parent).Discard();
     }
 
+    /// <summary>
+    ///     For a batched message handler, isolate the failing member(s): re-run each member of the batch
+    ///     as its own size-1 batch so only the message that actually causes this exception is
+    ///     dead-lettered, while the healthy members succeed. Composes with the retry verbs on other
+    ///     exception types (e.g. RetryWithCooldown for transient errors). No effect on non-batched
+    ///     messages. See GH-3289.
+    /// </summary>
+    public IAdditionalActions IsolateBatchMembers()
+    {
+        return new FailureActions(_match, _parent).IsolateBatchMembers();
+    }
+
     /// <summary>
     /// Pause the sending agent for the specified duration, then automatically resume.
     /// Only applicable when used with sending failure policies.

src/Wolverine/Runtime/Batching/ProbeIndividuallyContinuation.cs

@@ -0,0 +1,66 @@
+using System.Diagnostics;
+using Wolverine.ErrorHandling;
+
+namespace Wolverine.Runtime.Batching;
+
+/// <summary>
+/// Built-in continuation source for the batch member-isolation error policy
+/// (<c>OnException&lt;T&gt;().IsolateBatchMembers()</c>). Used for <em>opaque</em> batch failures where the
+/// handler cannot name the bad item: re-run each member as its own size-1 batch so only the one carrying
+/// the bad element dead-letters and the healthy members succeed. See GH-3289.
+/// </summary>
+internal class ProbeIndividuallyContinuationSource : IContinuationSource
+{
+    public string Description => "Isolate the failing batch member by re-running each member individually";
+
+    public IContinuation Build(Exception ex, Envelope envelope)
+    {
+        return new ProbeIndividuallyContinuation(ex);
+    }
+}
+
+internal class ProbeIndividuallyContinuation : IContinuation
+{
+    private readonly Exception _exception;
+
+    public ProbeIndividuallyContinuation(Exception exception)
+    {
+        _exception = exception ?? throw new ArgumentNullException(nameof(exception));
+    }
+
+    public async ValueTask ExecuteAsync(IEnvelopeLifecycle lifecycle, IWolverineRuntime runtime,
+        DateTimeOffset now, Activity? activity)
+    {
+        var batch = lifecycle.Envelope;
+        var members = batch?.Batch;
+
+        // A single-member "batch" is already isolated -> dead-letter it. This is also the terminating
+        // case: a multi-item batch is probed into singletons, and each failing singleton lands here.
+        if (batch is null || members is null || members.Length <= 1)
+        {
+            await lifecycle.MoveToDeadLetterQueueAsync(_exception).ConfigureAwait(false);
+            await lifecycle.CompleteAsync().ConfigureAwait(false);
+            if (batch is not null)
+            {
+                runtime.MessageTracking.MessageFailed(batch, _exception);
+            }
+
+            return;
+        }
+
+        // Re-run each member as its own size-1 batch (via the shared re-execution primitive) so only the
+        // member carrying the bad element fails and dead-letters; the healthy members succeed. Bounded
+        // and one-time: the singletons cannot be split further (handled by the guard above).
+        foreach (var member in members)
+        {
+            await BatchReplay.EnqueueReducedBatchAsync(runtime, batch, new[] { member.Message! })
+                .ConfigureAwait(false);
+        }
+
+        // Settle the original batch; every member is now re-represented as its own singleton batch.
+        await lifecycle.CompleteAsync().ConfigureAwait(false);
+
+        runtime.MessageTracking.MessageFailed(batch, _exception);
+        activity?.AddEvent(new ActivityEvent("Wolverine.Batch.ProbedIndividually"));
+    }
+}