Il pattern Outbox transazionale in Go con PostgreSQL

Due scritture che dovrebbero riuscire insieme finiscono per fallire separatamente. Il tuo servizio ordini salva l’ordine nel database, quindi pubblica un evento order.created su un message broker.

Queste due operazioni vengono eseguite una dopo l’altra.

Nel frattempo, qualcosa va storto: il broker è offline, il network va in timeout, il processo si riavvia o il container viene eliminato. La scrittura nel database è riuscita. La pubblicazione non lo è. Il servizio downstream che deve conoscere il nuovo ordine non viene mai informato. Nessuno se ne accorge finché un cliente non chiama.

Questo è il problema del dual-write ed è una delle cause più comuni di perdita silente di dati nei sistemi distribuiti. Il pattern Transactional Outbox è la soluzione standard.

Il problema del dual-write

La modalità di guasto è facile da analizzare una volta individuata:

BEGIN;
  INSERT INTO orders ...   -- succeeds
COMMIT;

PUBLISH order.created ...  -- fails, crashes, or is never reached

Il database e il message broker non condividono un confine di transazione. Non esiste un rollback che copra entrambi. Ogni servizio che esegue save -> publish in sequenza presenta questa lacuna. Il pattern si manifesta in molte forme:

• db.Save(order) seguito da events.Publish(OrderCreated{...})
• Handler HTTP che commita una transazione e poi chiama un webhook esterno
• Worker che elabora un record da una coda e scrive i risultati in un’altra

L’esito in tutti i casi è lo stesso: un lato ha successo mentre l’altro fallisce, e il sistema finisce in uno stato invisibile al monitoraggio perché entrambe le operazioni individuali hanno restituito successo in qualche momento.

Un ciclo di retry non risolve il problema. Riprovare la pubblicazione dopo il commit del database funziona solo se il retry stesso è affidabile – il che richiede proprio la garanzia di durabilità che non hai.

Cosa fa il pattern Transactional Outbox

Il pattern Outbox elimina la lacuna rimuovendo completamente la pubblicazione diretta. Invece di chiamare il broker all’interno della tua logica di business, scrivi un record di evento in una tabella outbox nella stessa transazione del database dei dati aziendali. Un processo in background separato – il relay – legge dalla tabella outbox e pubblica sul broker.

BEGIN;
  INSERT INTO orders ...         -- business data
  INSERT INTO outbox_events ...  -- event record
COMMIT;

-- Relay process (separately):
SELECT ... FROM outbox_events FOR UPDATE SKIP LOCKED;
PUBLISH order.created ...
UPDATE outbox_events SET processed_at = NOW() WHERE id = $1;

Entrambe le scritture riescono o entrambe falliscono. La garanzia di transazione che già hai da PostgreSQL ora copre anche il record dell’evento. Il relay può ritentare la pubblicazione tutte le volte necessarie perché l’evento risiede in storage durabile. Se il relay si arresta in modo anomalo, si riavvia e ritenta. L’esito peggiore è che l’evento venga pubblicato più di una volta – il che viene gestito rendendo i consumatori idempotenti (vedi Idempotency in Distributed Systems).

Schema PostgreSQL per la tabella outbox

Lo schema è intenzionalmente semplice:

CREATE TABLE outbox_events (
    id             UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id   VARCHAR(100) NOT NULL,
    event_type     VARCHAR(100) NOT NULL,
    payload        JSONB        NOT NULL,
    attempts       INT          NOT NULL DEFAULT 0,
    created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    processed_at   TIMESTAMPTZ
);

-- Partial index: only indexes unprocessed rows, stays small as rows are marked done
CREATE INDEX idx_outbox_unprocessed
    ON outbox_events (created_at)
    WHERE processed_at IS NULL;

L’indice parziale su created_at WHERE processed_at IS NULL è importante. Senza di esso, l’indice cresce con ogni evento mai scritto e la query di polling del relay diventa più lenta nel tempo. Con esso, l’indice copre solo le righe pendenti, che in stato stabile sono un insieme piccolo e limitato indipendentemente da quanti eventi siano stati pubblicati.

Scelte chiave dei campi:

• aggregate_type e aggregate_id descrivono a quale entità appartiene l’evento. Utili per garanzie di ordinamento e routing.
• event_type è il nome dell’evento che i consumatori si aspettano.
• payload JSONB memorizza il corpo dell’evento. Usa JSONB invece di TEXT così puoi interrogarlo se necessario.
• attempts tiene traccia di quante volte il relay ha tentato di pubblicare questa riga. Usato per limiti di retry e gestione delle dead-letter.
• processed_at è NULL per le righe pendenti e impostato quando il relay pubblica con successo.

Scrittura dei dati aziendali e dell’evento outbox in una singola transazione

La logica di business scrive entrambi i record all’interno di una singola chiamata BeginTx / Commit. Non c’è alcuna chiamata di publish qui – solo scritture nel database.

type OrderService struct {
    db *sql.DB
}

func (s *OrderService) CreateOrder(ctx context.Context, order Order) error {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback()

    if _, err := tx.ExecContext(ctx, `
        INSERT INTO orders (id, customer_id, total, created_at)
        VALUES ($1, $2, $3, NOW())
    `, order.ID, order.CustomerID, order.Total); err != nil {
        return fmt.Errorf("insert order: %w", err)
    }

    payload, err := json.Marshal(map[string]any{
        "order_id":    order.ID,
        "customer_id": order.CustomerID,
        "total":       order.Total,
    })
    if err != nil {
        return fmt.Errorf("marshal payload: %w", err)
    }

    if _, err := tx.ExecContext(ctx, `
        INSERT INTO outbox_events
            (aggregate_type, aggregate_id, event_type, payload)
        VALUES ($1, $2, $3, $4)
    `, "order", order.ID, "order.created", payload); err != nil {
        return fmt.Errorf("insert outbox event: %w", err)
    }

    return tx.Commit()
}

Se tx.Commit() fallisce, né la riga dell’ordine né quella dell’outbox vengono memorizzate. Se ha successo, entrambe sono garantite essere nel database. Il relay può pubblicare l’evento in qualsiasi momento dopo – immediatamente, in un secondo, o dopo il riavvio del relay seguito da un crash.

Questa è l’unica modifica al codice richiesta nel tuo layer di business. Il resto del pattern vive nel relay.

Implementazione Go del relay

Il relay è un worker in background che esegue il polling sulla tabella outbox su un timer. Recupera un batch di righe non elaborate, pubblica ciascuna e la segna come completata. Tienilo nello stesso binario della tua applicazione o eseguilo come processo separato – entrambi funzionano, ma lo stesso binario è più semplice da gestire.

type OutboxRelay struct {
    db          *sql.DB
    publisher   Publisher
    logger      *slog.Logger
    batchSize   int
    pollInterval time.Duration
    maxAttempts  int
}

func (r *OutboxRelay) Run(ctx context.Context) error {
    ticker := time.NewTicker(r.pollInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-ticker.C:
            if err := r.processBatch(ctx); err != nil {
                r.logger.Error("outbox relay batch failed", "err", err)
            }
        }
    }
}

Il relay rispetta la cancellazione del contesto, il che lo rende facile da integrare con lo shutdown graceful. Per un trattamento dettagliato della durata del contesto e dei pattern di cancellazione, vedi Go context.Context Done Right.

FOR UPDATE SKIP LOCKED: il pattern del worker concurrently

La funzione processBatch usa FOR UPDATE SKIP LOCKED per gestire in modo sicuro i worker del relay concorrenti:

func (r *OutboxRelay) processBatch(ctx context.Context) error {
    tx, err := r.db.BeginTx(ctx, nil)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback()

    rows, err := tx.QueryContext(ctx, `
        SELECT id, aggregate_type, aggregate_id, event_type, payload
        FROM outbox_events
        WHERE processed_at IS NULL
          AND attempts < $1
        ORDER BY created_at
        LIMIT $2
        FOR UPDATE SKIP LOCKED
    `, r.maxAttempts, r.batchSize)
    if err != nil {
        return fmt.Errorf("query outbox: %w", err)
    }
    defer rows.Close()

    type row struct {
        id            string
        aggregateType string
        aggregateID   string
        eventType     string
        payload       json.RawMessage
    }

    var batch []row
    for rows.Next() {
        var e row
        if err := rows.Scan(
            &e.id, &e.aggregateType, &e.aggregateID, &e.eventType, &e.payload,
        ); err != nil {
            return fmt.Errorf("scan row: %w", err)
        }
        batch = append(batch, e)
    }
    if err := rows.Err(); err != nil {
        return err
    }

    for _, e := range batch {
        if err := r.publisher.Publish(ctx, e.eventType, e.aggregateID, e.payload); err != nil {
            r.logger.Error("publish failed", "event_id", e.id, "err", err)
            if _, err := tx.ExecContext(ctx,
                `UPDATE outbox_events SET attempts = attempts + 1 WHERE id = $1`, e.id,
            ); err != nil {
                r.logger.Error("increment attempts failed", "event_id", e.id, "err", err)
            }
            continue
        }

        if _, err := tx.ExecContext(ctx,
            `UPDATE outbox_events SET processed_at = NOW() WHERE id = $1`, e.id,
        ); err != nil {
            return fmt.Errorf("mark processed: %w", err)
        }
    }

    return tx.Commit()
}

FOR UPDATE SKIP LOCKED fa due cose. Primo, FOR UPDATE blocca le righe selezionate per la durata della transazione, impedendo ad altre transazioni di selezionarle. Secondo, SKIP LOCKED significa che se una riga è già bloccata da un’altra transazione, la query la salta invece di attendere. Il risultato è che più worker del relay possono essere eseguiti in parallelo e ciascuno raccoglierà un sottoinsieme non sovrapposto di righe.

Senza SKIP LOCKED, un secondo worker bloccherebbe fino a quando la prima transazione non commita prima di vedere le stesse righe – momento in cui sarebbero già segnate come completate. Con SKIP LOCKED, il secondo worker raccoglie immediatamente righe diverse invece di attendere, garantendo una scalabilità orizzontale sicura.

Nota la separazione tra scan e publish nel codice sopra: tutte le righe vengono scansionate in una slice prima che inizi il ciclo di publish. Questo evita di mantenere un cursore *sql.Rows aperto attraverso chiamate di rete al broker, il che manterrebbe la transazione aperta più a tempo del necessario.

Idempotenza e deduplicazione

Il relay pubblica almeno una volta. Se pubblica un evento e poi crasha prima di commitare l’aggiornamento processed_at, pubblicherà lo stesso evento di nuovo al riavvio. Questo è inevitabile – la consegna esattamente-once tra un database e un message broker senza un coordinatore di transazioni distribuite richiede questo compromesso.

I consumatori devono essere idempotenti. L’approccio più semplice è tenere traccia degli ID degli eventi elaborati in una tabella processed_events:

CREATE TABLE processed_events (
    event_id   UUID PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

func (h *OrderHandler) HandleOrderCreated(ctx context.Context, eventID string, payload []byte) error {
    // Deduplicate using the event ID as the natural key
    _, err := h.db.ExecContext(ctx, `
        INSERT INTO processed_events (event_id) VALUES ($1)
        ON CONFLICT (event_id) DO NOTHING
    `, eventID)
    if err != nil {
        return fmt.Errorf("dedup check: %w", err)
    }

    // Check if the insert actually happened (1 row) or was a no-op (0 rows)
    // A simpler approach: use RETURNING or check rows affected
    // If 0 rows affected, this is a duplicate -- skip it
    ...
}

In pratica, molti team si affidano ai propri header di deduplicazione del broker (come il campo key di Kafka per gli argomenti log-compacted, o l’header message-id di RabbitMQ) e trattano la deduplicazione a livello di database come un fallback. Entrambi sono livelli validi da applicare.

Includi l’id dell’evento outbox (un UUID) nel messaggio pubblicato come chiave di deduplicazione. I consumatori possono quindi usarlo indipendentemente dal meccanismo di deduplicazione che preferiscono.

Policy di retry e messaggi velenosi

La colonna attempts guida la policy di retry. Il relay salta le righe dove attempts >= maxAttempts e tratta quelle righe come dead-letter. Un processo separato o un alert dell’operatore le gestisce.

Una vista semplice per le dead-letter:

CREATE VIEW outbox_dead_letters AS
SELECT *
FROM outbox_events
WHERE attempts >= 5
  AND processed_at IS NULL
ORDER BY created_at;

Una buona policy di retry per la produzione:

• Imposta maxAttempts a 5-10 a seconda di quanto sono costosi i retry.
• Considera l’exponential backoff: includi una colonna retry_after e salta le righe dove retry_after > NOW().
• Imposta un alert su COUNT(*) FROM outbox_dead_letters che supera una soglia.
• Fornisci un percorso di retry manuale: un endpoint admin o uno script che resetta attempts = 0 e retry_after = NULL per righe specifiche.

I messaggi velenosi – righe che falliscono costantemente a causa di un bug nel consumatore o di una mismatch di schema – non dovrebbero bloccare i messaggi sani. Poiché il relay elabora un batch per tick e segna i fallimenti con un incremento del tentativo invece di rimuoverli dalla coda, le righe sane procedono normalmente mentre quelle velenose accumulano tentativi fino a raggiungere la soglia della dead-letter. Questa vista outbox_dead_letters è una versione lato database dello stesso pattern che i broker-native dead-letter queues implementano – quarantena dopo una soglia, alert sul volume e richiede una decisione deliberata prima del replay.

Ordinamento degli eventi e partizionamento

La query di polling ordina per created_at, che fornisce un ordinamento first-in-first-out all’interno di un batch. Per la maggior parte dei casi d’uso questo è sufficiente. Quando l’ordinamento per entità è strettamente necessario – ad esempio, assicurarsi che order.updated non venga mai pubblicato prima di order.created per lo stesso ordine – hai bisogno di un ordinamento per aggregate.

Aggiungi aggregate_id alla clausola ORDER BY e usalo come chiave del messaggio quando pubblichi su un argomento partizionato come Apache Kafka. Kafka instrada tutti i messaggi con la stessa chiave verso la stessa partizione, e le partizioni vengono consumate in ordine. Questo ti fornisce garanzie di ordinamento per aggregate senza ordinamento globale, che richiederebbero una singola istanza del relay.

ORDER BY aggregate_id, created_at

Per i broker che non supportano l’ordinamento partizionato (come le code AMQP di base), il relay a istanza singola o i controlli di ordinamento a livello di applicazione nel consumatore sono le alternative pratiche.

Ridurre la latenza di polling con LISTEN/NOTIFY

Un intervallo di polling di un secondo significa una latenza media degli eventi di 500 millisecondi. Per la maggior parte dei carichi di lavoro questo va bene. Per i casi in cui hai bisogno di una latenza quasi zero, il meccanismo LISTEN/NOTIFY di PostgreSQL permette al relay di svegliarsi immediatamente quando viene inserita una nuova riga outbox.

Aggiungi un trigger alla tabella outbox:

CREATE OR REPLACE FUNCTION notify_outbox_insert() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify('outbox_event', NEW.id::text);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER outbox_insert_notify
AFTER INSERT ON outbox_events
FOR EACH ROW EXECUTE FUNCTION notify_outbox_insert();

Nel relay, ascolta sul canale e svegliati sugli notifiche mentre mantieni il fallback al polling periodico:

func (r *OutboxRelay) Run(ctx context.Context) error {
    listener := pq.NewListener(r.dsn, 10*time.Second, time.Minute, nil)
    defer listener.Close()

    if err := listener.Listen("outbox_event"); err != nil {
        return fmt.Errorf("listen: %w", err)
    }

    ticker := time.NewTicker(5 * time.Second) // fallback poll
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-listener.Notify:
            if err := r.processBatch(ctx); err != nil {
                r.logger.Error("outbox batch failed (notify)", "err", err)
            }
        case <-ticker.C:
            if err := r.processBatch(ctx); err != nil {
                r.logger.Error("outbox batch failed (poll)", "err", err)
            }
        }
    }
}

Il ticker di fallback gestisce eventuali notifiche perse durante un riavvio del relay o un intoppo di rete. Mantieni l’intervallo di fallback a pochi secondi invece che millisecondi – il suo lavoro è il recupero, non la bassa latenza.

Osservabilità: metriche, log e alert

L’outbox è infrastruttura. Trattala come infrastruttura e strumentala di conseguenza.

Metriche chiave:

var (
    outboxPublished = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "outbox_events_published_total",
        Help: "Total outbox events successfully published.",
    })
    outboxFailed = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "outbox_events_failed_total",
        Help: "Total outbox publish failures by event type.",
    }, []string{"event_type"})
    outboxPending = prometheus.NewGauge(prometheus.GaugeOpts{
        Name: "outbox_events_pending",
        Help: "Current number of unprocessed outbox events.",
    })
    outboxBatchDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
        Name:    "outbox_batch_duration_seconds",
        Help:    "Duration of each outbox processing batch.",
        Buckets: prometheus.DefBuckets,
    })
)

Aggiornamento Gauge: esegui una query periodica per mantenere outbox_events_pending accurata:

SELECT COUNT(*) FROM outbox_events WHERE processed_at IS NULL;

Soglie di alert da considerare:

• outbox_events_pending > 1000 per più di due minuti: il relay è in ritardo o bloccato.
• outbox_events_pending in crescita monotona: il broker è offline o il relay è crashato.
• Conteggio dead-letter non zero: bug di schema o del consumatore necessita di indagine.
• outbox_batch_duration_seconds p95 > 5s: il database è lento o la dimensione del batch è troppo grande.

Campi log strutturati: includi event_id, event_type, aggregate_id e attempt in ogni riga di log dal relay. Questi campi ti permettono di correlare una pubblicazione fallita con la specifica riga outbox e il trace del consumatore downstream.

Outbox vs. coda diretta vs. saga

Il pattern outbox non è lo strumento giusto per ogni problema di coordinamento. Ecco il confronto:

Il pattern outbox garantisce che un singolo servizio emetta in modo affidabile eventi che riflettono i suoi stessi cambiamenti di stato. Non coordina i cambiamenti di stato attraverso più servizi – questo è ciò per cui è lo Saga pattern. La scelta del broker – sia RabbitMQ, SQS, o Kafka – è indipendente dal pattern outbox stesso; il relay pubblica sul broker che il tuo sistema usa.

Se stai costruendo una saga, il pattern outbox è comunque utile: ogni partecipante nella saga scrive il suo cambiamento di stato locale e il suo evento saga in una transazione usando l’outbox, poi l’orchestratore o la choreografia della saga legge quegli eventi in modo affidabile.

CDC basato su WAL come relay alternativo

Invece di fare polling, puoi seguire il Write-Ahead Log (WAL) di PostgreSQL e leggere le inserzioni outbox direttamente dallo stream di replication. Strumenti come Debezium fanno questo. I vantaggi sono una latenza inferiore e nessuna pressione sui lock sulla tabella outbox. Gli svantaggi sono la complessità operativa, un slot di replication PostgreSQL dedicato e un servizio esterno da eseguire e monitorare.

Per la maggior parte dei team, il relay di polling descritto sopra è il punto di partenza giusto. Il tailing del WAL ha senso quando hai alti tassi di inserzione outbox (decine di migliaia al secondo), hai bisogno di latenza degli eventi inferiore a 100ms, o stai già eseguendo Debezium per altre esigenze di change-capture.

Integrazione con sqlc

Se usi sqlc per codice database Go type-safe, le query outbox si adattano naturalmente:

-- name: InsertOutboxEvent :exec
INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload)
VALUES (@aggregate_type, @aggregate_id, @event_type, @payload);

-- name: FetchOutboxBatch :many
SELECT id, aggregate_type, aggregate_id, event_type, payload
FROM outbox_events
WHERE processed_at IS NULL
  AND attempts < @max_attempts
ORDER BY created_at
LIMIT @batch_size
FOR UPDATE SKIP LOCKED;

-- name: MarkOutboxProcessed :exec
UPDATE outbox_events SET processed_at = NOW() WHERE id = @id;

-- name: IncrementOutboxAttempts :exec
UPDATE outbox_events SET attempts = attempts + 1 WHERE id = @id;

-- name: OutboxPendingCount :one
SELECT COUNT(*) FROM outbox_events WHERE processed_at IS NULL;

sqlc genera funzioni type-safe per ogni query, il che evita errori di interpolazione di stringhe e mantiene la logica delle query outbox co-localizzata con il resto del tuo layer di accesso al database.

Checklist per la produzione

Usa questa prima di rilasciare un’implementazione outbox:

Database

• Outbox table has the partial index on created_at WHERE processed_at IS NULL
• attempts column present with a default of 0
• Dead-letter view or query defined
• Old processed rows are periodically archived or deleted (a nightly cleanup job suffices)

Relay

• FOR UPDATE SKIP LOCKED used in the polling query
• Relay runs inside a transaction (begin before query, commit after all updates)
• Batch size is bounded (50-200 rows is typical)
• Relay respects context cancellation for graceful shutdown
• Failed publishes increment attempts rather than causing the batch to abort

Idempotency

• Published message includes the outbox id as a deduplication key
• Consumers are idempotent or the broker provides deduplication
• See Idempotency in Distributed Systems for deduplication patterns

Observability

• outbox_events_pending gauge is monitored and alerted on
• Dead-letter count is alerted on
• Relay batch duration is tracked
• Structured logs include event_id, event_type, and aggregate_id

Operations

• Manual retry path exists for dead-letter rows
• Relay restart behavior is tested (does it re-publish correctly?)
• Broker outage behavior is tested (does the outbox grow and drain correctly?)

Considerazioni finali

Il problema del dual-write è facile da liquidare come un caso limite finché non causa un incidente. Il pattern Transactional Outbox lo risolve con gli strumenti che hai già: una transazione PostgreSQL, un goroutine in background e una tabella extra. Il relay è semplice da costruire, semplice da gestire e semplice da analizzare.

Il costo è che i consumatori devono essere progettati per la consegna at-least-once. Questo è un compromesso ragionevole. La consegna esattamente-once tra un database e un broker senza transazioni distribuite non è realizzabile in pratica – e fingere il contrario porta a sistemi che perdono o elaborano due volte gli eventi in silenzio sotto condizioni di guasto.

Scrivi l’evento con i dati. Rilascialo in modo affidabile. Rendi i consumatori idempotenti. Questo è l’intero pattern.

Questo articolo fa parte del cluster App Architecture in Production.

Fonti

• PostgreSQL FOR UPDATE SKIP LOCKED
• PostgreSQL LISTEN/NOTIFY
• Debezium outbox event router
• lib/pq LISTEN support
• database/sql package