Exactly-once semantics

Зачем это нужно

«Exactly-once» — самая хайповая и одновременно самая недопонимаемая тема в distributed systems. Половина инженеров считает это маркетингом («delivery exactly-once невозможен в network — Two Generals Problem»), другая половина думает «в Kafka есть флаг, включил и всё». Истина посередине: exactly-once delivery невозможен, но exactly-once effects (processing) — реальная и достижимая цель.

Цена ошибки огромная. Платежи на at-least-once без идемпотентности — двойные списания. Inventory — overselling и отрицательные остатки. Notifications — 5 SMS вместо одной. На больших масштабах 0.1% дублей = тысячи обращений в support и прямой денежный убыток. С другой стороны, exactly-once не бесплатно: транзакции Kafka добавляют 100-200 ms latency, режут throughput на 10-20%, 2PC sink стоит CPU и удваивает checkpoint cost.

Понимание EOS = понимание где именно ты закрываешь границу ошибок: producer (idempotent send), broker (transactions), consumer (manual offset commit), sink (idempotent write / 2PC). Без этой ментальной модели ты либо overengineer'ишь там, где at-least-once + dedup достаточно, либо теряешь и дублируешь данные там, где это критично.

Mental model

«Exactly-once delivery в network невозможен (Two Generals); но exactly-once processing effects = at-least-once delivery + idempotent consumer (operation_id-based dedup) + atomic commit (output вместе с offset/state в одной транзакции). Все три ингредиента обязательны.»

Что на диаграмме

Топология показывает полный pipeline EOS и его участников:

• Producers — Producer (idempotent + txn) с PID/epoch и Flink Source, читающий из Kafka.
• Kafka Cluster — два broker'а (leader topic A/B) и Tx Coordinator, держащий state в __transaction_state.
• Flink JobManager + Operators — JobManager рассылает checkpoint barriers, Operator обрабатывает поток, 2PC Sink коммитит атомарно с external storage.
• Consumers & Sinks — Consumer (read_committed), Redis (dedup set, TTL 24h), External API (Idempotency-Key), PostgreSQL (MERGE / XA).

Edges — это физические соединения, ответы идут по тем же edges в обратном направлении (reverse animation). Например, prod → tc отвечает за initTransactions и за commitTransaction; abort markers едут от координатора во все partitions через те же edges.

Сценарии

В скрипте восемь сценариев — от «как сломать» до «как правильно»:

1. at-least-once duplicate — классический double-billing: producer не получает ACK, retry'ит, broker без idempotence пишет второй раз, consumer списывает деньги дважды. Это baseline ошибки, против которой работают все остальные сценарии.
2. kafka-eos — happy path: initTransactions (PID, epoch), send в несколько partitions, sendOffsetsToTransaction (consumer offsets в той же txn), двухфазный commit (PrepareCommit в txn log → commit markers во все partitions), read_committed consumer видит данные только после marker, dedup через Redis, idempotent POST.
3. idem-retry — idempotent producer: ACK потерян, producer переотправляет с тем же (PID, seq=2), broker узнаёт дубль и тихо возвращает success без записи. Это решает только проблему ретрая одного инстанса.
4. txn-abort — producer crash перед commit → transaction.timeout.ms истекает → координатор пишет Abort marker → consumer с read_committed фильтрует aborted batch (LSO не двигается) → ноль downstream effects.
5. flink-2pc — Flink TwoPhaseCommitSinkFunction: checkpoint barrier (Chandy-Lamport snapshot) → sink делает pre-commit (XA prepare + Kafka begin), но данные не видны → JobManager собирает ACK от всех операторов → notifyCheckpointComplete → sink делает commit → данные visible downstream.
6. flink-recovery — operator падает после pre-commit, до notifyCheckpointComplete. JobManager детектит, sink делает abort/rollback pending транзакций, job рестартует с последнего completed checkpoint, процессинг повторяется — но без duplicate effects благодаря 2PC.
7. idem-consumer — паттерн для non-transactional sink: dedup на consumer через Redis (SISMEMBER → skip / SADD ... EX 86400 → process). Работает с любым sink, цена — extra round-trip и dedup storage.
8. saga-needed — честный сценарий, когда EOS невозможен: внешний API не поддерживает Idempotency-Key. Решение — требовать idempotency на стороне sink, либо Saga + compensations.

Trade-offs (ADR)

ADR-001: Когда EOS критично vs at-least-once + idempotency достаточно. Транзакционный producer + read_committed добавляют 100-200 ms latency на commit и режут throughput на 10-20%. Idempotent consumer через Redis — extra round-trip и storage. Не везде это оправдано. Дерево выбора:

• Side effect monetary / regulatory (charge, refund, ledger entry, декремент inventory) — обязательно EOS effects: либо Kafka transactions + idempotent sink, либо at-least-once + Idempotency-Key на consumer.
• Side effect user-visible но не critical (notification, email) — at-least-once + dedup table с TTL 7 дней. Пользователь переживёт редкий дубль, бизнес — тоже.
• Side effect internal (analytics, materialized view rebuild) — at-least-once + idempotent MERGE/UPSERT в sink. Нет dedup-state, нет отдельного хранилища.
• Pure observability (метрики, логи) — at-most-once OK, accept loss.

Правило: чем дороже дубль (в деньгах или доверии), тем выше уровень гарантии. EOS через Kafka transactions включаем только для inter-Kafka pipelines (Streams, Flink read-process-write); для external sinks предпочитаем idempotent writes (MERGE, Idempotency-Key, conditional update) — они проще, дешевле, не зависят от distributed coordinator.

ADR-002: Effective EOS = at-least-once delivery + idempotent consumer + atomic commit. Two Generals Problem доказывает: exactly-once delivery в network невозможен. Producer не может узнать, дошёл ли пакет с ACK обратно. Любая EOS-система на самом деле гарантирует exactly-once effects через дедупликацию + atomic state commit. Маркетинг от Kafka/Flink этот нюанс скрывает, разработчики думают, что «поставил флаг — получил магию».

Mental model команды — EOS состоит из трёх компонентов, все обязательны:

1. at-least-once delivery (producer retry + consumer manual offset commit) — гарантирует no loss.
2. idempotent consumer (dedup по operation_id, либо UPSERT/MERGE по unique key, либо conditional update) — гарантирует no duplicate effect.
3. atomic commit (output + offset/state в одной транзакции, либо 2PC sink, либо sendOffsetsToTransaction) — гарантирует, что после crash replay не создаст phantom output.

Если хоть один компонент отсутствует — это не EOS, не называйте это так в дизайн-доках. В code review требуем явного указания всех трёх для любого flow с monetary side effect.

Реальные системы

• Stripe Idempotency-Key header — golden standard для public APIs. Каждый POST принимает Idempotency-Key: <uuid>, Stripe сохраняет результат (status + body) на 24 часа, повторный запрос возвращает закэшированный ответ.
• Kafka Streams / Confluent Cloud — processing.guarantee=exactly_once_v2 по умолчанию для внутренних pipeline'ов. v2 объединяет транзакции per app instance вместо per task — меньше overhead.
• Flink + Kafka на Uber — большинство стримовых job используют EOS для consistency между метриками и операционными системами; критично там, где данные стрима попадают в payouts.
• Netflix Keystone — at-least-once + dedup downstream (Druid имеет встроенную дедупликацию по event_id). Сознательный выбор: EOS не везде оправдан.
• Pinterest Singer — at-least-once log delivery, downstream процессоры дедуплицируют через event_id в Druid/ClickHouse.
• Snowflake / BigQuery MERGE INTO — sink использует MERGE с unique key, получая idempotent insert. Популярный pattern для ETL без транзакций.
• Iceberg / Delta Lake / Hudi — snapshot-based atomic commits на уровне manifest файла, естественная EOS-семантика для batch и streaming sink.

Anti-patterns

• «Включил enable.idempotence=true — теперь у меня EOS». Нет. Idempotent producer защищает только от повторов одного producer'а на retry в окне последних 5 сообщений per partition. Если producer упадёт и перезапустится без transactional.id — новый PID, дубли возможны.
• acks=1 + idempotence. Не работает. Idempotence требует acks=all.
• Transactions на каждое сообщение. Каждая транзакция — +20-50 ms latency. На high-throughput собирай батч в одну транзакцию, commit'ай по времени или размеру.
• Long-running transactions (минуты-часы). transaction.timeout.ms (default 60 сек) сработает, координатор сделает abort, вся работа потеряна. Дизайн: одна txn на маленький batch, commit часто.
• Использовать EOS для at-most-once use cases (logs, metrics). Лишний overhead, ноль ценности.
• Забыть isolation.level=read_committed на consumer. Producer делает transactions, consumer читает с дефолтным read_uncommitted → видит aborted данные → EOS сломан незаметно.
• Один transactional.id на много инстансов. Epoch fencing убьёт работающего producer'а каждый раз, когда другой стартует. Используй уникальный txid per instance (app-${pod_name}).
• Idempotent consumer без TTL на dedup set. Set растёт бесконечно → OOM в Redis за недели. TTL = max replay window + safety (24-72 часа).
• Reverse edges для ответов в топологии. Анимация ответов идёт по тем же forward edges — не плоди дублирующие connections, обходящие координатор.

Когда НЕ использовать

• External non-idempotent HTTP API без Idempotency-Key. Если внешний сервис не поддерживает дедупликацию — EOS-effects невозможен. Не делай вид, что есть. Требуй idempotency от вендора, либо строй Saga с явными compensating actions.
• Multi-system writes (write to DB и send email и charge card). Распределённой транзакции через гетерогенные системы нет. Решение — transactional outbox + saga, EOS закрывается per-step.
• Best-effort метрики, логи, телеметрия. at-most-once / at-least-once дешевле, accept loss или редкие duplicates. EOS overhead не оправдан.
• Очень короткий time-to-market прототип. EOS добавляет операционной сложности (coordinator monitoring, fence diagnostics, replay scenarios). Сначала at-least-once + dedup, потом — при росте — переход на транзакции.
• External sinks с native upsert. Если есть MERGE / UPSERT / conditional PUT — это уже idempotent write. Транзакции Kafka сверху не нужны, проще и дешевле обойтись.

Дальше читать

• idempotency — фундамент любого EOS: Idempotency-Key, dedup tables, conditional updates.
• kafka-deep — глубоко про partitions, ISR, KRaft, transaction coordinator под капотом.
• transactional-outbox — паттерн атомарной записи DB + event для multi-system writes.
• apache-flink-deep — Chandy-Lamport snapshots, TwoPhaseCommitSinkFunction, savepoint vs checkpoint.
• stream-processing — общая модель: state, time, windowing — фундамент для понимания EOS в стримовых системах.
• Confluent blog (Jun Rao): «Exactly-once Semantics Are Possible: Here's How» — главная статья про KIP-98.
• Flink docs: «End-to-end Exactly-Once with Apache Flink and Kafka».
• Stripe blog: «Designing robust and predictable APIs with idempotency».
• Tyler Akidau, "Streaming Systems" Ch.5 (Exactly-Once and Side Effects).
• Two Generals' Problem — формальное основание, почему delivery невозможен.