From 6d6e14bc4bd0ee09f3d874469f86f802ed8ce321 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 5 Jul 2026 16:21:26 -0500 Subject: [PATCH] feat(batching): IsolateBatchMembers error policy for opaque batch failures (GH-3289 Phase 3b) Phase 3b of the #3289 batch-processing plan: item 3. For an OPAQUE batch failure (the handler throws but can't name the bad item), the new error policy opts.Policies.OnException().IsolateBatchMembers(); re-runs each member of the failed batch as its own size-1 batch (via 3a's shared BatchReplay primitive), so only the member that actually reproduces the failure is dead-lettered while the healthy members succeed. - New ProbeIndividuallyContinuation(+Source): splits a batch (>1 member) into singletons; a size-<=1 batch is dead-lettered (the terminating/base case, so a probed singleton that fails again just DLQs). - New IsolateBatchMembers() verb on IFailureActions / FailureActions / PolicyExpression, installed as an every-attempt ContinueWith. Composes with the retry verbs by exception type: a transient SqlException can RetryWithCooldown the whole batch while a deterministic ValidationException isolates. Tests (DB-free): opaque-failure isolation (only the bad item dead-letters, healthy members succeed, captured via IDeadLetterInterceptor) + non-batched fallback to plain dead-lettering. 253 error-handling/batch tests still green. Follow-ups (called out in docs): count-based ProbeIndividuallyAfter(N) (item 2) and poisoning every member that collapsed into a coalesced CoalesceBy key. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/guide/handlers/batching.md | 29 +++- .../Acceptance/batch_isolate_members.cs | 155 ++++++++++++++++++ .../ErrorHandling/PolicyExpression.cs | 24 +++ .../Batching/ProbeIndividuallyContinuation.cs | 66 ++++++++ 4 files changed, 271 insertions(+), 3 deletions(-) create mode 100644 src/Testing/CoreTests/Acceptance/batch_isolate_members.cs create mode 100644 src/Wolverine/Runtime/Batching/ProbeIndividuallyContinuation.cs diff --git a/docs/guide/handlers/batching.md b/docs/guide/handlers/batching.md index ab2e0f792..f80647b19 100644 --- a/docs/guide/handlers/batching.md +++ b/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().IsolateBatchMembers(); + +// ...while a transient error still retries the whole batch (the two policies compose by exception type). +opts.Policies.OnException().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 diff --git a/src/Testing/CoreTests/Acceptance/batch_isolate_members.cs b/src/Testing/CoreTests/Acceptance/batch_isolate_members.cs new file mode 100644 index 000000000..62c215c5a --- /dev/null +++ b/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(); + + opts.Services.AddSingleton(_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().IsolateBatchMembers(); + + opts.BatchMessagesOf(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().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(); + opts.Services.AddSingleton(_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().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().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 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(); + } + } + } +} diff --git a/src/Wolverine/ErrorHandling/PolicyExpression.cs b/src/Wolverine/ErrorHandling/PolicyExpression.cs index e3ee32367..9092eef86 100644 --- a/src/Wolverine/ErrorHandling/PolicyExpression.cs +++ b/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 /// IAdditionalActions Discard(); + /// + /// 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. + /// + IAdditionalActions IsolateBatchMembers(); + /// /// Schedule the message for additional attempts with a delay. Use this @@ -671,6 +683,18 @@ public IAdditionalActions Discard() return new FailureActions(_match, _parent).Discard(); } + /// + /// 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. + /// + public IAdditionalActions IsolateBatchMembers() + { + return new FailureActions(_match, _parent).IsolateBatchMembers(); + } + /// /// Pause the sending agent for the specified duration, then automatically resume. /// Only applicable when used with sending failure policies. diff --git a/src/Wolverine/Runtime/Batching/ProbeIndividuallyContinuation.cs b/src/Wolverine/Runtime/Batching/ProbeIndividuallyContinuation.cs new file mode 100644 index 000000000..497c54783 --- /dev/null +++ b/src/Wolverine/Runtime/Batching/ProbeIndividuallyContinuation.cs @@ -0,0 +1,66 @@ +using System.Diagnostics; +using Wolverine.ErrorHandling; + +namespace Wolverine.Runtime.Batching; + +/// +/// Built-in continuation source for the batch member-isolation error policy +/// (OnException<T>().IsolateBatchMembers()). Used for opaque 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. +/// +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")); + } +}