IWireTap not invoked for messages sent via DurabilityAgent (store-and-forward pattern)

[Maflend]

Describe the bug
IWireTap.RecordSuccessAsync() is not called when messages are sent via the DurabilityAgent (store-and-forward pattern).
WireTap only fires when using IDbContextOutbox.FlushOutgoingMessagesAsync() or IDbContextOutbox.SaveChangesAndFlushMessagesAsync().
When using the transactional outbox pattern without explicit flush:

await outbox.PublishAsync(new MyMessage(...));
await db.SaveChangesAsync();
// DurabilityAgent will pick up and send later

The message is persisted to wolverine_outgoing_envelopes and eventually sent by the DurabilityAgent, but IWireTap is never invoked for this send.

To Reproduce

1. Register an IWireTap implementation in DI
2. Configure an endpoint with .UseWireTap()
3. Use the outbox pattern without FlushOutgoingMessagesAsync():

await outbox.PublishAsync(new MyMessage(...));
await db.SaveChangesAsync();

4. Wait for DurabilityAgent to send the message
5. Observe that IWireTap.RecordSuccessAsync() was never called

Expected behavior
IWireTap.RecordSuccessAsync() should be called for every message that is successfully sent, regardless of whether it was sent inline (via FlushOutgoingMessagesAsync()) or deferred (via DurabilityAgent).

Additional context
This behavior is not documented and can be confusing for users implementing audit/logging via IWireTap. The store-and-forward pattern (relying on DurabilityAgent without explicit flush) is a valid and supported Wolverine pattern, but WireTap silently does not work in this mode.
Workarounds:

1. Always use FlushOutgoingMessagesAsync() — but this changes the message delivery semantics
2. Implement audit at the handler level instead of transport level — but this misses the "wire tap" concept

[jeremydmiller]

@Maflend I think you're a little off. The DurabilityAgent has nothing to do with send and forward mechanics. Do you mean messages that are recovered by the DurabilityAgent, or scheduled messages? And the wire tap is only for received messages.

And have we had the conversation yet where I suggest that maybe you have fewer issues if you use Wolverine.Http and try to be more idiomatic rather than trying to use the outbox outside like that?

I think I need much more information from you here. Do you have any kind of reproduction code? I don't think the wire tap should fire at all in any of the code you pasted up above.

I'm moving this to a discussion as I'm not sure there's anything really wrong yet.

[Maflend]

"And the wire tap is only for received messages." not true. It is also intended for published messages. https://wolverinefx.net/guide/logging.html#enabling-wire-taps-on-endpoints

"Do you mean messages that are recovered by the DurabilityAgent, or scheduled messages?"
I meant messages that are recovered by the DurabilityAgent.
Okay, I realized that the task of the DurabilityAgent in this case is to send a message to the broker. It does not invoke registered WireTaps.

You mentioned "or scheduled messages" and I went to check.

It turns out that "IDbContextOutbox.ScheduleAsync" also does not use WireTap at the time the message is published.
Here is a sample code:

builder.Host.UseWolverine(opts =>
{
    opts.UseRuntimeCompilation();
    opts.ServiceLocationPolicy = ServiceLocationPolicy.AlwaysAllowed;
    opts.PersistMessagesWithPostgresql(connStr, "wolverine");
    opts.UseEntityFrameworkCoreTransactions();
    
    opts.Services.AddKeyedSingleton<IWireTap, AuditWireTap>("audit");
    
    opts.Policies.AllListeners(x => x.UseWireTap("audit"));
    opts.Policies.AllSenders(x => x.UseWireTap("audit"));
    
    opts.UseKafka(kafkaServers).AutoProvision();
    opts.EnvelopeIdGeneration = EnvelopeIdGeneration.GuidV7;
    
    opts.PublishMessage<MyMessage>()
        .ToKafkaTopic("wiretapaudit")
        .UseDurableOutbox();
    
    opts.ListenToKafkaTopic("wiretapaudit")
        .ConfigureConsumer(c => c.GroupId = "wiretapaudit-consumer")
        .UseDurableInbox();
});

var app = builder.Build();

app.MapGet("/schedule", async (IDbContextOutbox outbox, AppDbContext db) =>
{
    outbox.Enroll(db);
    await outbox.ScheduleAsync(new MyMessage("Message", Guid.NewGuid()), TimeSpan.FromSeconds(10));
    await outbox.SaveChangesAndFlushMessagesAsync();
});

app.MapGet("/publish", async (IDbContextOutbox outbox, AppDbContext db) =>
{
    outbox.Enroll(db);
    await outbox.PublishAsync(new MyMessage("Message", Guid.NewGuid()));
    await outbox.SaveChangesAndFlushMessagesAsync();
});

public class AuditWireTap(ILogger<AuditWireTap> logger) : IWireTap
{
    public ValueTask RecordSuccessAsync(Envelope envelope)
    {
        var status = envelope.Listener is not null ? "received" : "sent";
        logger.LogWarning($"RecordSuccessAsync. Status = {status}");
        return ValueTask.CompletedTask;
    }

    public ValueTask RecordFailureAsync(Envelope envelope, Exception exception)
    {
        const string status = "failed";
        logger.LogWarning($"RecordFailureAsync. Status = {status}");
        return ValueTask.CompletedTask;
    }
}

By calling /publish in the logs, I see two messages:

1. RecordSuccessAsync. Status = sent
2. RecordSuccessAsync. Status = received

By calling /schedule in the logs, I see one message:
2. RecordSuccessAsync. Status = received

And that seems to me to be a problem. Because AuditWireTap did not work at the time of posting the message to kafka.
I use AuditWireTap to listen to messages being sent and processed.

Hence the question, is it accurate that "outbox.ScheduleAsync" should not call WireTap when publishing a message to the broker?
You write (https://wolverinefx.net/guide/logging.html#auditing-and-compliance-considerations) that WireTap can be used for auditing purposes, but as you can see, WireTap will not be called for a message sent via ScheduleAsync.

[jeremydmiller]

@Maflend Point blank, you ask for way too much help and you're taking more of my time than I would like. I'm happy to take a pull request, or for the 4th or 5th time, I'm going to ask you to look into a support contract. For whatever reasons, you're one of those people who just stumble into much more trouble than other people.

And lastly, are you sure the wire tap isn't triggering when the message is actually delivered? I mean, when it's actually sent?

[Maflend]

Yes, I'm really sure that WireTap is not called for Schedule. When publishing a deferred message to kafka, WireTap does not work.

Full source code:

using Humanizer;
using JasperFx.CodeGeneration.Model;
using Microsoft.EntityFrameworkCore;
using Wolverine;
using Wolverine.EntityFrameworkCore;
using Wolverine.Kafka;
using Wolverine.Postgresql;

var builder = WebApplication.CreateBuilder(args);
var kafkaServers = "localhost:9092";
var connStr = "Host=localhost;Port=5432;Database=wolverine_demo;Username=postgres;Password=postgres;IncludeErrorDetail=true";

builder.Services.AddDbContext<AppDbContext>((sp, opts) =>
{
    opts.UseNpgsql(connStr);
});

builder.Host.UseWolverine(opts =>
{
    opts.UseRuntimeCompilation();
    opts.ServiceLocationPolicy = ServiceLocationPolicy.AlwaysAllowed;
    opts.PersistMessagesWithPostgresql(connStr, "wolverine");
    opts.UseEntityFrameworkCoreTransactions();

    opts.Services.AddKeyedSingleton<IWireTap, AuditWireTap>("audit");
    opts.Policies.AllListeners(x => x.UseWireTap("audit"));
    opts.Policies.AllSenders(x => x.UseWireTap("audit"));
    
    opts.UseKafka(kafkaServers).AutoProvision();
    
    opts.PublishMessage<MyMessage>()
        .ToKafkaTopic("wire-tap-audit-mymessage")
        .UseDurableOutbox();
    
    opts.ListenToKafkaTopic("wire-tap-audit-mymessage").UseDurableInbox();
});

var app = builder.Build();

app.MapGet("/schedule", async (IDbContextOutbox outbox, AppDbContext db) =>
{
    outbox.Enroll(db);
    await outbox.ScheduleAsync(new MyMessage("Message", Guid.NewGuid()), TimeSpan.FromSeconds(10));
    await outbox.SaveChangesAndFlushMessagesAsync();
});

app.MapGet("/publish", async (IDbContextOutbox outbox, AppDbContext db) =>
{
    outbox.Enroll(db);
    await outbox.PublishAsync(new MyMessage("Message", Guid.NewGuid()));
    await outbox.SaveChangesAndFlushMessagesAsync();
});

app.Run();

public class AuditWireTap(ILogger<AuditWireTap> logger) : IWireTap
{
    public ValueTask RecordSuccessAsync(Envelope envelope)
    {
        var message = envelope.Listener is not null ? $"Message was received. {envelope.Id}" : $"Message was sent {envelope.Id}";
        logger.LogWarning($"AuditWireTap. {message}");
        return ValueTask.CompletedTask;
    }

    public ValueTask RecordFailureAsync(Envelope envelope, Exception exception) => ValueTask.CompletedTask;
}

public record MyMessage(string Name, Guid Id);
public class MyMessageConsumer
{
    public async Task ConsumeAsync(MyMessage msg, IMessageContext ctx)
    {
        await Task.Delay(5.Seconds());
    }
}

public class AppDbContext : DbContext
{
    public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }
    
    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.MapWolverineEnvelopeStorage("wolverine");
        base.OnModelCreating(modelBuilder);
    }
}

Screenshots of the execution:

1. Called /publish
   We see two logs.

1. Called /schedule
   We see one log.

I need to audit sent and processed messages. The documentation says that WireTap can be used for this purpose. But as we can see, WireTap does not work for scheduled messages when they are sent to Kafka.
@jeremydmiller

[jeremydmiller]

Already solved in the most recent version.