diff --git a/docs/guide/handlers/batching.md b/docs/guide/handlers/batching.md index a68bb690a..d7bcb7d1b 100644 --- a/docs/guide/handlers/batching.md +++ b/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 diff --git a/src/Testing/CoreTests/Acceptance/batch_item_isolation.cs b/src/Testing/CoreTests/Acceptance/batch_item_isolation.cs new file mode 100644 index 000000000..24db21e42 --- /dev/null +++ b/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(() => ApplyItemException.DeadLetterAndReplayOthers()); + Should.Throw(() => 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(); + + opts.Services.AddSingleton(_deadLetters); + + opts.BatchMessagesOf(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().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().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().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? Throw; + public static List 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 DeadLettered { get; } = new(); + public TaskCompletionSource Signal { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public ValueTask BeforeStoreAsync(Envelope envelope, Exception? exception, + CancellationToken cancellation) + { + lock (DeadLettered) + { + if (envelope.Message != null) + { + DeadLettered.Add(envelope.Message); + } + } + + Signal.TrySetResult(); + return new ValueTask(exception); + } +} diff --git a/src/Wolverine/ApplyItemException.cs b/src/Wolverine/ApplyItemException.cs new file mode 100644 index 000000000..2fbfb4a67 --- /dev/null +++ b/src/Wolverine/ApplyItemException.cs @@ -0,0 +1,116 @@ +namespace Wolverine; + +/// +/// Disposition for the non-poison ("surviving") members of a batch when a batch handler throws an +/// to isolate poison items. +/// +public enum NonPoisonItems +{ + /// Acknowledge every surviving item as-is; do not re-run any of them. + AckAll, + + /// Re-run the batch handler over every surviving item as a fresh, reduced batch. + Replay, + + /// + /// Acknowledge the explicitly listed and replay the + /// remaining survivors. + /// + AckSelected +} + +/// +/// 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 , so a single bad message +/// no longer dead-letters the whole batch. This mirrors JasperFx.Events' ApplyEventException / +/// SkipApplyErrors, 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. +/// +public sealed class ApplyItemException : Exception +{ + private ApplyItemException(IReadOnlyList poisonItems, NonPoisonItems disposition, + IReadOnlyList ackItems, Exception? inner) + : base(buildMessage(poisonItems, disposition), inner) + { + PoisonItems = poisonItems; + Disposition = disposition; + AckItems = ackItems; + } + + /// The item(s) to dead-letter. Never empty. + public IReadOnlyList PoisonItems { get; } + + /// What to do with the surviving (non-poison) items. + public NonPoisonItems Disposition { get; } + + /// + /// The survivors to acknowledge as-is. Consulted only when is + /// ; the remaining survivors are replayed. + /// + public IReadOnlyList AckItems { get; } + + /// + /// Dead-letter the poison item(s) and re-run the batch handler over all the remaining items. + /// + public static ApplyItemException DeadLetterAndReplayOthers(params object[] poison) + { + return new ApplyItemException(requireNonEmpty(poison), NonPoisonItems.Replay, + Array.Empty(), null); + } + + /// + /// Dead-letter a single poison item (recording the underlying exception) and re-run the batch + /// handler over all the remaining items. Mirrors ApplyEventException(@event, inner). + /// + public static ApplyItemException DeadLetterAndReplayOthers(object poison, Exception inner) + { + return new ApplyItemException(requireNonEmpty(new[] { poison }), NonPoisonItems.Replay, + Array.Empty(), inner); + } + + /// + /// 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. + /// + public static ApplyItemException DeadLetterAndAckOthers(params object[] poison) + { + return new ApplyItemException(requireNonEmpty(poison), NonPoisonItems.AckAll, + Array.Empty(), null); + } + + /// + /// Dead-letter the poison item(s), acknowledge the explicitly listed + /// as-is, and replay the remaining survivors. + /// + public static ApplyItemException DeadLetter(object[] poison, object[] ackItems) + { + return new ApplyItemException(requireNonEmpty(poison), NonPoisonItems.AckSelected, + ackItems ?? Array.Empty(), 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 poisonItems, NonPoisonItems disposition) + { + var count = poisonItems?.Count ?? 0; + return $"Isolating {count} poison item(s) from the batch (surviving items: {disposition})."; + } +} diff --git a/src/Wolverine/Runtime/Batching/ApplyItemContinuation.cs b/src/Wolverine/Runtime/Batching/ApplyItemContinuation.cs new file mode 100644 index 000000000..311091589 --- /dev/null +++ b/src/Wolverine/Runtime/Batching/ApplyItemContinuation.cs @@ -0,0 +1,183 @@ +using System.Diagnostics; +using System.Diagnostics.CodeAnalysis; +using Wolverine.ErrorHandling; +using Wolverine.Runtime.WorkerQueues; +using Wolverine.Transports.Local; + +namespace Wolverine.Runtime.Batching; + +/// +/// Built-in continuation source for . Registered as an indefinite +/// (every-attempt) failure rule on the exception type in the WolverineOptions constructor, so +/// throwing the exception from a batch handler is itself the opt-in. See GH-3289. +/// +internal class ApplyItemContinuationSource : IContinuationSource +{ + public string Description => "Isolate the poison item(s) named by ApplyItemException from the batch"; + + public IContinuation Build(Exception ex, Envelope envelope) + { + // The rule only matches ApplyItemException; the fallback is purely defensive. + return ex is ApplyItemException apply + ? new ApplyItemContinuation(apply) + : new MoveToErrorQueue(ex); + } +} + +/// +/// Resolves an thrown by a batch handler: dead-letters only the poison +/// member envelopes, dispositions the survivors (ack and/or replay), and settles the original batch. +/// +internal class ApplyItemContinuation : IContinuation +{ + private readonly ApplyItemException _exception; + + public ApplyItemContinuation(ApplyItemException 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; + + // Defensive: ApplyItemException thrown from something that is not actually a batch. Nothing to + // isolate, so fall back to dead-lettering the whole envelope. + if (batch is null || members is null || members.Length == 0) + { + await lifecycle.MoveToDeadLetterQueueAsync(_exception).ConfigureAwait(false); + await lifecycle.CompleteAsync().ConfigureAwait(false); + if (batch is not null) + { + runtime.MessageTracking.MessageFailed(batch, _exception); + } + + return; + } + + // Partition members by reference identity: a thrown poison item is matched to the member + // envelope whose Message IS that instance. This is exact for a normal (non-coalesced) batch. + // NOTE (GH-3289): under CoalesceBy the handler sees the last-wins instance for a key, which is + // the Message of only ONE member, so only that member is dead-lettered here; the earlier + // same-key members fall into the survivor set (acked/replayed). Poisoning every member that + // collapsed into a coalesced key is a deliberate follow-up (items 2/3). + var poison = members.Where(m => containsByReference(_exception.PoisonItems, m.Message)).ToArray(); + var survivors = members.Where(m => !poison.Contains(m)).ToArray(); + + var replay = _exception.Disposition switch + { + NonPoisonItems.AckAll => Array.Empty(), + NonPoisonItems.Replay => survivors, + NonPoisonItems.AckSelected => survivors + .Where(m => !containsByReference(_exception.AckItems, m.Message)).ToArray(), + _ => survivors + }; + + // 1. Dead-letter ONLY the poison members (the final CompleteAsync then settles them too). + if (poison.Length > 0) + { + if (lifecycle is MessageContext context) + { + await context.MoveBatchMembersToDeadLetterQueueAsync(poison, _exception).ConfigureAwait(false); + } + else + { + // No fine-grained control on this lifecycle -> fall back to whole-batch dead-lettering. + await lifecycle.MoveToDeadLetterQueueAsync(_exception).ConfigureAwait(false); + } + } + + // 2. Replay the survivors we are re-running as a fresh, reduced batch on the same local queue. + if (replay.Length > 0) + { + var items = new object[replay.Length]; + for (var i = 0; i < replay.Length; i++) + { + items[i] = replay[i].Message!; + } + + await BatchReplay.EnqueueReducedBatchAsync(runtime, batch, items).ConfigureAwait(false); + } + + // 3. Settle every ORIGINAL member (ack survivors, the now-dead-lettered poison, and the + // replayed originals whose work now rides on the fresh reduced batch) plus the batch itself. + await lifecycle.CompleteAsync().ConfigureAwait(false); + + runtime.MessageTracking.MessageFailed(batch, _exception); + activity?.AddEvent(new ActivityEvent("Wolverine.Batch.ItemsIsolated")); + } + + private static bool containsByReference(IReadOnlyList items, object? message) + { + if (message is null) + { + return false; + } + + for (var i = 0; i < items.Count; i++) + { + if (ReferenceEquals(items[i], message)) + { + return true; + } + } + + return false; + } +} + +/// +/// Shared re-execution primitive for the batch item-isolation features (GH-3289): re-enqueue a reduced +/// batch (a subset of the original items) onto the batch's own local queue so the batch handler runs +/// again over only those items. No separate single-element handler is needed. +/// +internal static class BatchReplay +{ + // Array.CreateInstance closes over the runtime batch element type, same reflective shape as the rest + // of the batching subsystem (DefaultMessageBatcher / BatchingOptions). AOT-clean apps supply their + // own IMessageBatcher and run pre-generated handlers in TypeLoadMode.Static - see the AOT guide. + [UnconditionalSuppressMessage("AOT", "IL3050", + Justification = "Reduced-batch array closed over runtime element type; AOT consumers register batchers explicitly. See AOT guide.")] + public static async ValueTask EnqueueReducedBatchAsync(IWolverineRuntime runtime, Envelope batchEnvelope, + IReadOnlyList items) + { + if (items.Count == 0) + { + return; + } + + var destination = batchEnvelope.Destination + ?? throw new InvalidOperationException( + "The batch envelope has no Destination local queue to replay a reduced batch onto"); + + if (runtime.Endpoints.AgentForLocalQueue(destination) is not ILocalQueue queue) + { + throw new InvalidOperationException( + $"Unable to resolve local queue '{destination}' to replay a reduced batch onto"); + } + + // Rebuild the typed T[] the batch handler expects from the batch message's element type. + var elementType = batchEnvelope.Message?.GetType().GetElementType() ?? items[0].GetType(); + var typedArray = Array.CreateInstance(elementType, items.Count); + var members = new Envelope[items.Count]; + for (var i = 0; i < items.Count; i++) + { + typedArray.SetValue(items[i], i); + + // Fresh member envelopes carry the items BY REFERENCE so a subsequent ApplyItemException on + // the replayed batch can still map thrown items back to their members. + members[i] = new Envelope(items[i]); + } + + var reduced = new Envelope(typedArray, members) + { + Destination = destination, + MessageType = batchEnvelope.MessageType, + TenantId = batchEnvelope.TenantId + }; + + await queue.EnqueueAsync(reduced).ConfigureAwait(false); + } +} diff --git a/src/Wolverine/Runtime/MessageContext.cs b/src/Wolverine/Runtime/MessageContext.cs index 41a2bdb50..cf6d5679b 100644 --- a/src/Wolverine/Runtime/MessageContext.cs +++ b/src/Wolverine/Runtime/MessageContext.cs @@ -408,6 +408,35 @@ public async Task MoveToDeadLetterQueueAsync(Exception exception) await _channel.CompleteAsync(Envelope).ConfigureAwait(false); } + // Dead-letter a specific SUBSET of a batch's member envelopes without touching the other members + // or completing the batch envelope. Used by the batch item-isolation continuations to move only the + // poison items to the DLQ while the survivors are acked or replayed. The batch envelope itself is + // settled afterwards by the caller's CompleteAsync(). GH-3289. + internal async Task MoveBatchMembersToDeadLetterQueueAsync(IReadOnlyList members, Exception exception) + { + if (_channel == null || Envelope == null) + { + throw new InvalidOperationException("No Envelope is active for this context"); + } + + var deadLetterQueue = tryGetDeadLetterQueue(_channel, Envelope); + if (deadLetterQueue is not null) + { + foreach (var member in members) + { + await deadLetterQueue.MoveToErrorsAsync(member, exception).ConfigureAwait(false); + } + + return; + } + + foreach (var member in members) + { + var ex = await interceptDeadLetterAsync(member, exception).ConfigureAwait(false); + await Storage.Inbox.MoveToDeadLetterStorageAsync(member, ex).ConfigureAwait(false); + } + } + // Give registered IDeadLetterInterceptor implementations a chance to mutate the envelope // (e.g. redact/encrypt the body) and/or replace the exception that gets persisted, before a // failed envelope is written to durable dead-letter storage. Interceptors run in registration diff --git a/src/Wolverine/WolverineOptions.cs b/src/Wolverine/WolverineOptions.cs index 3cce9a6ec..16492cbb8 100644 --- a/src/Wolverine/WolverineOptions.cs +++ b/src/Wolverine/WolverineOptions.cs @@ -270,6 +270,10 @@ public WolverineOptions(string? assemblyName) this.OnException().Discard(); + // GH-3289: a batch handler can throw ApplyItemException to isolate poison items from a batch. + // Throwing the exception is the opt-in; this built-in rule resolves it on every attempt. + this.OnException().ContinueWith(new Runtime.Batching.ApplyItemContinuationSource()); + MessagePartitioning = new MessagePartitioningRules(this); InternalRouteSources.Insert(0, Transports.GetOrCreate());