Das Transactional-Outbox-Muster in Go mit PostgreSQL

Zwei Schreibvorgänge, die gemeinsam erfolgreich sein sollten, werden sich im Laufe der Zeit separat fehlschlagen.

Ihr Order-Service speichert die Bestellung in der Datenbank und veröffentlicht anschließend ein order.created-Ereignis an einen Message Broker.

Diese beiden Operationen laufen nacheinander ab.

Dazwischen kann es zu Fehlern kommen: Der Broker ist nicht verfügbar, das Netzwerk timeoutet, der Prozess wird neu gestartet oder der Container wird entfernt. Der Datenbank-Schreibvorgang war erfolgreich. Die Veröffentlichung erfolgte nicht. Der nachgelagerte Service, der über die neue Bestellung informiert werden muss, erfährt davon nichts. Niemand merkt es, bis ein Kunde anruft.

Dies ist das Dual-Write-Problem (Problematik des doppelten Schreibens), und es ist eine der häufigsten Ursachen für stille Datenverluste in verteilten Systemen. Das Transactional-Outbox-Muster (Transaktionale-Outbox-Muster) ist die Standardlösung.

Das Dual-Write-Problem

Der Ausfallmodus ist leicht zu verstehen, sobald man ihn erkannt hat:

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

PUBLISH order.created ...  -- fehlschlägt, absturz oder wird nie erreicht

Die Datenbank und der Message Broker teilen sich keinen Transaktionsgrenzwert. Es gibt kein Rollback, das beide Vorgänge abdeckt. Jeder Service, der save -> publish in Sequenz ausführt, hat diese Lücke. Das Muster zeigt sich in vielen Formen:

• db.Save(order) gefolgt von events.Publish(OrderCreated{...})
• HTTP-Handler, der eine Transaktion committet und dann einen externen Webhook aufruft
• Worker, der einen Datensatz aus einer Warteschlange verarbeitet und die Ergebnisse in eine andere schreibt

Das Ergebnis ist in allen Fällen dasselbe: Eine Seite gelingt, während die andere fehlschlägt, und das System endet in einem Zustand, der für die Überwachung unsichtbar ist, da beide einzelnen Vorgänge zu irgendeinem Zeitpunkt erfolgreich zurückgegeben wurden.

Eine Retry-Schleife behebt dies nicht. Das erneute Versuchen der Veröffentlichung nach dem Datenbank-Commit funktioniert nur, wenn der Retry selbst zuverlässig ist – was genau die Dauerhaftigkeitsgarantie erfordert, die Sie nicht haben.

Was das Transactional-Outbox-Muster tut

Das Outbox-Muster beseitigt die Lücke, indem es die direkte Veröffentlichung vollständig entfernt. Anstatt den Broker innerhalb Ihrer Geschäftslogik aufzurufen, schreiben Sie einen Ereignisdatensatz in eine outbox-Tabelle in derselben Datenbanktransaktion wie die Geschäftsdaten. Ein separater Hintergrundprozess – der Relay – liest aus der Outbox-Tabelle und veröffentlicht sie an den Broker.

BEGIN;
  INSERT INTO orders ...         -- Geschäftsdaten
  INSERT INTO outbox_events ...  -- Ereignisdatensatz
COMMIT;

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

Beide Schreibvorgänge gelingen oder beide scheitern. Die Transaktionsgarantie, die Sie bereits von PostgreSQL haben, deckt nun auch den Ereignisdatensatz ab. Der Relay kann so oft neu versuchen zu veröffentlichen, wie nötig ist, da das Ereignis in einem dauerhaft gespeicherten Speicher liegt. Wenn der Relay abstürzt, startet er neu und versucht es erneut. Das Schlimmste, was passieren kann, ist, dass das Ereignis mehr als einmal veröffentlicht wird – was dadurch behandelt wird, dass die Konsumenten idempotent gemacht werden (siehe Idempotenz in verteilten Systemen).

PostgreSQL-Schema für die Outbox-Tabelle

Das Schema ist absichtlich einfach gehalten:

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
);

-- Partielle Index: indiziert nur unverarbeitete Zeilen, bleibt klein, da Zeilen als erledigt markiert werden
CREATE INDEX idx_outbox_unprocessed
    ON outbox_events (created_at)
    WHERE processed_at IS NULL;

Der partielle Index auf created_at WHERE processed_at IS NULL ist wichtig. Ohne ihn wächst der Index mit jedem jemals geschriebenen Ereignis und die Polling-Abfrage des Relays wird mit der Zeit langsamer. Mit ihm deckt der Index nur die ausstehenden Zeilen ab, was im steady state eine kleine, begrenzte Menge ist, unabhängig davon, wie viele Ereignisse veröffentlicht wurden.

Wichtige Feldauswahl:

• aggregate_type und aggregate_id beschreiben, zu welcher Entität das Ereignis gehört. Nützlich für Ordnungsgarantien und Routing.
• event_type ist der Ereignisname, den Ihre Konsumenten erwarten.
• payload JSONB speichert den Ereignisinhalt. Verwenden Sie JSONB statt TEXT, sodass Sie es bei Bedarf abfragen können.
• attempts verfolgt, wie oft der Relay versucht hat, diese Zeile zu veröffentlichen. Wird für Retry-Limits und Dead-Letter-Verarbeitung verwendet.
• processed_at ist NULL für ausstehende Zeilen und wird gesetzt, wenn der Relay erfolgreich veröffentlicht hat.

Geschäftsdaten und Outbox-Ereignis in einer Transaktion schreiben

Die Geschäftslogik schreibt beide Datensätze innerhalb eines einzelnen BeginTx / Commit-Aufrufs. Hier gibt es keinen Publish-Aufruf – nur Datenbank-Schreibvorgänge.

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()
}

Wenn tx.Commit() fehlschlägt, wird weder die Order-Zeile noch die Outbox-Zeile gespeichert. Wenn es erfolgreich ist, sind beide garantiert in der Datenbank. Der Relay kann das Ereignis zu jedem Zeitpunkt danach veröffentlichen – sofort, nach einer Sekunde oder nachdem der Relay nach einem Absturz neu gestartet wurde.

Dies ist die einzige Codeänderung, die in Ihrer Business-Schicht erforderlich ist. Der Rest des Musters lebt im Relay.

Go-Relay-Implementierung

Der Relay ist ein Hintergrund-Worker, der die Outbox-Tabelle in einem Timer abfragt. Er ruft einen Stapel unverarbeiteter Zeilen ab, veröffentlicht jede einzelne und markiert sie als erledigt. Halten Sie ihn in derselben Binärdatei wie Ihre Anwendung oder führen Sie ihn als separaten Prozess aus – beides funktioniert, aber dieselbe Binärdatei ist einfacher zu bedienen.

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)
            }
        }
    }
}

Der Relay respektiert die Kontext-Kündigung, was die Integration mit einem graceful Shutdown einfach macht. Für eine detaillierte Behandlung von Kontext-Lebensdauer und Kündigungsverhalten siehe [Go context.Context Done Right](https://www.glukhov.org/de/app-architecture/code-architecture/go-context-cancellation-timeouts/ “Meistern Sie Go Context für Kündigungen, Timeouts und anfragungsbezogene Werte. Behandelt HTTP-Handler, Datenbankaufrufe, Hintergrund-Worker, Goroutine-Leaks und graceful Shutdown.”}).

FOR UPDATE SKIP LOCKED: Das parallele Worker-Muster

Die Funktion processBatch verwendet FOR UPDATE SKIP LOCKED, um parallele Relay-Worker sicher zu handhaben:

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 macht zwei Dinge. Erstens sperrt FOR UPDATE die ausgewählten Zeilen für die Dauer der Transaktion und verhindert, dass andere Transaktionen sie auswählen. Zweitens bedeutet SKIP LOCKED, dass, wenn eine Zeile bereits von einer anderen Transaktion gesperrt ist, die Abfrage sie überspringt, anstatt zu warten. Das Ergebnis ist, dass mehrere Relay-Worker parallel laufen können und jeder eine nicht überlappende Teilmenge von Zeilen aufnimmt.

Ohne SKIP LOCKED würde ein zweiter Worker blockieren, bis die erste Transaktion committet, bevor er dieselben Zeilen sieht – zu diesem Zeitpunkt wären sie bereits als erledigt markiert. Mit SKIP LOCKED nimmt der zweite Worker sofort andere Zeilen auf, anstatt zu warten, was Ihnen eine sichere horizontale Skalierung bietet.

Beachten Sie die Trennung von Scan und Publish im obigen Code: Alle Zeilen werden in einen Schnitt (Slice) gescannt, bevor die Publish-Schleife beginnt. Dies vermeidet das Halten eines offenen *sql.Rows-Cursors über Netzwerkaufrufe an den Broker hinweg, was die Transaktion länger offen halten würde als nötig.

Idempotenz und Deduplizierung

Der Relay veröffentlicht mindestens einmal. Wenn er ein Ereignis veröffentlicht und dann abstürzt, bevor er das processed_at-Update committet, wird er dasselbe Ereignis beim Neustart erneut veröffentlichen. Dies ist unvermeidbar – genau einmalige Lieferung (exactly-once delivery) über eine Datenbank und einen Message Broker ohne verteilten Transaktionskoordinator erfordert diesen Trade-off.

Konsumenten müssen idempotent sein. Der einfachste Ansatz ist, verarbeitete Ereignis-IDs in einer processed_events-Tabelle zu verfolgen:

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 {
    // Deduplizierung anhand der Event-ID als natürlicher Schlüssel
    _, 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)
    }

    // Prüfen, ob der Einfügevorgang tatsächlich stattgefunden hat (1 Zeile) oder ein No-Op war (0 Zeilen)
    // Ein einfacherer Ansatz: RETURNING verwenden oder betroffene Zeilen prüfen
    // Wenn 0 Zeilen betroffen, ist dies ein Duplikat -- überspringen
    ...
}

In der Praxis verlassen sich viele Teams auf die eigene Deduplizierungsheader des Brokers (wie das key-Feld von Kafka für log-kompakte Topics oder den message-id-Header von RabbitMQ) und betrachten die Datenbank-Deduplizierung als Fallback. Beide sind gültige Schichten, die angewendet werden können.

Schließen Sie die Outbox-Ereignis-id (eine UUID) in die veröffentlichte Nachricht als Deduplizierungsschlüssel ein. Konsumenten können sie dann unabhängig davon verwenden, welchen Deduplizierungsmechanismus sie bevorzugen.

Retry-Policy und Poison Messages (Giftige Nachrichten)

Die attempts-Spalte treibt die Retry-Policy an. Der Relay überspringt Zeilen, bei denen attempts >= maxAttempts, und behandelt diese Zeilen als Dead Letters. Ein separater Prozess oder Operator-Alert übernimmt deren Handhabung.

Eine einfache Dead-Letter-Ansicht:

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

Eine gute Produktions-Retry-Policy:

• Setzen Sie maxAttempts auf 5-10, abhängig davon, wie teuer Retries sind.
• Erwägen Sie exponentielles Backoff: Fügen Sie eine retry_after-Spalte hinzu und überspringen Sie Zeilen, bei denen retry_after > NOW().
• Alert bei COUNT(*) FROM outbox_dead_letters, die einen Schwellenwert überschreitet.
• Bieten Sie einen manuellen Retry-Pfad an: Ein Admin-Endpunkt oder Skript, das attempts = 0 und retry_after = NULL für bestimmte Zeilen zurücksetzt.

Poison Messages – Zeilen, die aufgrund eines Fehlers im Konsumenten oder eines Schema-Mismatches konsistent fehlschlagen – sollten gesunde Nachrichten nicht blockieren. Da der Relay einen Stapel pro Tick verarbeitet und Fehler mit einem Attempt-Inkrement markiert, anstatt sie aus der Warteschlange zu entfernen, laufen gesunde Zeilen normal weiter, während vergiftete sich ansammeln, bis sie den Dead-Letter-Schwellenwert erreichen. Diese outbox_dead_letters-Ansicht ist eine datenbankseitige Version desselben Musters, das broker-native Dead-Letter-Warteschlangen implementieren – Quarantäne nach einem Schwellenwert, Alert bei Volumen, und eine bewusste Entscheidung vor dem Replay erforderlich.

Ereignis-Reihenfolge und Partitionierung

Die Polling-Abfrage ordnet nach created_at, was eine First-In-First-Out-Reihenfolge innerhalb eines Stapels ergibt. Für die meisten Anwendungsfälle ist das ausreichend. Wenn eine strenge pro-Entitäts-Reihenfolge wichtig ist – zum Beispiel, um sicherzustellen, dass order.updated niemals vor order.created für dieselbe Bestellung veröffentlicht wird – benötigen Sie pro-Aggregat-Reihenfolge.

Fügen Sie aggregate_id zur ORDER BY-Klausel hinzu und verwenden Sie es als Nachrichten-Schlüssel beim Veröffentlichen in ein partitioniertes Topic wie [Apache Kafka](https://www.glukhov.org/de/data-infrastructure/stream-processing/apache-kafka/ “Lernen Sie Apache Kafka 4.2 schnell mit Tarball oder Docker, starten Sie einen lokalen KRaft-Broker, meistern Sie die wichtigsten CLI-Tools und führen Sie praktische Producer-, Consumer- und Connect-Beispiele aus.”}). Kafka leitet alle Nachrichten mit demselben Schlüssel an dieselbe Partition weiter, und Partitionen werden in Reihenfolge konsumiert. Dies gibt Ihnen pro-Aggregat-Reihenfolgegarantien ohne globale Reihenfolge, was eine einzelne Relay-Instanz erfordern würde.

ORDER BY aggregate_id, created_at

Für Broker, die keine partitionierte Reihenfolge unterstützen (wie grundlegende AMQP-Warteschlangen), sind einzelne Relay-Instanzen oder anwendungsseitige Reihenfolgeprüfungen im Konsument die praktischen Alternativen.

Polling-Latenz mit LISTEN/NOTIFY reduzieren

Ein Polling-Intervall von einer Sekunde bedeutet eine durchschnittliche Ereignis-Latenz von 500 Millisekunden. Für die meisten Workloads ist das in Ordnung. Für Fälle, in denen Sie eine nahezu Null-Latenz benötigen, ermöglicht PostgreSQLs LISTEN/NOTIFY-Mechanismus dem Relay, sofort aufzuwachen, wenn eine neue Outbox-Zeile eingefügt wird.

Fügen Sie einen Trigger zur Outbox-Tabelle hinzu:

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();

Im Relay hören Sie auf dem Kanal und wachen bei Benachrichtigungen auf, fallen aber weiterhin auf periodisches Polling zurück:

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)
            }
        }
    }
}

Der Fallback-Ticker behandelt alle während eines Relay-Neustarts oder Netzwerk-Problems verpassten Benachrichtigungen. Halten Sie das Fallback-Intervall bei einigen Sekunden statt Millisekunden – seine Aufgabe ist die Wiederherstellung, nicht niedrige Latenz.

Beobachtbarkeit: Metriken, Logs und Alerts

Die Outbox ist Infrastruktur. Behandeln Sie sie wie Infrastruktur und instrumentieren Sie sie entsprechend.

Wichtige Metriken:

var (
    outboxPublished = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "outbox_events_published_total",
        Help: "Gesamtzahl der erfolgreich veröffentlichten Outbox-Ereignisse.",
    })
    outboxFailed = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "outbox_events_failed_total",
        Help: "Gesamtzahl der Outbox-Veröffentlichungsfehler nach Ereignistyp.",
    }, []string{"event_type"})
    outboxPending = prometheus.NewGauge(prometheus.GaugeOpts{
        Name: "outbox_events_pending",
        Help: "Aktuelle Anzahl unverarbeiteter Outbox-Ereignisse.",
    })
    outboxBatchDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
        Name:    "outbox_batch_duration_seconds",
        Help:    "Dauer jedes Outbox-Verarbeitungsbatches.",
        Buckets: prometheus.DefBuckets,
    })
)

Gauge-Aktualisierung: Führen Sie eine periodische Abfrage aus, um outbox_events_pending aktuell zu halten:

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

Zu berücksichtigende Alert-Schwellenwerte:

• outbox_events_pending > 1000 für mehr als zwei Minuten: Relay hinkt hinterher oder steht.
• outbox_events_pending wächst monoton: Broker ist down oder Relay ist abgestürzt.
• Dead-Letter-Zahl nicht null: Schema- oder Konsumentenfehler erfordert Untersuchung.
• outbox_batch_duration_seconds p95 > 5s: Datenbank ist langsam oder Batch-Größe ist zu groß.

Strukturierte Log-Felder: Schließen Sie event_id, event_type, aggregate_id und attempt in jede Log-Zeile vom Relay ein. Diese Felder ermöglichen es Ihnen, einen fehlgeschlagenen Publish mit der spezifischen Outbox-Zeile und der nachgelagerten Konsumententrace zu korrelieren.

Outbox vs. direkte Warteschlange vs. Saga

Das Outbox-Muster ist nicht das richtige Werkzeug für jedes Koordinationsproblem. Hier ist der Vergleich:

Das Outbox-Muster garantiert, dass ein einzelner Service zuverlässig Ereignisse emittiert, die seine eigenen Zustandsänderungen widerspiegeln. Es koordiniert keine Zustandsänderungen über mehrere Dienste hinweg – dafür ist das [Saga-Muster](https://www.glukhov.org/de/app-architecture/integration-patterns/saga-transactions-in-microservices/ “Vollständiger Leitfaden zur Saga-Muster-Implementierung in Go für Microservices. Lernen Sie Orchestrierung vs. Choreographie, Kompensationsstrategien, Idempotenz, Event Sourcing und Best Practices mit praktischen Beispielen.”}) gedacht. Die Wahl des Brokers – ob RabbitMQ, SQS, oder Kafka – ist unabhängig vom Outbox-Muster selbst; der Relay veröffentlicht zu dem Broker, den Ihr System verwendet.

Wenn Sie eine Saga bauen, ist das Outbox-Muster dennoch nützlich: Jeder Teilnehmer an der Saga schreibt seine lokale Zustandsänderung und sein Saga-Ereignis in einer Transaktion unter Verwendung der Outbox, dann liest der Saga-Orchestrator oder die Choreographie diese Ereignisse zuverlässig.

WAL-basiertes CDC als alternative Relay

Anstatt zu pollen, können Sie das Write-Ahead Log (WAL) von PostgreSQL tailen und Outbox-Einfügungen direkt aus dem Replikationsstream lesen. Tools wie Debezium tun dies. Die Vorteile sind niedrigere Latenz und kein Lock-Druck auf der Outbox-Tabelle. Die Nachteile sind operative Komplexität, ein dedizierter PostgreSQL-Replikations-Slot und ein externer Dienst, den es zu betreiben und zu überwachen gilt.

Für die meisten Teams ist der oben beschriebene Polling-Relay der richtige Startpunkt. WAL-Tailing macht Sinn, wenn Sie hohe Outbox-Einfügungsraten (Zehntausende pro Sekunde) haben, eine Ereignis-Latenz von unter 100ms benötigen oder bereits Debezium für andere Change-Capture-Bedürfnisse einsetzen.

sqlc-Integration

Wenn Sie sqlc für typsicheren Go-Datenbankcode verwenden, passen die Outbox-Abfragen natürlich:

-- 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 generiert typsichere Funktionen für jede Abfrage, was Fehler durch String-Interpolation vermeidet und die Outbox-Abfrage-Logik ko-lokal mit dem Rest Ihrer Datenbankzugriffsschicht hält.

Produktions-Checkliste

Verwenden Sie diese vor dem Release einer Outbox-Implementierung:

Datenbank

• Outbox-Tabelle hat den partiellen Index auf created_at WHERE processed_at IS NULL
• attempts-Spalte vorhanden mit einem Standardwert von 0
• Dead-Letter-Ansicht oder -Abfrage definiert
• Alte verarbeitete Zeilen werden regelmäßig archiviert oder gelöscht (ein nächtlicher Aufräumauftrag reicht aus)

Relay

• FOR UPDATE SKIP LOCKED in der Polling-Abfrage verwendet
• Relay läuft innerhalb einer Transaktion (beginnen Sie vor der Abfrage, committen Sie nach allen Updates)
• Batch-Größe ist begrenzt (50-200 Zeilen sind typisch)
• Relay respektiert Kontext-Kündigung für graceful Shutdown
• Fehlgeschlagene Publishes erhöhen attempts, anstatt den Batch abzubrechen

Idempotenz

• Veröffentlichte Nachricht enthält die Outbox-id als Deduplizierungsschlüssel
• Konsumenten sind idempotent oder der Broker bietet Deduplizierung
• Siehe Idempotenz in verteilten Systemen für Deduplizierungsmuster

Beobachtbarkeit

• outbox_events_pending-Gauge wird überwacht und darauf alertiert
• Dead-Letter-Zahl wird alarmiert
• Relay-Batch-Dauer wird verfolgt
• Strukturierte Logs enthalten event_id, event_type und aggregate_id

Betrieb

• Manueller Retry-Pfad für Dead-Letter-Zeilen vorhanden
• Relay-Neustartverhalten getestet (veröffentlicht es korrekt neu?)
• Broker-Ausfallverhalten getestet (wächst und leert sich die Outbox korrekt?)

Letzte Gedanken

Das Dual-Write-Problem ist leicht als Randfall abzutun, bis es einen Vorfall verursacht. Das Transactional-Outbox-Muster löst es mit Tools, die Sie bereits haben: eine PostgreSQL-Transaktion, ein Hintergrund-Goroutine und eine zusätzliche Tabelle. Der Relay ist einfach zu bauen, einfach zu betreiben und einfach zu durchdenken.

Die Kosten bestehen darin, dass Konsumenten für mindestens einmalige Lieferung (at-least-once delivery) entworfen werden müssen. Das ist ein angemessener Trade-off. Genau einmalige Lieferung über eine Datenbank und einen Broker ohne verteilte Transaktionen ist praktisch nicht erreichbar – und so zu tun, als wäre sie möglich, führt zu Systemen, die unter Ausfallbedingungen stillschweigend Ereignisse droppt oder doppelt verarbeitet.

Schreiben Sie das Ereignis mit den Daten. Leiten Sie es zuverlässig weiter. Machen Sie Konsumenten idempotent. Das ist das gesamte Muster.

Dieser Artikel ist Teil des App Architecture in Production-Clusters.

Quellen

• PostgreSQL FOR UPDATE SKIP LOCKED
• PostgreSQL LISTEN/NOTIFY
• Debezium Outbox Event Router
• lib/pq LISTEN-Unterstützung
• database/sql-Paket