Надёжная доставка: outbox, inbox и саги

В уроке про DI и слои данных заказ с письмом закончился фразой «письмо уезжает в outbox», и мы пообещали объяснить, почему так, а не просто await mailer.send() после коммита. Пришло время платить по счетам. Этот урок про то, как не потерять и не задвоить: почему запись в базу и публикация события не бывают атомарными сами по себе, как outbox и inbox чинят это с двух сторон, и что делать, когда бизнес-процесс растянулся на несколько сервисов и одной транзакцией уже не накрывается.

Проблема двойной записи

Сценарий из урока 13: пользователь оформляет заказ, мы сохраняем его в Postgres и отправляем письмо-подтверждение. Две операции, два внешних мира: база и почтовый сервис. Кажется, какая разница, в каком порядке их делать. Разница есть, и оба порядка ломаются.

Сначала коммит, потом отправка. Транзакция закоммичена, заказ в базе. Процесс умирает до mailer.send(): деплой, OOM, вытащенный шнур. Заказ есть, письма нет и никогда не будет: никакая строчка кода уже не выполнится, а база не помнит, что кому-то что-то обещали.

Сначала отправка, потом коммит. Письмо «ваш заказ подтверждён» ушло, а коммит упал: конфликт, нарушенный констрейнт, обрыв соединения с базой. Пользователь читает подтверждение заказа, которого не существует. Пожалуй, даже хуже первого варианта.

Это проблема двойной записи, и важно понять её фундаментальность: между Postgres и почтовым сервером не существует общей транзакции. Они не знают друг о друге, у них нет общего журнала, их нельзя закоммитить атомарно. То же самое с базой и Kafka, базой и платёжным API, базой и чем угодно вторым.

Честная попытка построить такую транзакцию в индустрии была: 2PC, распределённые транзакции с координатором. Индустрия от него в основном отказалась, и причины перекликаются с уроком про CAP: координатор это точка отказа ровно в духе задачи двух генералов (упал между фазами, и участники держат блокировки, не зная, коммитить или откатывать), все участники должны поддерживать протокол (почтовый API не умеет «подготовиться к отправке»), а блокировки на время сетевых раундтрипов убивают пропускную способность.

Выход в переформулировке задачи. Нам не нужна атомарность «запись + отправка». Нам достаточно атомарности «запись + намерение отправить», а намерение можно записать туда же, куда и заказ: в ту же базу, в ту же транзакцию. Доставкой намерения до адресата займётся отдельный процесс, в своём темпе и со своими повторами.

Outbox: событие едет в той же транзакции

Так мы приходим к паттерну transactional outbox. Механика в три шага:

1. Рядом с бизнес-таблицами заводится таблица outbox.
2. Бизнес-запись и событие вставляются в одной транзакции. Либо закоммитятся обе, либо ни одной: проблема двойной записи исчезла, потому что запись теперь одна.
3. Отдельный процесс (поллер) читает неопубликованные события из outbox, публикует их в брокер или шлёт письма и помечает отправленными.

CREATE TABLE outbox (
  id           bigserial PRIMARY KEY,
  event_type   text NOT NULL,        -- 'order.created'
  payload      jsonb NOT NULL,       -- всё, что нужно потребителю
  created_at   timestamptz NOT NULL DEFAULT now(),
  published_at timestamptz           -- NULL = ещё не уехало
);

Транзакцию ты уже видел: это ровно Unit of Work из урока 13, тот самый код, который тогда заканчивался загадочным outbox.add:

uow.run(async ({ orders, outbox }) => {
  const order = await orders.insert({ userId, ...input });
  await outbox.add({
    type: 'order.created',
    payload: { orderId: order.id, email: user.email },
  });
  return order;
});

Поллер интереснее. Его может быть несколько (инстансы приложения дублируются), и они не должны хватать одни и те же события. В Postgres для этого есть идиома FOR UPDATE SKIP LOCKED: взять пачку строк, заблокировав их, а уже заблокированные другим поллером молча пропустить:

const publishPending = () =>
  db.transaction(async (tx) => {
    const batch = await tx.execute(sql`
      SELECT id, event_type, payload
      FROM outbox
      WHERE published_at IS NULL
      ORDER BY id
      LIMIT 100
      FOR UPDATE SKIP LOCKED
    `);

    for (const event of batch) {
      await broker.publish(event.event_type, event.payload);
    }

    await tx.execute(sql`
      UPDATE outbox SET published_at = now()
      WHERE id IN ${sql.join(batch.map((e) => e.id))}
    `);
  });

Теперь честно про гарантию. Посмотри на поллер внимательно: что случится, если процесс умрёт между broker.publish и UPDATE? События уже улетели в брокер, но пометка не закоммитилась. Поллер перезапустится, увидит published_at IS NULL и опубликует их ещё раз. Outbox гарантирует at-least-once, не exactly-once: ничего не потеряется, но дубли возможны. Это не недоработка наброска, а свойство задачи, и через главу мы разберёмся с дублями на стороне получателя.

Пара слов об альтернативе поллеру: CDC. Инструменты вроде Debezium читают WAL Postgres напрямую и публикуют вставки в outbox-таблицу в Kafka без всякого поллера: меньше задержка, нет нагрузки запросами. Цена: ещё один инфраструктурный компонент со своим мониторингом. Подробнее эта механика всплывёт в разделе про data engineering; для большинства систем поллер с SKIP LOCKED это честная рабочая лошадка.

Эксплуатационный минимум, без которого outbox в проде болеет:

• Очистка. Отправленные строки копятся миллионами; нужен фоновый DELETE WHERE published_at < now() - interval '7 days' или партиционирование по времени.
• Метрика отставания. Возраст старейшей неопубликованной записи это главный индикатор здоровья: растёт, значит поллер умер или не справляется.
• Порядок. Поллер с ORDER BY id сохраняет порядок публикации, но несколько конкурентных поллеров уже нет; если порядок критичен (события одного заказа), партиционируй по ключу агрегата, об этом в следующем уроке.

Inbox и дедупликация

Перейдём на сторону получателя. Outbox честно доставил событие order.created, и, как мы выяснили, иногда доставит его дважды. Consumer, который шлёт письмо на каждое событие, отправит пользователю два письма. Consumer, который начисляет бонусные баллы, начислит их дважды, и это уже не косметика.

Сначала смиримся с теоретической частью. Tyler Treat в статье из ресурсов разбирает это подробно, краткая версия такова: exactly-once delivery невозможна. Это снова задача двух генералов: брокер не может знать, обработал ли consumer сообщение, если ack потерялся, и обязан выбрать между «возможно, потерять» (at-most-once) и «возможно, задвоить» (at-least-once). Третьего варианта нет ни у Kafka, ни у RabbitMQ, ни у кого-либо ещё, что бы ни писал маркетинг.

Но прикладному коду и не нужна exactly-once доставка. Ему нужна exactly-once обработка: пусть сообщение приходит сколько угодно раз, лишь бы эффект был один. А это уже собирается из двух деталей, обе знакомые: дедупликация плюс идемпотентность.

Механика называется inbox и зеркальна outbox-у:

const handleOrderCreated = (message: BrokerMessage) =>
  db.transaction(async (tx) => {
    // заявка на обработку: вставилось, значит мы первые
    const inserted = await tx
      .insert(processedEvents)
      .values({ eventId: message.id })
      .onConflictDoNothing()
      .returning();

    if (inserted.length === 0) return; // дубль: уже обработано, молча выходим

    // эффект в той же транзакции, что и пометка
    await accrueBonusPoints(tx, message.payload.orderId);
  });

Вся соль в том, что пометка «обработано» и сам эффект живут в одной транзакции. Упали после вставки, но до начисления: транзакция откатилась целиком, redelivery обработается заново. Пришёл дубль после успешного коммита: onConflictDoNothing вернул пустоту, выходим без эффекта. Окно для двойной обработки не существует.

Узнаёшь приём? Это idempotency key из прошлого урока, приложенный к другой границе. Там клиент защищал сервер от своих повторов ключом в заголовке; тут consumer защищает себя от повторов брокера ключом event_id. Одна и та же идея обслуживает обе стороны контракта at-least-once: отправитель не теряет, получатель не двоит. Эти два предложения, по сути, весь урок.