Към съдържанието
wannacode

Transactional outbox

Накратко: Transactional outbox записва събитието в същата база и в същата транзакция като промяната, а отделен изпращач го праща към брокера, затова не се губи нито едно събитие.

Проблем

Платено, а пакет не тръгва

От предишния урок магазинът пуска събитие, щом поръчката е записана. Записът е в базата, а събитието отива при брокера. Ето два начина да се напише това. В тях db е базата, queue е опашката на брокера, а crash.at е място, където процесът може да падне по записан сценарий:

# НЕ ТАКА: записът и събитието са в две отделни системи
def save_then_publish(order_id, total):
    with db.transaction() as tx:
        tx.orders[order_id] = total
    crash.at("след записа")
    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("преди commit")

Във всяка функция процесът пада веднъж, на своето място:

crash = CrashScript("след записа")
try:
    save_then_publish(1001, 120)
except Crash as e:
    print(f"✗ процесът пада {e}")
crash = CrashScript("преди commit")
try:
    publish_then_save(1002, 80)
except Crash as e:
    print(f"✗ процесът пада {e}")
print(f"поръчки: [{', '.join(map(str, db.orders))}]")
print(f"опашка: [{', '.join(queue)}]")
Изход
✗ процесът пада след записа
✗ процесът пада преди commit
поръчки: [1001]
опашка: [evt-1002: OrderPlaced 1002]

Поръчка 1001 е в базата, но събитие за нея няма. Клиентът е платил, а складът не знае, че трябва да прати пакет. Събитието за 1002 е в опашката, но поръчка 1002 я няма: транзакцията се е върнала. Складът опакова поръчка, която не съществува.

Сривът не е рядкост. Процесът се рестартира при всеки deploy, машината губи ток, мрежата до брокера прекъсва за секунда. При хиляди поръчки на ден все някоя попада точно между двата реда.

Идеята

Две системи, без обща транзакция

Транзакцията обещава „всичко или нищо“, но само за записите в своята база. Брокерът е друга система. Commit на базата не знае за него, а съобщение, пратено към брокера, не може да се „върне“. Затова каквото и да е редът на двата реда, срив между тях оставя едната система без другата.

Има протокол за обща транзакция през две системи (two-phase commit), но повечето брокери не го поддържат, а където го има, е бавен и крехък.

Решението е да няма две системи в момента на записа. Събитието се записва като ред в таблица на същата база, в същата транзакция като поръчката. Таблицата се казва outbox („изходяща поща“). Commit записва и двата реда, или нито един. После отделен изпращач (relay) чете outbox, праща непратените редове към брокера и ги отмята. Това е шаблонът transactional outbox.

Виж го

Базата и сривът в паметта

Базата тук е обект с две таблици. Транзакцията трупа записите в Tx и ги слага в таблиците наведнъж, само ако работата стигне до края. Ако по средата хвърли, нищо не е записано: това е rollback.

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




# записите на една транзакция, преди commit
class Tx:
    def __init__(self):
        self.orders = {}
        self.outbox = []




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


    # всичко или нищо: записите влизат в таблиците, само ако работата стигне до края
    @contextmanager
    def transaction(self):
        tx = Tx()
        yield tx
        self.orders.update(tx.orders)  # commit
        self.outbox.extend(tx.outbox)
В Python транзакцията е with блок: @contextmanager дава tx на блока, а ако блокът хвърли, грешката излиза от yield и редовете след него (commit) не се изпълняват. Така изглеждат и транзакциите в sqlite3 и SQLAlchemy.

Сривът е изключение на точно определено място. Базата и опашката са отделни системи: оцеляват след срива, а програмата продължава като рестартиран процес.

class Crash(Exception):
    pass




# записан сценарий: къде пада процесът (без място: никъде)
class CrashScript:
    def __init__(self, point=""):
        self.point = point


    def at(self, here):
        if here == self.point:
            raise Crash(here)
db = Db()
queue = []  # опашката на брокера
crash = CrashScript()

Как изглежда същото с истинска база, видя в урока за базите данни.

Виж го

Поръчката и редът в outbox заедно

Записът на поръчката слага и ред в outbox, в същата транзакция. Към брокера не праща нищо:

def place_order(order_id, total):
    with db.transaction() as tx:
        tx.orders[order_id] = total
        # в същата транзакция
        tx.outbox.append(OutboxRow(f"evt-{order_id}", f"OrderPlaced {order_id}"))
        crash.at("преди commit")

Изпращачът минава по outbox, праща всеки непратен ред и го отмята:

# изпращачът праща непратените редове от outbox
def relay():
    for row in db.outbox:
        if row.sent:
            continue
        queue.append(f"{row.id}: {row.event}")
        crash.at("в изпращача")
        row.sent = True
def show():
    print(f"поръчки: [{', '.join(map(str, db.orders))}]")
    print(f"outbox: [{', '.join(r.id + (' ✓' if r.sent else '') for r in db.outbox)}]")
    print(f"опашка: [{', '.join(queue)}]")

Поръчка 1003 минава. Поръчка 1004 пада преди commit. После изпращачът тръгва, както след рестарт:

place_order(1003, 120)
show()
crash = CrashScript("преди commit")
try:
    place_order(1004, 80)
except Crash as e:
    print(f"✗ процесът пада {e}")
crash = CrashScript()
relay()  # след рестарта
show()
Изход
поръчки: [1003]
outbox: [evt-1003]
опашка: []
✗ процесът пада преди commit
поръчки: [1003]
outbox: [evt-1003 ✓]
опашка: [evt-1003: OrderPlaced 1003]

def place_order(order_id, total):
    with db.transaction() as tx:
        tx.orders[order_id] = total
        # в същата транзакция
        tx.outbox.append(OutboxRow(f"evt-{order_id}", f"OrderPlaced {order_id}"))
        crash.at("преди commit")
Стъпка по стъпка, с думи
  1. Сценарий 1: първо запис, после събитие. Поръчка 1001 влиза в базата с commit.
  2. Процесът пада точно между двата реда. Поръчката е в базата, а събитието не тръгва никога.
  3. Този ред не се изпълнява. Складът не научава за 1001: клиентът е платил, а пакет няма да тръгне.
  4. Сценарий 2: обратното. Събитието evt-1002 тръгва първо.
  5. После поръчката се записва в транзакция.
  6. Процесът пада преди commit и базата връща записа. Поръчка 1002 я няма.
  7. А събитието вече е в опашката: складът опакова поръчка, която не съществува.
  8. Сценарий 3, outbox: поръчка 1003 и събитието ѝ като ред в таблицата outbox, в една транзакция.
  9. Commit: двата реда влизат наведнъж. Или и двата, или нито един.
  10. Процесът пада, преди да прати нещо. Нищо не е изгубено: редът чака в outbox.
  11. След рестарта изпращачът чете outbox и намира непратения ред.
  12. Праща evt-1003 в опашката и складът го получава.
  13. Чак тогава отмята реда. Срив между двете праща реда пак, затова потребителят трябва да е идемпотентен.
  14. Записът и събитието вече не се разминават: и двете стигат, рано или късно.

След commit поръчката и редът в outbox са в базата, а опашката още е празна: процесът може да падне тук и нищо не се губи. Поръчка 1004 не оставя нищо: нито поръчка, нито ред, защото двете са в една транзакция. Изпращачът намира непратения ред и го праща.

В истинската система изпращачът е цикъл, който на всяка секунда пита базата: „дай ми непратените редове“. Затова събитието тръгва малко по-късно от commit, а не в същия миг.

Пробвай

Предвиди изхода

Изпращачът праща реда и пада, преди да го отметне. След рестарта тръгва пак. Какво има в опашката?

Последният ред на изхода е:

place_order(1003, 120)
crash = CrashScript("в изпращача")
try:
    relay()
except Crash as e:
    print(f"✗ процесът пада {e}")
crash = CrashScript()
relay()  # след рестарта
show()
Отговор

Пробвай

Намери грешката

Колега решава, че дубликатите пречат, и разменя два реда в изпращача. Сривът е на същото място, между двата реда:

Изход
✗ процесът пада в изпращача
поръчки: [1003]
outbox: [evt-1003 ✓]
опашка: []

Редът в outbox е отметнат, а опашката е празна. На кой ред е грешката?

Избери реда с грешката.

Пробвай

Подреди редовете

Подреди тялото на цикъла в изпращача така, че нищо да не се губи.

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

Идеята

Кога outbox не си струва

Outbox добавя таблица, изпращач и закъснение. Не го ползвай, когато:

  • няма отделен брокер. Ако реакциите са в същия процес и пишат в същата база (Observer в същата транзакция), една транзакция стига;
  • изгубено съобщение не вреди. Обновяване на статистика или затопляне на кеш може да се изгуби веднъж на месец, без някой да пострада;
  • събитието трябва да тръгне в същата милисекунда. Изпращачът чете на интервали. Секунда закъснение обикновено е наред, но не винаги.

А когато го ползваш, мисли за:

  • дубликатите. Изпращачът праща поне веднъж. Потребителите трябва да са идемпотентни;
  • реда. Един изпращач праща по реда на outbox. Няколко изпращача наведнъж могат да разменят реда или да пратят един ред два пъти, затова редовете се заключват при четене;
  • растежа. Пратените редове се трият след няколко дни, иначе таблицата расте вечно.

От другата страна има огледален шаблон, inbox: потребителят записва id-то на полученото събитие в своята база, в същата транзакция като работата си. Това е seen на склада от предишния урок, но в таблица.

По-точно, за любопитните

Изпращачът в истинска система. Най-простият е цикъл със заявка SELECT … FROM outbox WHERE sent = false ORDER BY id LIMIT 100. В PostgreSQL FOR UPDATE SKIP LOCKED позволява няколко изпращача, без да вземат един и същ ред. Друг начин е change data capture: Debezium чете дневника на базата (WAL в PostgreSQL, binlog в MySQL) и праща новите редове на outbox в Kafka, без заявки на интервали. Как се пуска и наблюдава Debezium или Kafka Connect е тема на DevOps курса ZeroToOps.

Готови решения. Spring Modulith има регистър на събитията, който е outbox. В .NET MassTransit и NServiceBus имат вграден outbox към Entity Framework. В Python и Node.js обикновено се пише на ръка, с таблица и малък изпращач, както в урока.

Two-phase commit. XA транзакциите (JTA в Java, System.Transactions в .NET) обещават обща транзакция през база и брокер. На практика Kafka не ги поддържа, а RabbitMQ и облачните опашки също. Затова outbox е обичайният отговор.

Защо не просто retry? „Ако пращането падне, опитай пак“ помага при грешка от брокера, но не и при срив на процеса: след рестарта никой не помни, че нещо е трябвало да тръгне. Outbox е паметта, която оцелява след срива.

Упражнение

Изпращач на партиди

Реши задачата в редактора си.

В истинската система изпращачът чете най-много няколко реда наведнъж (LIMIT в заявката). Направи relay(limit), което праща най-много limit непратени реда и връща колко е пратило. Запиши три поръчки и върти изпращача с limit 2, докато не върне 0:

Изход
кръг 1: пратени 2
кръг 2: пратени 1
кръг 3: пратени 0
опашка: 3

Още задачи

  • Отказана поръчка. Добави cancelOrder(id), който в една транзакция маха поръчката и слага в outbox OrderCancelled. Провери, че срив преди commit не маха поръчката и не праща нищо.
  • Чистене. Добави към реда момента на пращане (тик) и функция, която трие пратените редове, по-стари от 10 тика.
  • Inbox. Направи склад, който пази id-тата на получените събития и хвърля дубликата от картата „Предвиди изхода“.

Тест

Провери се

1. Защо транзакцията в базата не пази и пращането към брокера?
2. Какво записва магазинът в една транзакция, когато ползва outbox?
3. Изпращачът праща реда и пада, преди да го отметне. Какво става след рестарта?
4. Кога outbox е излишен?

Създаден от Иво Гергов като част от кампанията на Evo Tech за достъпно обучение по програмиране ·