feat(batching): CoalesceBy de-duplication + IBatchContext (GH-3289 Phase 1)

docs/guide/handlers/batching.md

@@ -195,6 +195,57 @@ The durable inbox behaves just a little bit differently for message batching. Wo
 "handle" the individual messages, but does not mark them as handled in the message store until a batch message
 that refers to the original message is completely processed. 
 
+## De-duplicating a batch with `CoalesceBy`
+
+A very common batching workload is a "trigger storm": a bulk operation fires hundreds or thousands of
+"recalculate" messages that actually concern only a few dozen distinct entities. Batching already collapses
+those into one handler invocation, but the handler still sees every duplicate and recomputes the same entity
+many times. `CoalesceBy` de-duplicates the batch by a key so the handler sees **one message per distinct key,
+last message wins**:
+
+```csharp
+opts.BatchMessagesOf<RecalculateScores>(batching =>
+{
+    batching.BatchSize = 500;
+    batching.TriggerTime = 10.Seconds();
+
+    // The handler sees at most one RecalculateScores per AggregateId (the latest one)
+    batching.CoalesceBy((RecalculateScores x) => x.AggregateId);
+});
+```
+
+::: tip
+The key selector's lambda parameter must be **explicitly typed** to the batched element type (e.g.
+`(RecalculateScores x) => ...`) so both the message and key type arguments can be inferred.
+:::
+
+`CoalesceBy` is just sugar over the [`IMessageBatcher`](#custom-batching-strategies) seam — it installs a
+built-in `CoalescingMessageBatcher<T, TKey>` instead of the default batcher. Like the default it first groups
+by tenant id, then de-duplicates within each tenant group.
+
+Crucially, coalescing only changes **what the handler sees** — never what gets acknowledged. Every original
+member message still rides on the batch, so the transactional inbox/outbox tracking and dead-lettering behave
+exactly as they do for a normal (non-coalesced) batch. If you drop from 1,000 messages to 40 distinct keys,
+the handler runs once over 40 items, but all 1,000 member messages are settled with that batch.
+
+## Batch identity with `IBatchContext`
+
+A batched handler can inject `IBatchContext` to get read-only information about the batch it is processing —
+useful for correlating log entries, emitting batch-level metrics, or making per-batch decisions:
+
+```csharp
+public static void Handle(Item[] items, IBatchContext batch)
+{
+    // batch.BatchId is a stable id for this assembled batch (correlate your logs with it)
+    // batch.Members describes each original member message: MessageId, Attempts, SentAt
+    logger.LogInformation("Processing batch {BatchId} of {Count} members", batch.BatchId, batch.Members.Count);
+}
+```
+
+`IBatchContext` is purely informational; reading it never changes what is acknowledged or how the batch is
+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.
+
 ## Custom Batching Strategies
 
 ::: info

src/Testing/CoreTests/Acceptance/batch_coalescing_and_context.cs

@@ -0,0 +1,212 @@
+using JasperFx.Core;
+using Microsoft.Extensions.Hosting;
+using Wolverine;
+using Wolverine.Runtime.Batching;
+using Wolverine.Tracking;
+using Xunit;
+
+namespace CoreTests.Acceptance;
+
+public class CoalescingMessageBatcherTests
+{
+    private static Envelope envelopeFor(ScoreEvent message, string? tenantId = null)
+    {
+        return new Envelope(message) { TenantId = tenantId };
+    }
+
+    [Fact]
+    public void dedupes_by_key_last_wins_but_keeps_every_member_on_the_batch()
+    {
+        var batcher = new CoalescingMessageBatcher<ScoreEvent, string>(x => x.AggregateId);
+
+        var envelopes = new[]
+        {
+            envelopeFor(new ScoreEvent("A", 1)),
+            envelopeFor(new ScoreEvent("A", 2)),
+            envelopeFor(new ScoreEvent("A", 3)),
+            envelopeFor(new ScoreEvent("B", 1)),
+            envelopeFor(new ScoreEvent("B", 2))
+        };
+
+        var batch = batcher.Group(envelopes).ShouldHaveSingleItem();
+
+        // What the handler SEES: one message per key, last wins
+        var seen = batch.Message.ShouldBeOfType<ScoreEvent[]>();
+        seen.Length.ShouldBe(2);
+        seen.Single(x => x.AggregateId == "A").Version.ShouldBe(3);
+        seen.Single(x => x.AggregateId == "B").Version.ShouldBe(2);
+
+        // What gets SETTLED: every original member envelope still rides on the batch, so inbox/outbox
+        // tracking and dead-lettering are identical to a non-coalescing batch.
+        batch.Batch!.Length.ShouldBe(5);
+    }
+
+    [Fact]
+    public void groups_by_tenant_id_before_coalescing()
+    {
+        var batcher = new CoalescingMessageBatcher<ScoreEvent, string>(x => x.AggregateId);
+
+        var envelopes = new[]
+        {
+            envelopeFor(new ScoreEvent("A", 1), "tenant1"),
+            envelopeFor(new ScoreEvent("A", 2), "tenant2"),
+            envelopeFor(new ScoreEvent("A", 3), "tenant1")
+        };
+
+        var batches = batcher.Group(envelopes).ToArray();
+        batches.Length.ShouldBe(2);
+
+        var tenant1 = batches.Single(x => x.TenantId == "tenant1");
+        tenant1.Message.ShouldBeOfType<ScoreEvent[]>().ShouldHaveSingleItem().Version.ShouldBe(3);
+        tenant1.Batch!.Length.ShouldBe(2);
+
+        var tenant2 = batches.Single(x => x.TenantId == "tenant2");
+        tenant2.Message.ShouldBeOfType<ScoreEvent[]>().ShouldHaveSingleItem().Version.ShouldBe(2);
+        tenant2.Batch!.Length.ShouldBe(1);
+    }
+
+    [Fact]
+    public void coalesce_by_rejects_a_selector_for_the_wrong_element_type()
+    {
+        var options = new BatchingOptions(typeof(ScoreEvent));
+        Should.Throw<ArgumentOutOfRangeException>(() => options.CoalesceBy((Item x) => x.Name));
+    }
+
+    [Fact]
+    public void coalesce_by_installs_a_coalescing_batcher()
+    {
+        var options = new BatchingOptions(typeof(ScoreEvent));
+        options.CoalesceBy((ScoreEvent x) => x.AggregateId);
+        options.Batcher.ShouldBeOfType<CoalescingMessageBatcher<ScoreEvent, string>>();
+    }
+}
+
+public class coalescing_batch_processing_end_to_end : IAsyncLifetime
+{
+    private IHost theHost = null!;
+
+    public async Task InitializeAsync()
+    {
+        CoalescedScoreHandler.LastBatch = null;
+
+        #region sample_batch_coalesce_by
+        theHost = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.BatchMessagesOf<ScoreEvent>(batching =>
+                {
+                    batching.BatchSize = 500;
+                    batching.TriggerTime = 1.Seconds();
+
+                    // The handler only sees ONE ScoreEvent per AggregateId (the last one wins), instead
+                    // of recomputing the same aggregate many times. Every member message still settles
+                    // with the batch.
+                    batching.CoalesceBy((ScoreEvent x) => x.AggregateId);
+                }).Sequential();
+            }).StartAsync();
+        #endregion
+    }
+
+    public async Task DisposeAsync()
+    {
+        await theHost.StopAsync();
+        theHost.Dispose();
+    }
+
+    [Fact]
+    public async Task handler_only_sees_the_coalesced_messages()
+    {
+        Func<IMessageContext, Task> publish = async c =>
+        {
+            await c.PublishAsync(new ScoreEvent("A", 1));
+            await c.PublishAsync(new ScoreEvent("A", 2));
+            await c.PublishAsync(new ScoreEvent("A", 3));
+            await c.PublishAsync(new ScoreEvent("B", 1));
+            await c.PublishAsync(new ScoreEvent("B", 2));
+        };
+
+        await theHost.TrackActivity()
+            .WaitForMessageToBeReceivedAt<ScoreEvent[]>(theHost)
+            .ExecuteAndWaitAsync(publish);
+
+        var batch = CoalescedScoreHandler.LastBatch.ShouldNotBeNull();
+        batch.Length.ShouldBe(2);
+        batch.Single(x => x.AggregateId == "A").Version.ShouldBe(3);
+        batch.Single(x => x.AggregateId == "B").Version.ShouldBe(2);
+    }
+}
+
+public class batch_context_injection : IAsyncLifetime
+{
+    private IHost theHost = null!;
+
+    public async Task InitializeAsync()
+    {
+        BatchContextItemHandler.LastContext = null;
+
+        theHost = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.BatchMessagesOf<ContextItem>(batching =>
+                {
+                    batching.BatchSize = 500;
+                    batching.TriggerTime = 1.Seconds();
+                }).Sequential();
+            }).StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await theHost.StopAsync();
+        theHost.Dispose();
+    }
+
+    [Fact]
+    public async Task batch_context_reports_the_batch_id_and_all_members()
+    {
+        Func<IMessageContext, Task> publish = async c =>
+        {
+            await c.PublishAsync(new ContextItem("one"));
+            await c.PublishAsync(new ContextItem("two"));
+            await c.PublishAsync(new ContextItem("three"));
+        };
+
+        await theHost.TrackActivity()
+            .WaitForMessageToBeReceivedAt<ContextItem[]>(theHost)
+            .ExecuteAndWaitAsync(publish);
+
+        var context = BatchContextItemHandler.LastContext.ShouldNotBeNull();
+        context.BatchId.ShouldNotBe(Guid.Empty);
+        context.Members.Count.ShouldBe(3);
+    }
+}
+
+#region sample_batch_coalesce_message
+public record ScoreEvent(string AggregateId, int Version);
+#endregion
+
+public static class CoalescedScoreHandler
+{
+    public static ScoreEvent[]? LastBatch;
+
+    public static void Handle(ScoreEvent[] events)
+    {
+        LastBatch = events;
+    }
+}
+
+public record ContextItem(string Name);
+
+public static class BatchContextItemHandler
+{
+    public static IBatchContext? LastContext;
+
+    #region sample_injecting_ibatchcontext
+    public static void Handle(ContextItem[] items, IBatchContext batch)
+    {
+        // batch.BatchId correlates all the log entries for this batch;
+        // batch.Members describes each original message that was grouped in.
+        LastContext = batch;
+    }
+    #endregion
+}

src/Wolverine/Runtime/Batching/BatchContextVariableSource.cs

@@ -0,0 +1,50 @@
+using JasperFx.CodeGeneration;
+using JasperFx.CodeGeneration.Frames;
+using JasperFx.CodeGeneration.Model;
+using JasperFx.Core.Reflection;
+
+namespace Wolverine.Runtime.Batching;
+
+/// <summary>
+/// Makes <see cref="IBatchContext"/> injectable into any message handler, built from the active
+/// <see cref="Envelope"/> the same way <c>TenantId</c> and the "now" clock are supplied. In a batched
+/// handler the envelope is the assembled batch envelope, so the context reports the batch id and its
+/// members; in an ordinary handler the membership is simply empty.
+/// </summary>
+internal class BatchContextVariableSource : IVariableSource
+{
+    public bool Matches(Type type)
+    {
+        return type == typeof(IBatchContext);
+    }
+
+    public Variable Create(Type type)
+    {
+        return new BatchContextResolutionFrame().BatchContext;
+    }
+}
+
+internal class BatchContextResolutionFrame : SyncFrame
+{
+    private Variable _envelope = null!;
+
+    public BatchContextResolutionFrame()
+    {
+        BatchContext = new Variable(typeof(IBatchContext), "batchContext", this);
+    }
+
+    public Variable BatchContext { get; }
+
+    public override IEnumerable<Variable> FindVariables(IMethodVariables chain)
+    {
+        _envelope = chain.FindVariable(typeof(Envelope));
+        yield return _envelope;
+    }
+
+    public override void GenerateCode(GeneratedMethod method, ISourceWriter writer)
+    {
+        writer.WriteLine(
+            $"var {BatchContext.Usage} = {typeof(BatchContext).FullNameInCode()}.For({_envelope.Usage});");
+        Next?.GenerateCode(method, writer);
+    }
+}

src/Wolverine/Runtime/Batching/BatchingOptions.cs

@@ -35,6 +35,33 @@ public BatchingOptions(Type elementType)
 
     public IMessageBatcher Batcher { get; set; }
 
+    /// <summary>
+    /// De-duplicate the batched messages by a key so the handler only ever sees one message per
+    /// distinct key (the last message for that key wins). This is sugar over the <see cref="Batcher"/>
+    /// seam: it installs a <c>CoalescingMessageBatcher&lt;T,TKey&gt;</c>. It only changes what the
+    /// handler sees - every member envelope still settles with the batch, exactly like a normal batch.
+    /// The lambda parameter must be explicitly typed to the batched element type so both type
+    /// arguments can be inferred, e.g. <c>CoalesceBy((RecalculateScores x) =&gt; x.AggregateId)</c>.
+    /// </summary>
+    /// <param name="keySelector">Selects the de-duplication key from each message</param>
+    /// <typeparam name="T">The batched element type (must match <see cref="ElementType"/>)</typeparam>
+    /// <typeparam name="TKey">The de-duplication key type</typeparam>
+    public void CoalesceBy<T, TKey>(Func<T, TKey> keySelector)
+    {
+        if (keySelector == null)
+        {
+            throw new ArgumentNullException(nameof(keySelector));
+        }
+
+        if (typeof(T) != ElementType)
+        {
+            throw new ArgumentOutOfRangeException(nameof(keySelector),
+                $"The coalescing key selector must accept the batched element type '{ElementType.FullNameInCode()}', but accepted '{typeof(T).FullNameInCode()}'. Use an explicitly typed lambda parameter, e.g. CoalesceBy(({ElementType.Name} x) => x.SomeKey).");
+        }
+
+        Batcher = new CoalescingMessageBatcher<T, TKey>(keySelector);
+    }
+
     /// <summary>
     /// The maximum size of the message batch. Default is 100.
     /// </summary>