History помнит обработанный event_id

Два прихода одного event-ticket-900 и зачем History нужна своя БД

В уроке 8.3 outbox publisher получил Kafka ack для event-ticket-900, но упал до того, как UPDATE published_at был устойчиво записан в PostgreSQL Ticket DB. Строка outbox осталась с published_at IS NULL; после перезапуска publisher может выбрать её снова. Если повторный Publish тоже получит ack, в топике tickets.status.v1 окажутся две подтверждённые копии с одинаковым event_id='event-ticket-900'. Ниже рассматриваем именно этот условный случай двух успешных доставок.

Это не два разных перехода Ticket — это один и тот же логический факт, доставленный дважды. Обе доставки содержат один и тот же сохранённый payload outbox-строки:

{
  "event_id": "event-ticket-900",
  "ticket_id": "ticket-42",
  "from_status": "open",
  "to_status": "resolved",
  "changed_at": "2026-10-03T08:00:00Z"
}

History service получает оба сообщения из Kafka. Его задача — сохранить проекцию переходов Ticket, по одной строке на логическое событие. Без дополнительной защиты второй приход event-ticket-900 создаст вторую строку history_events с теми же данными, то есть появится дубль в локальной проекции.

History владеет собственной PostgreSQL БД с таблицами inbox и history_events. Ни Ticket DB, ни Kafka не участвуют в этой транзакции — History не может просто «спросить» у Ticket DB, был ли этот event уже обработан, потому что Ticket DB вообще не знает о проекции History. Kafka offset тоже не годится как признак обработки: он отражает позицию consumer в топике, а не факт успешного сохранения в History DB. Partition группирует записи и может содержать события многих Ticket; key ticket-42 и ticket_id указывают на этот Ticket, но ни partition, ни key, ни ticket_id не различают его логические события: у двух разных событий одного Ticket может быть одинаковый key.

Единственный стабильный ключ дедупликации — event_id, который фиксируется в outbox row при создании и не меняется при повторных доставках. Инвариант, который History обязан поддерживать: для каждого event_id в history_events существует не более одной строки.

Схема inbox + history_events и атомарная вставка в одной транзакции

Чтобы инвариант «одна строка на event_id» был гарантирован даже при конкурентных доставках, недостаточно проверить наличие строки через SELECT перед INSERT: два consumer могут одновременно прочитать «пусто» и оба вставить дубль. Настоящим арбитром становится уникальное ограничение PostgreSQL.

CREATE TABLE inbox (
    event_id text PRIMARY KEY
);

CREATE TABLE history_events (
    event_id   text        PRIMARY KEY REFERENCES inbox(event_id),
    ticket_id  text        NOT NULL,
    from_status text       NOT NULL,
    to_status  text        NOT NULL,
    changed_at timestamptz NOT NULL
);

PRIMARY KEY на inbox.event_id создаёт уникальный индекс. history_events.event_id тоже объявлен PRIMARY KEY и ссылается на inbox через внешний ключ — это дополнительно гарантирует, что строка истории не появится без маркера в inbox. При этом ticket_id в history_events не уникален: у одного Ticket могут быть несколько логических событий с разными event_id, и все они должны сохраняться.

В одной sql.Tx History выполняет два оператора по порядку.

Первый — попытка занять маркер в inbox:

INSERT INTO inbox(event_id)
VALUES ($1)
ON CONFLICT (event_id) DO NOTHING
RETURNING event_id;

ON CONFLICT (event_id) DO NOTHING означает: если строка с таким event_id уже есть, INSERT молча ничего не делает и не возвращает строк. RETURNING event_id возвращает вставленную строку только если INSERT действительно добавил новую запись. В Go это различие ловится через QueryRowContext(...).Scan(...): для нового ID Scan прочитает значение, для дубля вернёт sql.ErrNoRows.

Второй оператор выполняется только если Scan вернул новый ID — то есть в той же sql.Tx, сразу после подтверждённого нового маркера:

INSERT INTO history_events(event_id, ticket_id, from_status, to_status, changed_at)
VALUES ($1, $2, $3, $4, $5);

Оба INSERT исполняются внутри одной транзакции. Commit() атомарно применяет пару: либо в базе появляются и маркер, и строка истории, либо не появляется ничего. Это исключает ситуацию, когда маркер есть, а строки истории нет, или наоборот.

Важно понять, чем плохи два отдельных коммита. Если сначала закоммитить inbox, а потом сохранять History, падение между коммитами оставит маркер без строки истории: следующий повтор увидит false от TryInbox и посчитает событие уже обработанным, хотя History row так и не появилась. Обратный порядок под показанной схемой не создаст устойчивую History-строку без маркера: внешний ключ history_events.event_id REFERENCES inbox(event_id) отклонит INSERT до появления inbox-строки. Это не делает два отдельных коммита правильными: допустимый inbox-first порядок всё ещё может оставить маркер без истории. Оба INSERT нужны в одной транзакции.

Три исхода TryInbox и правильный порядок вызовов в Go

ProcessHistoryOnce работает через интерфейс репозитория и может получить от него ровно три логических ответа. Понять их различие важно, прежде чем писать orchestration-код.

Когда TryInbox(ctx, event.EventID) возвращает true, nil, inbox не знал об этом event_id: маркер только что создан внутри текущей транзакции, но ещё не зафиксирован. History row тоже не сохранена. Только после SaveHistory и подтверждённого Commit() == nil исход становится OutcomeApplied.

Когда TryInbox возвращает false, nil, в базе уже есть закоммиченная пара inbox-маркер + History row для этого event_id. Новое состояние не создаётся: SaveHistory и Commit вызывать не нужно. Транзакцию нужно закрыть через Rollback как cleanup и вернуть OutcomeAlreadyApplied, nil. Это не ошибка — это подтверждённый дубль.

Когда любой метод — Begin, TryInbox, SaveHistory или Commit — возвращает ненулевую ошибку, результат неизвестен. Нельзя утверждать, что пара сохранена или что её нет. Исход — OutcomeUnknown с тем же самым error-значением, без обёртки.

Отдельный случай — ошибка Commit после того, как fake применил пару (модель потерянного ответа сети). Rollback вызывается как cleanup, но уже не может отменить применённые данные. Функция всё равно возвращает OutcomeUnknown и injected error: она не может знать, случился ли commit на стороне сервера. При следующей доставке того же event-ticket-900 TryInbox вернёт false, nil — и исход станет OutcomeAlreadyApplied. Пара в базе по-прежнему одна.

Отличие event-ticket-901 принципиально. У него другой event_id, хотя ticket_id тот же — ticket-42, переход resolved → closed, время 2026-10-03T08:01:00Z. TryInbox не найдёт конфликта, вставит новый маркер, SaveHistory добавит вторую строку в history_events. После коммита в базе будут два маркера и две строки истории — по одной на каждый логический переход одного Ticket. Это корректное поведение, не дубль.

Три результата удобно увидеть как данные, без реализации сдаваемой функции:

package main

import (
    "errors"
    "fmt"
)

func main() {
    observations := []struct {
        inserted bool
        err      error
        meaning  string
    }{
        {true, nil, "новый маркер подготовлен, но ещё не закоммичен"},
        {false, nil, "пара inbox + History уже была закоммичена"},
        {false, errors.New("storage unavailable"), "исход неизвестен"},
    }
    for _, observation := range observations {
        fmt.Println(observation.inserted, observation.err, observation.meaning)
    }
}

Это иллюстрация значений, а не код транзакции. После успешного Begin ветка дубля и каждый путь без подтверждённого Commit() == nil требуют ровно одного Rollback как cleanup; после успешного Commit вызывать Rollback нельзя по контракту fake. При ошибке Commit cleanup не доказывает отсутствие применённой пары. Внешний consumer не продвигает Kafka offset по OutcomeUnknown. Полный порядок методов и возвратов предстоит реализовать в CODE-шаге.

Crash после локального Commit, replay и граница SQL-атомарности

Предположим, ProcessHistoryOnce вернула OutcomeApplied, nil для event-ticket-900. Пара inbox + history_events закоммичена в локальной History DB. Теперь внешний consumer вправе подтвердить позицию за этим сообщением, если все более ранние offsets этой partition разрешены. В рассматриваемой трассе процесс упал до такого подтверждения.

Если запись всё ещё хранится в Kafka, а сохранённая позиция consumer group остаётся перед её offset, после рестарта возможен повторный fetch event-ticket-900. Тогда TryInbox возвращает false, nil: подтверждённая пара уже есть в базе. Повторная обработка вернёт OutcomeAlreadyApplied, nil, не вызовет SaveHistory и не создаст вторую строку в history_events. Инвариант сохранён при этих условиях.

Такой replay после crash before offset commit не требует отдельной правки History-строки: уникальный индекс PostgreSQL распознаёт сохранённый event_id. Но SQL-инвариант не гарантирует, что Kafka сохранит запись до рестарта или что group всё ещё сможет прочитать её по прежней позиции.

Теперь обратный порядок. Если consumer подтвердит позицию за этим сообщением до успешной локальной DB-операции, а затем обработка не завершится, обычное чтение этой group начнётся после него: History рискует потерять событие. Ручной seek или новая group могли бы прочитать ещё хранимую запись, поэтому утверждать, что повтор невозможен вообще, нельзя. OutcomeApplied, nil и OutcomeAlreadyApplied, nil только разрешают внешнему consumer рассмотреть продвижение; checkpoint по partition должен оставаться непрерывным и не перескакивать через более ранний нерешённый offset.

OutcomeUnknown блокирует продвижение checkpoint за offset этого сообщения в данной partition, даже если более позднее сообщение уже дало OutcomeApplied или OutcomeAlreadyApplied. Если запись сохранится и group checkpoint останется перед ней, повторная доставка возможна: после отката TryInbox может вставить маркер заново; после состоявшегося, но неподтверждённого клиенту Commit он обнаружит уже сохранённую пару. Без этих условий повтор нельзя обещать.

Граница, которую локальная SQL-транзакция не пересекает: она атомарна для двух таблиц в одной PostgreSQL БД, но не имеет никакого отношения к Kafka offset. Нет общей транзакции «PostgreSQL + Kafka». Срок хранения Kafka ограничен настроенной retention-политикой; повторная доставка возможна, пока запись доступна и положение group позволяет её читать. Поэтому локальная дедупликация нужна, но сама по себе не обеспечивает replay.

Вопрос об атомарности, который разбирается в CHOICE-шаге, звучит так: что произойдёт, если inbox INSERT закоммитить отдельной транзакцией до того, как History INSERT даже начнётся, а процесс упадёт между ними? Ответ следует из схемы выше:

Статичная схема (PNG)

-- Отдельная транзакция 1: маркер закоммичен
BEGIN;
INSERT INTO inbox(event_id) VALUES ('event-ticket-900');
COMMIT;

-- Процесс падает. History INSERT так и не выполнен.

-- При повторной доставке:
BEGIN;
INSERT INTO inbox(event_id) VALUES ('event-ticket-900')
ON CONFLICT (event_id) DO NOTHING
RETURNING event_id;
-- Возвращает 0 строк: маркер уже есть.
-- TryInbox -> false -> AlreadyApplied.
-- History row никогда не создана.
ROLLBACK;

Маркер без History row — это потеря события. Повтор не исправит ситуацию, потому что inbox уже считает event-ticket-900 обработанным. Именно поэтому оба INSERT должны быть в одной транзакции, а не в двух последовательных.