Как реализовать гарантию доставки ровно один раз?

Как реализовать гарантию доставки ровно один раз

Это одна из самых сложных задач в распределённых системах. "Ровно один раз" = exactly-once.

Три типа гарантий доставки

1. At-most-once (максимум один раз)
   - Сообщение может быть потеряно
   - Пример: UDP

2. At-least-once (как минимум один раз)
   - Сообщение дойдёт, но может быть дублировано
   - Требует идемпотентности
   - Пример: RabbitMQ, Kafka

3. Exactly-once (ровно один раз) ← Самая сложная!
   - Дойдёт ровно один раз

Решение 1: Идемпотентные операции + дедупликация

class PaymentService:
    def process_payment(self, order_id: str, amount: float, payment_id: str):
        # payment_id — уникальный ID платежа
        existing = self.db.query(Payment).filter_by(
            payment_id=payment_id
        ).first()
        
        if existing:
            return existing  # Идемпотентно возвращаем
        
        payment = Payment(
            id=payment_id,
            order_id=order_id,
            amount=amount,
            status='completed'
        )
        self.db.add(payment)
        self.db.commit()
        return payment

# Клиент отправляет с идентификатором
payment_id = str(uuid.uuid4())

# Первый вызов
result1 = client.process_payment('order-123', 100.0, payment_id)

# Повтор (сеть упала)
result2 = client.process_payment('order-123', 100.0, payment_id)

# Обе операции идемпотентны
assert result1.id == result2.id

Решение 2: Outbox Pattern (самое практичное!)

class OrderService:
    def create_order(self, user_id: int, items: list):
        try:
            order = Order(
                id=str(uuid.uuid4()),
                user_id=user_id,
                items=items
            )
            self.db.add(order)
            
            # Записываем событие в outbox в ОДНОЙ транзакции
            outbox_event = OutboxEvent(
                id=str(uuid.uuid4()),
                order_id=order.id,
                event_type='order.created',
                payload={
                    'order_id': order.id,
                    'user_id': user_id,
                    'items': items
                },
                status='pending'
            )
            self.db.add(outbox_event)
            self.db.commit()  # Одна транзакция!
            return order
        
        except Exception:
            self.db.rollback()
            raise

# Отдельный процесс читает outbox и отправляет события
class OutboxProcessor:
    def process(self):
        events = self.db.query(OutboxEvent).filter_by(
            status='pending'
        ).limit(100).all()
        
        for event in events:
            try:
                self.event_bus.publish(event.event_type, event.payload)
                event.status = 'sent'
                self.db.commit()
            except Exception:
                logger.error(f"Failed to send event {event.id}")

Решение 3: Kafka с транзакциями

from kafka import KafkaProducer

class KafkaExactlyOnce:
    def __init__(self):
        self.producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            transactional_id='my-producer'  # Ключ для exactly-once
        )
    
    def process_order_exactly_once(self, order_data):
        with self.producer.transaction():
            # В одной транзакции Kafka:
            result = self.process_order(order_data)
            self.producer.send('order-processed', value=result)
            # Откатывается или коммитится атомарно

Решение 4: Saga Pattern (микросервисы)

class OrderSaga:
    def create_order(self, order_id: str, items):
        try:
            # Шаг 1: Зарезервировать товар
            self.inventory_service.reserve(items)
            
            # Шаг 2: Обработать платёж
            self.payment_service.charge(order_id)
            
            # Шаг 3: Отправить заказ
            self.shipping_service.ship(order_id)
            
        except Exception:
            # Компенсирующие транзакции
            self.inventory_service.release(items)
            self.payment_service.refund(order_id)
            raise

Решение 5: REST API с идемпотентностью

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()

@app.post("/orders")
def create_order(
    order_data: OrderCreate,
    idempotency_key: str = Header(None)
):
    # Клиент отправляет: Idempotency-Key: uuid-123
    if not idempotency_key:
        raise HTTPException(status_code=400)
    
    # Проверяем: был ли уже такой заказ?
    existing = db.query(Order).filter_by(
        idempotency_key=idempotency_key
    ).first()
    
    if existing:
        return existing  # Идемпотентно
    
    order = Order(
        id=str(uuid.uuid4()),
        idempotency_key=idempotency_key,
        **order_data.dict()
    )
    db.add(order)
    db.commit()
    return order

Когда каждый подход

Идемпотентные операции:
- Простые операции
- REST API
- Платежи

Outbox Pattern:
- Микросервисы
- Event sourcing
- Message-driven

Kafka транзакции:
- Потоковые обработки
- Real-time системы

Saga Pattern:
- Распределённые транзакции
- Долгоживущие процессы

Ключевые факты

• Exactly-once в распределённых системах очень сложно
• На практике: idempotency key + at-least-once = exactly-once эффект
• Outbox pattern — золотой стандарт
• Kafka генерирует unique offset для каждого сообщения
• Идемпотентность = ключ к exactly-once