Zum Inhalt
wannacode

Transactional Outbox

Kurz gesagt: Die Transactional Outbox schreibt das Ereignis in derselben Transaktion wie die Änderung in dieselbe Datenbank, und ein eigenes Relay sendet es an den Broker, damit kein Ereignis verloren geht.

Problem

Bezahlt, und kein Paket geht raus

Seit der vorigen Lektion veröffentlicht der Shop ein Ereignis, sobald die Bestellung gespeichert ist. Das Speichern geht an die Datenbank, und das Ereignis geht an den Broker. Hier sind zwei Arten, das zu schreiben. Darin ist db die Datenbank, queue die Warteschlange des Brokers, und crash.at eine Stelle, an der der Prozess nach einem aufgezeichneten Szenario sterben kann:

# SO NICHT: Das Speichern und das Ereignis liegen in zwei getrennten Systemen
def save_then_publish(order_id, total):
    with db.transaction() as tx:
        tx.orders[order_id] = total
    crash.at("nach dem Speichern")
    queue.append(f"evt-{order_id}: OrderPlaced {order_id}")
def publish_then_save(order_id, total):
    queue.append(f"evt-{order_id}: OrderPlaced {order_id}")
    with db.transaction() as tx:
        tx.orders[order_id] = total
        crash.at("vor dem Commit")

In jeder Funktion stirbt der Prozess einmal, an ihrer eigenen Stelle:

crash = CrashScript("nach dem Speichern")
try:
    save_then_publish(1001, 120)
except Crash as e:
    print(f"✗ der Prozess stirbt {e}")
crash = CrashScript("vor dem Commit")
try:
    publish_then_save(1002, 80)
except Crash as e:
    print(f"✗ der Prozess stirbt {e}")
print(f"Bestellungen: [{', '.join(map(str, db.orders))}]")
print(f"Warteschlange: [{', '.join(queue)}]")
Ausgabe
✗ der Prozess stirbt nach dem Speichern
✗ der Prozess stirbt vor dem Commit
Bestellungen: [1001]
Warteschlange: [evt-1002: OrderPlaced 1002]

Bestellung 1001 ist in der Datenbank, aber für sie gibt es kein Ereignis. Der Kunde hat bezahlt, und das Lager weiß nicht, dass es ein Paket schicken muss. Das Ereignis für 1002 ist in der Warteschlange, aber Bestellung 1002 gibt es nicht: Die Transaktion wurde zurückgerollt. Das Lager packt eine Bestellung, die es nicht gibt.

Ein Absturz ist nicht selten. Der Prozess startet bei jedem Deploy neu, der Maschine geht der Strom aus, das Netz zum Broker fällt für eine Sekunde aus. Bei Tausenden Bestellungen am Tag landet irgendeine genau zwischen den beiden Zeilen.

Die Idee

Zwei Systeme, keine gemeinsame Transaktion

Eine Transaktion verspricht „alles oder nichts“, aber nur für die Schreibvorgänge in ihrer eigenen Datenbank. Der Broker ist ein anderes System. Der Datenbank-Commit weiß nichts von ihm, und eine Nachricht, die an den Broker gegangen ist, lässt sich nicht „zurückholen“. Egal in welcher Reihenfolge die beiden Zeilen stehen: Ein Absturz dazwischen lässt ein System ohne das andere zurück.

Es gibt ein Protokoll für eine gemeinsame Transaktion über zwei Systeme (Two-Phase-Commit), aber die meisten Broker unterstützen es nicht, und wo es existiert, ist es langsam und anfällig.

Die Lösung ist, im Moment des Speicherns gar nicht zwei Systeme zu haben. Das Ereignis wird als Zeile in einer Tabelle derselben Datenbank geschrieben, in derselben Transaktion wie die Bestellung. Die Tabelle heißt Outbox. Der Commit schreibt beide Zeilen oder keine. Dann liest ein eigenes Relay die Outbox, sendet die noch nicht gesendeten Zeilen an den Broker und hakt sie ab. Das ist das Muster Transactional Outbox.

Sieh es dir an

Die Datenbank und der Absturz im Speicher

Die Datenbank ist hier ein Objekt mit zwei Tabellen. Die Transaktion sammelt die Schreibvorgänge in Tx und legt sie auf einmal in die Tabellen, nur wenn die Arbeit bis zum Ende kommt. Wirft sie auf halbem Weg, wird nichts geschrieben: Das ist ein Rollback.

@dataclass
class OutboxRow:
    id: str
    event: str
    sent: bool = False




# die Schreibvorgänge einer Transaktion, vor dem Commit
class Tx:
    def __init__(self):
        self.orders = {}
        self.outbox = []




class Db:
    def __init__(self):
        self.orders = {}
        self.outbox = []


    # alles oder nichts: Die Schreibvorgänge landen nur in den Tabellen, wenn die Arbeit bis zum Ende kommt
    @contextmanager
    def transaction(self):
        tx = Tx()
        yield tx
        self.orders.update(tx.orders)  # commit
        self.outbox.extend(tx.outbox)
In Python ist die Transaktion ein with-Block: @contextmanager gibt dem Block tx, und wirft der Block, kommt der Fehler aus yield, und die Zeilen danach (der Commit) laufen nicht. So sehen Transaktionen auch in sqlite3 und SQLAlchemy aus.

Der Absturz ist eine Exception an einer genauen Stelle. Die Datenbank und die Warteschlange sind getrennte Systeme: Sie überleben den Absturz, und das Programm läuft weiter wie ein neu gestarteter Prozess.

class Crash(Exception):
    pass




# ein aufgezeichnetes Szenario: wo der Prozess stirbt (ohne Stelle: nirgends)
class CrashScript:
    def __init__(self, point=""):
        self.point = point


    def at(self, here):
        if here == self.point:
            raise Crash(here)
db = Db()
queue = []  # die Warteschlange des Brokers
crash = CrashScript()

Wie dasselbe mit einer echten Datenbank aussieht, hast du in der Lektion über Datenbanken gesehen.

Sieh es dir an

Die Bestellung und die Outbox-Zeile zusammen

Das Speichern der Bestellung legt auch eine Zeile in die Outbox, in derselben Transaktion. Es sendet nichts an den Broker:

def place_order(order_id, total):
    with db.transaction() as tx:
        tx.orders[order_id] = total
        # in derselben Transaktion
        tx.outbox.append(OutboxRow(f"evt-{order_id}", f"OrderPlaced {order_id}"))
        crash.at("vor dem Commit")

Das Relay geht die Outbox durch, sendet jede noch nicht gesendete Zeile und hakt sie ab:

# das Relay sendet die ungesendeten Zeilen der Outbox
def relay():
    for row in db.outbox:
        if row.sent:
            continue
        queue.append(f"{row.id}: {row.event}")
        crash.at("im Relay")
        row.sent = True
def show():
    print(f"Bestellungen: [{', '.join(map(str, db.orders))}]")
    print(f"outbox: [{', '.join(r.id + (' ✓' if r.sent else '') for r in db.outbox)}]")
    print(f"Warteschlange: [{', '.join(queue)}]")

Bestellung 1003 geht durch. Bestellung 1004 stirbt vor dem Commit. Dann startet das Relay, wie nach einem Neustart:

place_order(1003, 120)
show()
crash = CrashScript("vor dem Commit")
try:
    place_order(1004, 80)
except Crash as e:
    print(f"✗ der Prozess stirbt {e}")
crash = CrashScript()
relay()  # nach dem Neustart
show()
Ausgabe
Bestellungen: [1003]
outbox: [evt-1003]
Warteschlange: []
✗ der Prozess stirbt vor dem Commit
Bestellungen: [1003]
outbox: [evt-1003 ✓]
Warteschlange: [evt-1003: OrderPlaced 1003]

def place_order(order_id, total):
    with db.transaction() as tx:
        tx.orders[order_id] = total
        # in derselben Transaktion
        tx.outbox.append(OutboxRow(f"evt-{order_id}", f"OrderPlaced {order_id}"))
        crash.at("vor dem Commit")
Schritt für Schritt, in Worten
  1. Szenario 1: erst speichern, dann das Ereignis. Bestellung 1001 kommt mit einem Commit in die Datenbank.
  2. Der Prozess stirbt genau zwischen den beiden Zeilen. Die Bestellung ist in der Datenbank, aber das Ereignis geht nie raus.
  3. Diese Zeile läuft nie. Das Lager erfährt nie von 1001: Der Kunde hat bezahlt, und kein Paket geht raus.
  4. Szenario 2: andersherum. Das Ereignis evt-1002 geht zuerst raus.
  5. Dann wird die Bestellung in einer Transaktion geschrieben.
  6. Der Prozess stirbt vor dem Commit, und die Datenbank verwirft den Schreibvorgang. Bestellung 1002 gibt es nicht.
  7. Aber das Ereignis ist schon in der Warteschlange: Das Lager packt eine Bestellung, die es nicht gibt.
  8. Szenario 3, die Outbox: Bestellung 1003 und ihr Ereignis, als Zeile in der Outbox-Tabelle, in einer Transaktion.
  9. Commit: Beide Zeilen gehen auf einmal rein. Entweder beide oder keine.
  10. Der Prozess stirbt, bevor er etwas sendet. Nichts geht verloren: Die Zeile wartet in der Outbox.
  11. Nach dem Neustart liest das Relay die Outbox und findet die ungesendete Zeile.
  12. Es sendet evt-1003 an die Warteschlange, und das Lager bekommt es.
  13. Erst dann hakt es die Zeile ab. Ein Absturz dazwischen sendet die Zeile noch einmal, deshalb muss der Consumer idempotent sein.
  14. Das Speichern und das Ereignis laufen nicht mehr auseinander: Beides kommt an, früher oder später.

Nach dem Commit sind die Bestellung und die Outbox-Zeile in der Datenbank, und die Warteschlange ist noch leer: Der Prozess darf hier sterben, und nichts geht verloren. Bestellung 1004 hinterlässt nichts: keine Bestellung und keine Zeile, weil beide in einer Transaktion liegen. Das Relay findet die noch nicht gesendete Zeile und sendet sie.

In einem echten System ist das Relay eine Schleife, die jede Sekunde die Datenbank fragt: „Gib mir die noch nicht gesendeten Zeilen.“ Deshalb geht das Ereignis etwas nach dem Commit raus, nicht im selben Augenblick.

Probier es aus

Sag die Ausgabe voraus

Das Relay sendet die Zeile und stirbt, bevor es sie abhakt. Nach dem Neustart läuft es wieder. Was steht in der Warteschlange?

Die letzte Zeile der Ausgabe ist:

place_order(1003, 120)
crash = CrashScript("im Relay")
try:
    relay()
except Crash as e:
    print(f"✗ der Prozess stirbt {e}")
crash = CrashScript()
relay()  # nach dem Neustart
show()
Antwort

Probier es aus

Finde den Fehler

Ein Kollege findet Duplikate lästig und vertauscht zwei Zeilen im Relay. Der Absturz ist an derselben Stelle, zwischen den beiden Zeilen:

Ausgabe
✗ der Prozess stirbt im Relay
Bestellungen: [1003]
outbox: [evt-1003 ✓]
Warteschlange: []

Die Outbox-Zeile ist abgehakt, und die Warteschlange ist leer. Welche Zeile hat den Fehler?

Wähle die Zeile mit dem Fehler.

Probier es aus

Bring die Zeilen in Ordnung

Bring den Rumpf der Relay-Schleife in eine Reihenfolge, in der nichts verloren geht.

  1. row.sent = True
  2. queue.append(f"{row.id}: {row.event}")
  3. if row.sent:
  4. continue

Die Idee

Wann sich eine Outbox nicht lohnt

Eine Outbox bringt eine Tabelle, ein Relay und eine Verzögerung mit. Benutze sie nicht, wenn:

  • es keinen eigenen Broker gibt. Liegen die Reaktionen im selben Prozess und schreiben in dieselbe Datenbank (Observer in derselben Transaktion), genügt eine Transaktion;
  • eine verlorene Nachricht nicht schadet. Statistiken aktualisieren oder einen Cache aufwärmen darf einmal im Monat verloren gehen, ohne dass jemand zu Schaden kommt;
  • das Ereignis in derselben Millisekunde rausmuss. Das Relay liest in Abständen. Eine Sekunde Verzögerung ist meist in Ordnung, aber nicht immer.

Und wenn du sie benutzt, denke an:

  • Duplikate. Das Relay sendet mindestens einmal. Consumer müssen idempotent sein;
  • Reihenfolge. Ein Relay sendet in der Reihenfolge der Outbox. Mehrere Relays gleichzeitig können die Reihenfolge vertauschen oder eine Zeile zweimal senden, deshalb werden Zeilen beim Lesen gesperrt;
  • Wachstum. Gesendete Zeilen werden nach ein paar Tagen gelöscht, sonst wächst die Tabelle ewig.

Auf der anderen Seite gibt es ein Spiegelmuster, die Inbox: Der Consumer schreibt die ID des empfangenen Ereignisses in seine eigene Datenbank, in derselben Transaktion wie seine Arbeit. Das ist das seen des Lagers aus der vorigen Lektion, nur als Tabelle.

Genauer, für Neugierige

Das Relay in einem echten System. Das einfachste ist eine Schleife mit der Abfrage SELECT … FROM outbox WHERE sent = false ORDER BY id LIMIT 100. In PostgreSQL lässt FOR UPDATE SKIP LOCKED mehrere Relays laufen, ohne dieselbe Zeile zu nehmen. Ein anderer Weg ist Change Data Capture: Debezium liest das Log der Datenbank (das WAL in PostgreSQL, das Binlog in MySQL) und sendet neue Outbox-Zeilen an Kafka, ohne Abfragen in Abständen. Wie Debezium oder Kafka Connect betrieben und überwacht werden, ist ein Thema des DevOps-Kurses ZeroToOps.

Fertige Lösungen. Spring Modulith hat eine Event Publication Registry, das ist eine Outbox. In .NET haben MassTransit und NServiceBus eine eingebaute Outbox auf Entity Framework. In Python und Node.js schreibt man sie meist von Hand, mit einer Tabelle und einem kleinen Relay, wie in dieser Lektion.

Two-Phase-Commit. XA-Transaktionen (JTA in Java, System.Transactions in .NET) versprechen eine gemeinsame Transaktion über eine Datenbank und einen Broker. In der Praxis unterstützen Kafka, RabbitMQ und die Cloud-Warteschlangen sie nicht. Deshalb ist die Outbox die übliche Antwort.

Warum nicht einfach wiederholen? „Wenn das Senden fehlschlägt, versuch es noch einmal“ hilft bei einem Fehler des Brokers, aber nicht bei einem Absturz des Prozesses: Nach dem Neustart erinnert sich niemand, dass etwas rausmusste. Die Outbox ist das Gedächtnis, das den Absturz überlebt.

Übung

Ein Relay in Paketen

Löse es in deinem eigenen Editor.

In einem echten System liest das Relay höchstens ein paar Zeilen auf einmal (LIMIT in der Abfrage). Schreibe relay(limit), das höchstens limit noch nicht gesendete Zeilen sendet und zurückgibt, wie viele es gesendet hat. Speichere drei Bestellungen und lass das Relay mit limit 2 laufen, bis es 0 zurückgibt:

Ausgabe
Runde 1: gesendet 2
Runde 2: gesendet 1
Runde 3: gesendet 0
Warteschlange: 3

Weitere Aufgaben

  • Eine stornierte Bestellung. Ergänze cancelOrder(id), das in einer Transaktion die Bestellung entfernt und OrderCancelled in die Outbox legt. Prüfe, dass ein Absturz vor dem Commit die Bestellung nicht entfernt und nichts sendet.
  • Aufräumen. Ergänze die Zeile um den Moment des Sendens (einen Tick) und schreibe eine Funktion, die gesendete Zeilen löscht, die älter als 10 Ticks sind.
  • Inbox. Schreibe ein Lager, das die IDs der empfangenen Ereignisse behält und das Duplikat aus der Karte „Sag die Ausgabe voraus“ verwirft.

Test

Teste dich

1. Warum schützt die Datenbanktransaktion nicht auch das Senden an den Broker?
2. Was schreibt der Shop in einer Transaktion, wenn er eine Outbox benutzt?
3. Das Relay sendet eine Zeile und stirbt, bevor es sie abhakt. Was passiert nach dem Neustart?
4. Wann ist eine Outbox unnötig?

Erstellt von Ivo Gergov im Rahmen der Kampagne von Evo Tech für zugängliche Programmierausbildung ·