Inbox e Outbox in .NET: messaggi affidabili tra database e broker

Nell’articolo sulla modularità il worker firmava un documento, portava la richiesta a Firmata e chiamava outbox.Accoda("DocumentoFirmato", …). Il messaggio per il sistema di notifiche non partiva subito: finiva in una tabella, e un secondo worker, OutboxDispatcher, lo pubblicava dopo. Restavano due domande: perché una tabella in mezzo, invece di pubblicare subito? E cosa succede quando lo stesso messaggio arriva due volte?

Le risposte sono due pattern che lavorano in coppia: l’Outbox, dal lato di chi invia, e l’Inbox, dal lato di chi riceve.

Gli esempi usano lo stesso documentale inventato, DocumentaleInventato, con EF Core e Azure Service Bus come broker, il servizio che porta i messaggi da un’applicazione all’altra. Il ragionamento è lo stesso con RabbitMQ o Kafka.

Due scritture, nessuna transazione

La versione più semplice salva lo stato e poi pubblica:

r.Stato = "Firmata";
await db.SaveChangesAsync(ct);

// e se il processo cade qui?
await bus.PubblicaAsync(Guid.NewGuid(), "DocumentoFirmato", payload, ct);

Sono due scritture su due sistemi diversi, il database e il broker. SaveChangesAsync fa il commit, e da lì il database non torna indietro, qualunque cosa succeda alla riga dopo.

Riprovare la pubblicazione aiuta solo nel secondo caso, e solo finché il processo resta in piedi: un retry in memoria muore con il processo.

Una transazione che copra entrambe le scritture non c’è. Service Bus ha le sue transazioni, ma raggruppano solo operazioni su Service Bus, come invii e completamenti: il database non ci entra, e la documentazione lo dice esplicitamente.

Se due scritture non possono stare in una transazione, la soluzione è farne una sola.

L’Outbox: il messaggio nel database

Invece di pubblicare subito, il servizio scrive il messaggio in una tabella, l’Outbox, nello stesso database dello stato. Un database, una transazione: o si salvano entrambi o nessuno dei due. Un worker separato legge la tabella e pubblica sul broker con calma.

1. La tabella

public sealed class OutboxMessage
{
    public Guid Id { get; init; } = Guid.NewGuid();
    public required string Tipo { get; init; }
    public required string Payload { get; init; }
    public DateTime CreatoIl { get; init; } = DateTime.UtcNow;
    public DateTime? InviatoIl { get; set; }
}

Payload è il contenuto del messaggio in JSON, Tipo dice come leggerlo. InviatoIl vuoto vuol dire ancora da pubblicare. L’Id lo genera il codice, non il database: è l’identità del messaggio, e viaggerà con lui fino a chi lo riceve.

Il dispatcher cercherà sempre le righe con InviatoIl vuoto, in ordine di creazione. Un indice filtrato contiene solo quelle, e resta piccolo anche quando la tabella ha milioni di righe già inviate:

modelBuilder.Entity<OutboxMessage>()
    .HasIndex(m => m.CreatoIl)
    .HasFilter("InviatoIl IS NULL");

Il filtro è SQL del database, scritto così com’è: va bene per SQL Server e SQLite, con PostgreSQL il nome della colonna va tra virgolette, "InviatoIl" IS NULL.

2. Una sola transazione

r.Stato = "Firmata";
db.Outbox.Add(new OutboxMessage
{
    Tipo = "DocumentoFirmato",
    Payload = JsonSerializer.Serialize(new { r.Id, r.DocumentoId })
});

// stato e messaggio: una transazione
await db.SaveChangesAsync(ct);

SaveChangesAsync scrive tutto quello che il DbContext ha registrato e, se il database supporta le transazioni, lo fa in una transazione sola. Se l’update della richiesta fallisce, il messaggio non viene scritto; se fallisce l’insert del messaggio, la richiesta resta InAttesa. Niente firma senza messaggio, niente messaggio senza firma.

Nell’articolo sulla modularità il caso d’uso non vedeva EF Core: chiamava outbox.Accoda. Dietro la porta c’è questo adapter, in Infrastructure:

public sealed class EfOutbox(FirmaDbContext db) : IOutbox
{
    public void Accoda(string tipo, object payload) =>
        db.Outbox.Add(new OutboxMessage
        {
            Tipo = tipo,
            Payload = JsonSerializer.Serialize(payload)
        });
}

Accoda non salva niente: aggiunge una riga al DbContext. È lo stesso DbContext del repository, perché tutti e due sono registrati come scoped, e la SaveChangesAsync del caso d’uso salva stato e messaggio insieme. Anche se payload è dichiarato object, JsonSerializer serializza il tipo reale: il tipo anonimo diventa {"Id":"…","DocumentoId":"…"}.

3. Il dispatcher

Il secondo worker è un BackgroundService come il FirmaWorker dell’articolo sulla modularità: un PeriodicTimer, uno scope nuovo a ogni giro, un try/catch che non lascia cadere l’host. A ogni giro fa questo:

var batch = await db.Outbox
    .Where(m => m.InviatoIl == null)
    .OrderBy(m => m.CreatoIl)
    .Take(20)
    .ToListAsync(ct);

foreach (var m in batch)
{
    await bus.PubblicaAsync(m.Id, m.Tipo, m.Payload, ct);
    m.InviatoIl = DateTime.UtcNow;
    await db.SaveChangesAsync(ct);
}

Prende al massimo venti messaggi, in ordine di creazione, li pubblica uno alla volta e segna InviatoIl dopo ciascuno. Se il broker non risponde, PubblicaAsync lancia un’eccezione: il messaggio resta da inviare, il catch del worker la scrive nei log e al giro dopo il dispatcher riprova. Lo stato della firma, intanto, è al sicuro: non dipende più dal broker.

Salvare dopo ogni messaggio, invece che una volta sola alla fine del batch, costa un round trip in più per messaggio, ma limita il danno se qualcosa va storto a metà. È il punto della prossima sezione.

Dall’altra parte di INotifichePublisher c’è l’adapter verso Service Bus:

public sealed class ServiceBusNotifichePublisher(ServiceBusSender sender)
    : INotifichePublisher
{
    public Task PubblicaAsync(Guid id, string tipo, string payload,
        CancellationToken ct) =>
        sender.SendMessageAsync(new ServiceBusMessage(payload)
        {
            MessageId = id.ToString(),
            Subject = tipo,
            ContentType = "application/json"
        }, ct);
}

L’Id della riga diventa il MessageId del messaggio, e il tipo va in Subject. L’Id arriva così fino al destinatario: servirà a lui.

Il prezzo dell’Outbox è un po’ di ritardo: il messaggio parte al giro successivo del dispatcher, non subito. Nell’articolo sulle trace quel tempo si vedeva nella cascata, come un buco fra la firma e la pubblicazione.

Almeno una volta

Il dispatcher pubblica, poi segna. Se cade in mezzo, dopo PubblicaAsync e prima di SaveChangesAsync, il messaggio è sul broker ma nel database risulta ancora da inviare. Al riavvio il dispatcher lo trova e lo pubblica una seconda volta, con lo stesso Id.

Invertire l’ordine non aiuta: segnare prima e pubblicare dopo riporta al problema di partenza, un messaggio segnato come inviato che non è mai partito. L’Outbox garantisce che nessun messaggio vada perso, non che arrivi una volta sola: la consegna è almeno una volta (at-least-once).

Service Bus aiuta dal lato dell’invio. Con il rilevamento dei duplicati attivo su una coda o un topic, scarta i messaggi con un MessageId già visto entro una finestra di tempo, che va da 20 secondi a 7 giorni; nel piano Basic non c’è. Il secondo invio del dispatcher viene ignorato, se arriva entro la finestra.

Ma non basta. Un dispatcher fermo più a lungo della finestra rimanda il messaggio, e questa volta passa. Soprattutto, i doppioni nascono anche dal lato di chi riceve. Di default Service Bus consegna in modalità peek-lock: blocca il messaggio per il destinatario e lo toglie dalla coda solo quando il destinatario lo completa. Se il sistema di notifiche salva la notifica e cade prima di completare, il blocco scade e il messaggio viene consegnato di nuovo. Nessun accorgimento dal lato dell’invio può evitarlo: i doppioni vanno gestiti da chi riceve.

L’Inbox: ricordarsi cosa si è già fatto

Il sistema di notifiche tiene una tabella con gli Id dei messaggi già elaborati: l’Inbox. Prima di elaborare un messaggio controlla se il suo Id c’è già. Se no, salva l’Id e l’effetto, cioè la notifica, nella stessa transazione. Una copia arrivata due volte viene scartata.

4. La tabella e il gestore

public sealed record InboxMessage(Guid Id)
{
    public DateTime ElaboratoIl { get; init; } = DateTime.UtcNow;
}

L’Id è la chiave primaria, ed è lo stesso della riga nell’Outbox, arrivato come MessageId. ElaboratoIl serve solo alla pulizia.

public sealed class DocumentoFirmatoHandler(NotificheDbContext db)
{
    public async Task GestisciAsync(Messaggio msg, CancellationToken ct)
    {
        if (await db.Inbox.AnyAsync(i => i.Id == msg.Id, ct))
            return; // già elaborato

        db.Inbox.Add(new InboxMessage(msg.Id));
        db.Notifiche.Add(Notifica.Da(msg));

        try
        {
            // registro ed effetto: una transazione
            await db.SaveChangesAsync(ct);
        }
        catch (DbUpdateException)
        {
            db.ChangeTracker.Clear();
            if (!await db.Inbox.AnyAsync(i => i.Id == msg.Id, ct))
                throw; // un errore vero, non un doppione

            // un'altra copia è arrivata prima: niente da fare
        }
    }
}

Registro ed effetto stanno nella stessa transazione. Se il processo cade prima del commit non resta né l’uno né l’altro, e alla consegna successiva il messaggio viene elaborato da capo. Se cade dopo, il registro c’è e la copia successiva viene scartata.

Il controllo iniziale da solo, però, non basta quando due copie arrivano insieme, a due istanze del servizio: tutte e due cercano l’Id, non lo trovano e vanno avanti. A quel punto decide il database. La chiave primaria non accetta due righe con lo stesso Id: la seconda SaveChangesAsync fallisce con una DbUpdateException, e non scrive nemmeno la notifica, perché è nella stessa transazione.

Il catch distingue i due casi. Se adesso l’Id c’è, l’altra copia è passata prima e non resta niente da fare. Se non c’è, l’errore è un altro e va rilanciato. ChangeTracker.Clear() toglie dal DbContext le due righe non salvate, così una SaveChangesAsync successiva nello stesso scope non prova a scriverle di nuovo.

Clear() vale solo per quel DbContext: le altre istanze del servizio e le altre richieste hanno il loro, e non se ne accorgono. Però toglie tutto quello che il DbContext sta tracciando, non solo le due righe. Qui non è un problema, perché ogni messaggio ha il suo scope e il suo DbContext, come si vede nella prossima sezione. Lo diventa se il gestore lavora su un DbContext condiviso con altre modifiche non ancora salvate: anche quelle sparirebbero senza errori. In quel caso si staccano solo le due entità del gestore:

var registro = db.Inbox.Add(new InboxMessage(msg.Id));
var notifica = db.Notifiche.Add(Notifica.Da(msg));

try
{
    // registro ed effetto: una transazione
    await db.SaveChangesAsync(ct);
}
catch (DbUpdateException)
{
    // stacca solo le due righe: il resto del DbContext non cambia
    registro.State = EntityState.Detached;
    notifica.State = EntityState.Detached;
    if (!await db.Inbox.AnyAsync(i => i.Id == msg.Id, ct))
        throw; // un errore vero, non un doppione

    // un'altra copia è arrivata prima: niente da fare
}

Add restituisce l’EntityEntry dell’entità aggiunta, e portarla a Detached la fa dimenticare al DbContext. Con un DbContext condiviso, però, la SaveChangesAsync del gestore scrive anche le modifiche degli altri, nella stessa transazione: se fallisce sulla chiave dell’Inbox, non vengono salvate neanche loro. Restano però nel DbContext, e il codice che le ha fatte può salvarle con la sua SaveChangesAsync.

5. Dal broker al gestore

Il messaggio arriva da un ServiceBusProcessor, che chiama un gestore per ogni messaggio ricevuto:

public sealed class NotificheConsumer(IServiceScopeFactory scopes)
{
    public async Task OnMessageAsync(ProcessMessageEventArgs args)
    {
        var m = args.Message;
        var msg = new Messaggio(Guid.Parse(m.MessageId),
            m.Subject, m.Body.ToString());

        await using var scope = scopes.CreateAsyncScope();
        await scope.ServiceProvider
            .GetRequiredService<DocumentoFirmatoHandler>()
            .GestisciAsync(msg, args.CancellationToken);
    }
}

Si collega con processor.ProcessMessageAsync += consumer.OnMessageAsync. Il processor vive quanto l’applicazione, come i worker, quindi anche qui il DbContext arriva da uno scope nuovo per ogni messaggio.

Il processor riceve in peek-lock e, se il gestore finisce senza eccezioni, completa il messaggio da solo: AutoCompleteMessages è true di default. Se il gestore lancia un’eccezione, il messaggio viene abbandonato e torna disponibile per un altro tentativo. L’ordine è quello giusto: prima il commit dell’Inbox, poi il completamento. Se il processo cade fra i due, il messaggio torna, e l’Inbox lo riconosce.

Insieme, l’effetto avviene una volta sola, anche se il messaggio viaggia più volte.

Quando l’Inbox non serve, e quando non basta

Non tutti gli effetti hanno bisogno di un registro. Se il gestore scrive un valore assoluto, ripeterlo non cambia niente: portare un documento a Firmato due volte lo lascia Firmato. Un’operazione così si dice idempotente, e i doppioni non le fanno danno. Una notifica inserita due volte, invece, è una notifica doppia, e un contatore incrementato due volte è sbagliato: lì serve l’Inbox.

Il limite opposto: l’Inbox protegge solo gli effetti che stanno nello stesso database, dentro la stessa transazione. Se il gestore manda una mail o chiama un’API esterna, siamo di nuovo a due scritture su due sistemi: la mail parte, il commit fallisce, e la copia successiva la rimanda. Lì serve che anche l’altro sistema riconosca i doppioni, per esempio ricevendo l’Id del messaggio come chiave di idempotenza.

La trace attraverso l’Outbox

Il messaggio parte in un altro momento e in un altro flusso rispetto alla firma, e il contesto della trace non lo segue da solo: senza accorgimenti, pubblicazione e notifiche finiscono in trace separate. Come salvarlo in una colonna TraceParent dell’Outbox e riprenderlo nel dispatcher è nell’articolo sulle trace.

In produzione

La pulizia

Le due tabelle crescono a ogni messaggio. Per l’Outbox basta cancellare i messaggi già inviati da qualche giorno:

var limite = DateTime.UtcNow.AddDays(-7);

await db.Outbox
    .Where(m => m.InviatoIl < limite)
    .ExecuteDeleteAsync(ct);

ExecuteDeleteAsync manda un solo DELETE al database, senza caricare le righe. Quelle mai inviate hanno InviatoIl vuoto, e il confronto non le prende.

Per l’Inbox la regola è più delicata: un Id va tenuto almeno finché il broker può ancora consegnare quel messaggio. Se lo si cancella prima, una consegna in ritardo passa per nuova. Contano i tentativi di consegna, la durata del blocco, la scadenza dei messaggi, e quelli rimessi in coda a mano dalla dead-letter, la coda dove finiscono i messaggi che non si riesce a elaborare, anche giorni dopo. Per questo la finestra è più lunga:

var limite = DateTime.UtcNow.AddDays(-30);

await db.Inbox
    .Where(i => i.ElaboratoIl < limite)
    .ExecuteDeleteAsync(ct);

Più istanze del dispatcher

Con due istanze del servizio partono due dispatcher, che leggono lo stesso batch e lo pubblicano due volte. Con l’Inbox dall’altra parte non è un errore, ma raddoppia i messaggi. Le strade sono due: un dispatcher solo, su un’istanza dedicata; oppure righe bloccate mentre si leggono, saltando quelle già prese da un altro dispatcher, con UPDLOCK e READPAST su SQL Server o FOR UPDATE SKIP LOCKED su PostgreSQL, in una transazione che dura quanto la pubblicazione del batch.

L’ordine e i messaggi che non partono mai

Il dispatcher pubblica in ordine di creazione e si ferma al primo errore: al giro dopo riparte da lì. È quello che serve se l’ordine conta, ma un messaggio che il broker rifiuta sempre, per esempio perché troppo grande, blocca tutti quelli dietro. Di solito si aggiunge un contatore di tentativi e, oltre una soglia, il messaggio si mette da parte e parte un allarme.

L’ordine di partenza, poi, non garantisce l’ordine di arrivo: con più destinatari in parallelo, o con un messaggio abbandonato e consegnato di nuovo, due messaggi dello stesso documento possono arrivare invertiti. Se conta, Service Bus ha le sessioni, che consegnano in ordine i messaggi con la stessa chiave.

Le librerie

Outbox, Inbox, retry, dead-letter e pulizia li fanno già alcune librerie per .NET: MassTransit, che con EF Core ha un’Outbox che fa anche da Inbox, NServiceBus e Wolverine. NServiceBus in produzione è a pagamento, con un piano gratuito per volumi piccoli; MassTransit è commerciale dalla versione 9; Wolverine è open source. Il meccanismo conviene conoscerlo anche quando lo fa una libreria: le garanzie sono quelle viste qui, e anche i limiti.

In breve

1. Salvare lo stato e pubblicare sono due scritture, e nessuna transazione le copre entrambe.
2. L’Outbox mette il messaggio nello stesso database, nella stessa SaveChangesAsync dello stato.
3. Il dispatcher pubblica dopo, e segna ogni messaggio appena inviato.
4. Un messaggio può partire due volte: la consegna è almeno una volta.
5. L’Id del messaggio viaggia fino a chi riceve.
6. L’Inbox salva l’Id e l’effetto nella stessa transazione, e la chiave primaria ferma le copie in parallelo.
7. Il messaggio si completa solo dopo il commit.
8. Pulizia, più istanze e messaggi bloccati si decidono prima di andare in produzione.

Codice semplificato a scopo di esempio: nomi e struttura servono a dare l'idea, non sono da copiare così.