Logical Message Deduplication with Wolverine

Wolverine has long had support for message idempotency using Wolverine’s transactional inbox to track messages that have already been handled, but that usage only applies to Wolverine metadata and does nothing from an upstream sender (or let’s be honest, your code) accidentally sending the same logical message.

Recent versions of Wolverine added “logical message deduplication” as well to allow you to specify business logic concerns as the identity for another level of protection.

For example:

The operator clicked Rebuild twice. The console republished the command after a timeout. A scheduling agent pre-published tonight’s 03:00 occurrence yesterday, and the scheduler published it again on the night. Should the projection rebuild four times?

Each of those is a different delivery of the same intent, so each carries a different Envelope.Id and every one of them gets through. What is needed is an identity for the intent, and that is what DeduplicationId is.

We probably should have added this feature ages ago, but we did so in 6.31 to get ready for Wolverine to have a first class (but minimal!) chron message scheduler and use the deduplication id as a way of preventing multiple executions in a chaotic world. We know users are already taking advantage of this because JasperFx has already done some enhancements and optimizations for one of our clients.

You do need to explicitly enable this (because it potentially adds database tables or columns and we do not make mandatory database changes outside of major version releases. Backwards compatibility is a pain, but it’s you know, kind of important).:

			
using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.PersistMessagesWithPostgresql("connection string");
        // Opt in to logical message deduplication. This provisions a new
        // "wolverine_deduplication" table -- nothing else about your message
        // storage changes, and leaving this off means no schema migration at all
        opts.Durability.EnableMessageDeduplication = true;
        // How long a logical id is honoured before the reaper removes it.
        // The default is 24 hours. This IS the guarantee, so size it against
        // how long a duplicate could plausibly arrive
        opts.Durability.DeduplicationWindow = 24.Hours();
    }).StartAsync();

		

And you can turn it on for a single handler like this:

			
public static class RebuildProjectionHandler
{
    // Wolverine will refuse to run this handler twice for the same
    // Envelope.DeduplicationId within the deduplication window
    [Deduplicated]
    public static void Handle(RebuildProjection command)
    {
        // rebuild the projection...
    }
}

		

and on the publishing side you can programmatically pass along the logical deduplication key:

			
public static ValueTask ScheduleNightlyRebuild(IMessageBus bus, string projectionName, DateTimeOffset occurrence)
{
    return bus.PublishAsync(new RebuildProjection(projectionName, occurrence), new DeliveryOptions
    {
        // The logical identity of the WORK, not of this particular delivery.
        // An operator double-click, a console republish, and an agent that
        // pre-published this occurrence yesterday all produce this same id
        DeduplicationId = $"{projectionName}|{occurrence:O}"
    });
}

		

Don’t worry, it’s Wolverine, so there are of course some magical policy ways to shorten the configuration.

A logical id is a string so it can be legible to users.

The second message with that id never reaches your handler. It is discarded, acknowledged to the broker, and logged at Information — a duplicate that vanished without a trace would be indistinguishable from a message that was lost.

Where the id comes from

By default a message handler reads Envelope.DeduplicationId. You can point at a member of the message itself instead, so publishers do not have to set DeliveryOptions:

			
public static class CreateOrderHandler
{
    // Derive the logical id from a member of the message itself rather than
    // asking every publisher to set DeliveryOptions
    [Deduplicated(ValueSource.InputMember, nameof(CreateOrder.Sku))]
    public static void Handle(CreateOrder command)
    {
        // create the order...
    }
}

		

ValueSource.Header reads an envelope header, and ValueSource.Anything uses the chain type’s natural default.

Deriving the id on the publishing side 6.31

Everything above is the receiving half. On the publishing side, asking every call site to remember DeliveryOptions.DeduplicationId is exactly the kind of repetition that eventually gets forgotten at one call site and silently un-protects a message. So a message type can declare its own logical identity once, the same way it can already declare a topic name with [Topic] or a saga id with [SagaIdentity]:

			
// The message type declares its own logical identity once, and every publisher
// gets it -- no DeliveryOptions at any call site
public record ArchiveInvoice([property: DeduplicationIdentity] string InvoiceNumber, DateOnly AsOf);

or, for a contract whose members you cannot decorate, name the member from the type:

			
// The same thing for a contract whose members you cannot decorate
[DeduplicationIdentity(nameof(ReceiveShipment.ShipmentId))]
public record ReceiveShipment(Guid ShipmentId, string Warehouse);

Either form is applied as an IEnvelopeRule when the message is routed, so it reaches every transport, the local queues, and the outbox alike. Non-string members are converted with ToString().

Or if you need to, or just want to do so explicitly, you also have this syntax:

			
using var host = await Host.CreateDefaultBuilder()
    .UseWolverine(opts =>
    {
        opts.PersistMessagesWithPostgresql("connection string");
        opts.Durability.EnableMessageDeduplication = true;
        // Compose the logical id from more than one member, or from anything
        // else you can reach from the message
        opts.MessageDeduplication.ByMessage<RebuildProjection>(
            x => $"{x.ProjectionName}|{x.OccurrenceUtc:O}");
        // Or, for generated message types you can neither decorate nor be
        // bothered writing a lambda for, use the first member that matches
        // one of these names
        opts.MessageDeduplication.ByMemberNamed("IdempotencyKey", "DeduplicationId");
        // Same thing as ByMessage<T>(), reached through the message type policies
        opts.Policies.ForMessagesOfType<CreateOrder>()
            .DeduplicateBy(x => $"{x.Sku}|{x.Quantity}");
    }).StartAsync();

		