Go: Повтор не создаёт второго предупреждения — задача с проверкой

В этой задаче вы реализуете одну функцию, которая делает обработку события логически однократной при доставке at-least-once. Kafka может передать одно и то же сообщение более одного раза: например, если процесс упал после записи в базу данных, но до подтверждения offset. Повторная доставка не означает нового измерения: она возможна, например, после падения до подтверждённой фиксации offset. Сам факт повтора не доказывает, какой checkpoint сохранил брокер.

Разбор payload и вычисление предупреждения уже выполнены до вызова вашей функции. Вы получаете готовый Alert и непустой eventID, который совпадает с alert.SourceEventID. Ваша задача — оркестрировать одну локальную транзакцию, которая записывает уникальный marker и Alert вместе, и вернуть один из трёх исходов.

Интерфейсы

Harness предоставляет следующие типы; вы не объявляете их сами.

type Outcome uint8

const (
    OutcomeUnknown       Outcome = iota
    OutcomeApplied
    OutcomeAlreadyApplied
)

type Alert struct {
    SourceEventID string
    DeviceID      string
    Message       string
}

type Store interface {
    Begin(context.Context) (Tx, error)
}

type Tx interface {
    TryMark(context.Context, string) (bool, error)
    SaveAlert(context.Context, Alert) error
    Commit() error
    Rollback()
}

Семантика методов Tx

• TryMark(ctx, eventID) возвращает true, если уникальный marker для этого eventID был успешно стадирован в текущей транзакции и ещё не зафиксирован. Возвращает false, если marker для этого eventID уже зафиксирован в хранилище: ничего не стадируется, ошибки нет. Возвращает ошибку, если исход неизвестен; в этом случае нужно остановиться.
• SaveAlert(ctx, alert) стадирует ровно один Alert в той же транзакции. Вызывается только после TryMark вернул true. До успешного Commit Alert не считается устойчивым.
• Commit() атомарно фиксирует стадированный marker и Alert вместе. Только успешный Commit означает OutcomeApplied. Ошибка Commit может быть неоднозначной: транзакция могла быть применена или не применена на стороне хранилища; Rollback после этого является очисткой клиентского состояния и не доказывает, что сервер откатил её.
• Rollback() вызывается один раз для любого пути после успешного Begin, который не завершился подтверждённым успешным Commit. Это включает ветку дублирования и все ветки ошибок.

Поведение функции

Вы реализуете:

func ProcessOnce(ctx context.Context, store Store, eventID string, alert Alert) (Outcome, error)

Ожидаемое поведение по ветвям:

• Первая доставка: Begin → TryMark возвращает true → SaveAlert → Commit возвращает nil → вернуть OutcomeApplied, nil. После подтверждённого успешного Commit Rollback не вызывается.
• Повторная доставка того же eventID: Begin → TryMark возвращает false → Rollback → вернуть OutcomeAlreadyApplied, nil. SaveAlert и Commit на этой ветке не вызываются.
• Ошибка Begin: вернуть OutcomeUnknown и ту же ошибку. Никакие методы Tx недоступны и не вызываются.
• Ошибка TryMark: вызвать Rollback один раз, вернуть OutcomeUnknown и ту же ошибку. SaveAlert и Commit не вызываются.
• Ошибка SaveAlert: вызвать Rollback один раз, вернуть OutcomeUnknown и ту же ошибку. Commit не вызывается.
• Ошибка Commit: вызвать Rollback один раз как очистку, вернуть OutcomeUnknown и ту же ошибку. Нельзя объявлять пару отсутствующей или устойчивой только на основании ошибки.

Критическое требование к ошибкам: возвращайте точно то значение ошибки, которое вернул store.Begin, tx.TryMark, tx.SaveAlert или tx.Commit — без оборачивания и без замены. Harness внедряет сравнимые sentinel-ошибки и проверяет returnedErr == injectedErr через сравнение значений интерфейса. Любое fmt.Errorf("...: %w", err) сломает проверку.

Два одинаковых сообщения

Fake-store harness проверяет следующий сценарий: два вызова ProcessOnce с одним и тем же eventID и одним и тем же Alert. Первый вызов должен вернуть OutcomeApplied, второй — OutcomeAlreadyApplied. После обоих вызовов в fake-хранилище должен быть ровно один marker и ровно один Alert для этого eventID. Это локальный инвариант fake-транзакций; он не является демонстрацией реального PostgreSQL или гарантией Kafka.

Harness также проверяет порядок вызовов, количество вызовов каждого метода, наличие Rollback на нужных путях и отсутствие SaveAlert и Commit на ветке дублирования и ветках ошибок.

Граница с Kafka

Ваша функция не взаимодействует с Kafka. Внешний consumer подтверждает offset только после того, как ProcessOnce вернул nil-ошибку с исходом OutcomeApplied или OutcomeAlreadyApplied. При любой ошибке consumer останавливается без подтверждения offset. Подтверждение offset — это отдельный вызов, не входящий в локальную транзакцию; этот вызов в данной функции не реализуется.

SQL-вопрос об атомарности

После теории и двух вопросов вы перешли к CODE-задаче; SQL-вопрос об атомарности уже был среди этих вопросов. В нём нужно объяснить, почему marker и Alert должны фиксироваться в одной транзакции, а не двумя отдельными Commit. Для приведённой ниже схемы рассмотрите marker-first и последствия падения между фиксациями. Для alert-first сначала решите, допускает ли показанный внешний ключ фиксацию Alert для нового event_id до marker. Вариант alert-first с возможной повторной вставкой Alert относится только к отдельной гипотетической схеме без этого внешнего ключа и без уникальности source_event_id, а не к DDL ниже. Учебная схема Alerts DB, которую иллюстрирует этот вопрос, выглядит следующим образом:

CREATE TABLE processed_reading_events (
    event_id text PRIMARY KEY
);

CREATE TABLE alerts (
    source_event_id text PRIMARY KEY
        REFERENCES processed_reading_events(event_id),
    device_id       text NOT NULL,
    message         text NOT NULL
);

Атомарная попытка вставить marker в этой схеме использует конструкцию:

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

Если строка возвращена — marker новый. Если строка не возвращена (sql.ErrNoRows) — marker уже зафиксирован, Alert не вставляется. Оба INSERT выполняются через один sql.Tx; только успешный tx.Commit делает пару устойчивой. Fake-store проверяет оркестрацию и детерминированное состояние; он не воспроизводит реальные изоляцию, блокировки или устойчивость PostgreSQL — это предмет отдельного SQL-вопроса.

Начальный код

func ProcessOnce(ctx context.Context, store Store, eventID string, alert Alert) (Outcome, error) {
    // Implement the supplied transaction contract.
    return OutcomeUnknown, nil
}