Transactional Outbox

Проблема: dual-write

Типовой сценарий -- создать запись в БД и отправить событие в Kafka:

// НЕ делайте так
$db->beginTransaction();
$db->exec("INSERT INTO orders ...");
$db->commit();

$kafka->publish('orders.created', $event); // а если сеть моргнула?

Четыре возможных сценария при сбое:

Сценарий	БД	Kafka	Последствие
Happy path	OK	OK	Консистентно
BD fail	FAIL	-	OK, ничего не произошло
Сеть умерла после commit	OK	FAIL	Заказ есть, события нет, склад не узнает
Сеть умерла после publish	???	OK	Событие послано, а commit? Падение перед commit = есть событие о несуществующем заказе

Поменяйте порядок -- проблема остаётся, только зеркально. Это и есть dual-write problem: две независимые системы не дают атомарности.

Решение: Transactional Outbox

Идея: записывать событие в ту же БД, что и основные данные, в одной транзакции. Отдельный worker читает таблицу outbox и публикует в Kafka.

┌─────────────────────────────┐
│      API Handler            │
│                             │
│   BEGIN TRANSACTION         │
│   INSERT INTO orders ...    │
│   INSERT INTO outbox ...    │
│   COMMIT                    │  <-- атомарно
└──────────┬──────────────────┘
           │
           │ (записано)
           v
┌─────────────────────────────┐
│    PostgreSQL               │
│  orders | outbox            │
└──────────┬──────────────────┘
           │
           │ pg_notify / polling
           v
┌─────────────────────────────┐
│   Outbox Publisher Worker   │
│   SELECT FOR UPDATE SKIP    │
│   Publish to Kafka          │
│   UPDATE sent_at = NOW()    │
└──────────┬──────────────────┘
           │
           v
         Kafka

Гарантия: at-least-once -- событие будет опубликовано, возможно несколько раз (если worker упал между publish и UPDATE). Консьюмеры должны быть идемпотентны (см. inbox pattern ниже).

Схема outbox таблицы

CREATE TABLE outbox (
    id           BIGSERIAL PRIMARY KEY,
    aggregate_type TEXT NOT NULL,       -- 'order', 'user', etc
    aggregate_id   TEXT NOT NULL,       -- ID сущности
    event_type     TEXT NOT NULL,       -- 'OrderCreated'
    payload        JSONB NOT NULL,      -- тело события
    headers        JSONB,               -- correlation_id, trace_id
    created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    sent_at        TIMESTAMPTZ,         -- NULL = pending
    attempts       INT NOT NULL DEFAULT 0,
    error          TEXT
);

-- Индекс для worker: берёт только pending, в порядке создания
CREATE INDEX idx_outbox_pending ON outbox (created_at)
WHERE sent_at IS NULL;

-- По aggregate для отладки
CREATE INDEX idx_outbox_agg ON outbox (aggregate_type, aggregate_id);

Важно: FIFO per aggregate -- для одного aggregate_id события должны публиковаться в порядке вставки. На Kafka решается через partition key = aggregate_id.

Producer в handler

<?php

declare(strict_types=1);

namespace App\Outbox;

use Doctrine\DBAL\Connection;

/**
 * Outbox writer. Call within the same transaction as your main INSERT/UPDATE.
 */
final class OutboxWriter
{
    public function __construct(private readonly Connection $conn) {}

    /**
     * @param array<string, mixed> $payload
     * @param array<string, mixed> $headers
     */
    public function append(
        string $aggregateType,
        string $aggregateId,
        string $eventType,
        array $payload,
        array $headers = [],
    ): void {
        $this->conn->insert('outbox', [
            'aggregate_type' => $aggregateType,
            'aggregate_id'   => $aggregateId,
            'event_type'     => $eventType,
            'payload'        => json_encode($payload, JSON_THROW_ON_ERROR),
            'headers'        => json_encode($headers, JSON_THROW_ON_ERROR),
        ]);
    }
}

final class CreateOrderHandler
{
    public function __construct(
        private readonly Connection $conn,
        private readonly OutboxWriter $outbox,
    ) {}

    public function handle(CreateOrderCommand $cmd): string
    {
        $this->conn->beginTransaction();
        try {
            $orderId = bin2hex(random_bytes(16));

            // Main business write
            $this->conn->insert('orders', [
                'id' => $orderId,
                'user_id' => $cmd->userId,
                'total_cents' => $cmd->totalCents,
                'status' => 'created',
            ]);

            // Outbox write in the same transaction -- atomic!
            $this->outbox->append(
                aggregateType: 'order',
                aggregateId: $orderId,
                eventType: 'OrderCreated',
                payload: [
                    'orderId' => $orderId,
                    'userId' => $cmd->userId,
                    'totalCents' => $cmd->totalCents,
                ],
                headers: [
                    'correlationId' => $cmd->correlationId,
                    'occurredAt' => (new \DateTimeImmutable())->format(DATE_RFC3339_EXTENDED),
                ],
            );

            $this->conn->commit();
            return $orderId;
        } catch (\Throwable $e) {
            $this->conn->rollBack();
            throw $e;
        }
    }
}

using System.Text.Json;
using Npgsql;

namespace App.Outbox;

/// Outbox writer. Call within the same transaction as your main INSERT/UPDATE.
public sealed class OutboxWriter
{
    public async Task AppendAsync(
        NpgsqlConnection conn,
        NpgsqlTransaction tx,
        string aggregateType,
        string aggregateId,
        string eventType,
        object payload,
        IReadOnlyDictionary<string, string>? headers,
        CancellationToken ct)
    {
        await using var cmd = new NpgsqlCommand(
            """
            INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload, headers)
            VALUES (@aggregateType, @aggregateId, @eventType, @payload::jsonb, @headers::jsonb)
            """, conn, tx);

        cmd.Parameters.AddWithValue("aggregateType", aggregateType);
        cmd.Parameters.AddWithValue("aggregateId", aggregateId);
        cmd.Parameters.AddWithValue("eventType", eventType);
        cmd.Parameters.AddWithValue("payload", JsonSerializer.Serialize(payload));
        cmd.Parameters.AddWithValue("headers", JsonSerializer.Serialize(headers ?? new Dictionary<string, string>()));

        await cmd.ExecuteNonQueryAsync(ct);
    }
}

public sealed class CreateOrderHandler(NpgsqlDataSource dataSource, OutboxWriter outbox)
{
    public async Task<string> HandleAsync(CreateOrderCommand cmd, CancellationToken ct)
    {
        await using var conn = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await conn.BeginTransactionAsync(ct);

        var orderId = Guid.NewGuid().ToString("N");

        // Main business write
        await using (var insert = new NpgsqlCommand(
            """
            INSERT INTO orders (id, user_id, total_cents, status)
            VALUES (@id, @userId, @totalCents, 'created')
            """, conn, tx))
        {
            insert.Parameters.AddWithValue("id", orderId);
            insert.Parameters.AddWithValue("userId", cmd.UserId);
            insert.Parameters.AddWithValue("totalCents", cmd.TotalCents);
            await insert.ExecuteNonQueryAsync(ct);
        }

        // Outbox write in the same transaction -- atomic!
        await outbox.AppendAsync(
            conn, tx,
            aggregateType: "order",
            aggregateId: orderId,
            eventType: "OrderCreated",
            payload: new
            {
                orderId,
                userId = cmd.UserId,
                totalCents = cmd.TotalCents,
            },
            headers: new Dictionary<string, string>
            {
                ["correlationId"] = cmd.CorrelationId,
                ["occurredAt"] = DateTimeOffset.UtcNow.ToString("o"),
            },
            ct);

        // Either both rows land or neither: disposing without commit rolls back
        await tx.CommitAsync(ct);
        return orderId;
    }
}

import json
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any

import asyncpg

@dataclass(frozen=True, slots=True)
class CreateOrderCommand:
    user_id: int
    total_cents: int
    correlation_id: str

class OutboxWriter:
    """Call within the same transaction as your main INSERT/UPDATE."""

    async def append(
        self,
        conn: asyncpg.Connection,
        aggregate_type: str,
        aggregate_id: str,
        event_type: str,
        payload: dict[str, Any],
        headers: dict[str, str] | None = None,
    ) -> None:
        await conn.execute(
            """
            INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload, headers)
            VALUES ($1, $2, $3, $4::jsonb, $5::jsonb)
            """,
            aggregate_type,
            aggregate_id,
            event_type,
            json.dumps(payload),
            json.dumps(headers or {}),
        )

class CreateOrderHandler:
    def __init__(self, pool: asyncpg.Pool, outbox: OutboxWriter) -> None:
        self._pool = pool
        self._outbox = outbox

    async def handle(self, cmd: CreateOrderCommand) -> str:
        order_id = uuid.uuid4().hex

        async with self._pool.acquire() as conn:
            # Context manager commits on success, rolls back on any exception
            async with conn.transaction():
                # Main business write
                await conn.execute(
                    """
                    INSERT INTO orders (id, user_id, total_cents, status)
                    VALUES ($1, $2, $3, 'created')
                    """,
                    order_id,
                    cmd.user_id,
                    cmd.total_cents,
                )

                # Outbox write in the same transaction -- atomic!
                await self._outbox.append(
                    conn,
                    aggregate_type="order",
                    aggregate_id=order_id,
                    event_type="OrderCreated",
                    payload={
                        "orderId": order_id,
                        "userId": cmd.user_id,
                        "totalCents": cmd.total_cents,
                    },
                    headers={
                        "correlationId": cmd.correlation_id,
                        "occurredAt": datetime.now(timezone.utc).isoformat(),
                    },
                )

        return order_id

## Publisher worker: построчная публикация

<?php

declare(strict_types=1);

namespace App\Outbox;

use Doctrine\DBAL\Connection;
use Psr\Log\LoggerInterface;
use RdKafka\Producer;

/**
 * Outbox publisher. Polls pending rows, publishes to Kafka, marks sent.
 * Runs as a long-lived worker (1 or more replicas, coordinate via SKIP LOCKED).
 */
final class OutboxPublisher
{
    private const BATCH_SIZE = 100;
    private const POLL_INTERVAL_MS = 200;

    public function __construct(
        private readonly Connection $conn,
        private readonly Producer $kafka,
        private readonly LoggerInterface $log,
    ) {}

    public function run(): never
    {
        while (true) {
            $processed = $this->tick();
            if ($processed === 0) {
                usleep(self::POLL_INTERVAL_MS * 1000);
            }
        }
    }

    private function tick(): int
    {
        // SKIP LOCKED allows N parallel workers without duplication
        $sql = <<<'SQL'
            SELECT id, aggregate_type, aggregate_id, event_type, payload, headers, attempts
            FROM outbox
            WHERE sent_at IS NULL
            ORDER BY id ASC
            LIMIT :limit
            FOR UPDATE SKIP LOCKED
        SQL;

        $this->conn->beginTransaction();
        try {
            $rows = $this->conn->fetchAllAssociative($sql, ['limit' => self::BATCH_SIZE]);

            foreach ($rows as $row) {
                $this->publish($row);
            }

            $this->conn->commit();
            return count($rows);
        } catch (\Throwable $e) {
            $this->conn->rollBack();
            $this->log->error('outbox tick failed', ['error' => $e->getMessage()]);
            return 0;
        }
    }

    /**
     * @param array<string, mixed> $row
     */
    private function publish(array $row): void
    {
        $topic = $this->kafka->newTopic($this->topicFor($row['aggregate_type']));

        // Partition key = aggregate_id to preserve ordering
        $topic->producev(
            partition: RD_KAFKA_PARTITION_UA,
            msgflags: 0,
            payload: $row['payload'],
            key: $row['aggregate_id'],
            headers: [
                'event_type' => $row['event_type'],
                'event_id' => (string) $row['id'],
            ],
        );
        $this->kafka->flush(5000);

        $this->conn->update(
            'outbox',
            ['sent_at' => (new \DateTimeImmutable())->format('Y-m-d H:i:s')],
            ['id' => $row['id']],
        );
    }

    private function topicFor(string $aggregateType): string
    {
        return match ($aggregateType) {
            'order' => 'orders.events',
            'user'  => 'users.events',
            default => throw new \InvalidArgumentException("unknown aggregate: $aggregateType"),
        };
    }
}

using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using Npgsql;

namespace App.Outbox;

/// Outbox publisher. Polls pending rows, publishes to Kafka, marks sent.
/// Runs as a long-lived worker (1 or more replicas, coordinate via SKIP LOCKED).
public sealed class OutboxPublisher(
    NpgsqlDataSource dataSource,
    IProducer<string, string> kafka,
    ILogger<OutboxPublisher> log)
{
    private const int BatchSize = 100;
    private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(200);

    public async Task RunAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            var processed = await TickAsync(ct);
            if (processed == 0)
            {
                await Task.Delay(PollInterval, ct);
            }
        }
    }

    private async Task<int> TickAsync(CancellationToken ct)
    {
        await using var conn = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await conn.BeginTransactionAsync(ct);

        try
        {
            // SKIP LOCKED allows N parallel workers without duplication
            var rows = new List<OutboxRow>(BatchSize);
            await using (var select = new NpgsqlCommand(
                """
                SELECT id, aggregate_type, aggregate_id, event_type, payload
                FROM outbox
                WHERE sent_at IS NULL
                ORDER BY id ASC
                LIMIT @limit
                FOR UPDATE SKIP LOCKED
                """, conn, tx))
            {
                select.Parameters.AddWithValue("limit", BatchSize);
                await using var reader = await select.ExecuteReaderAsync(ct);
                while (await reader.ReadAsync(ct))
                {
                    rows.Add(new OutboxRow(
                        reader.GetInt64(0), reader.GetString(1),
                        reader.GetString(2), reader.GetString(3), reader.GetString(4)));
                }
            }

            foreach (var row in rows)
            {
                await PublishAsync(conn, tx, row, ct);
            }

            await tx.CommitAsync(ct);
            return rows.Count;
        }
        catch (Exception e)
        {
            await tx.RollbackAsync(ct);
            log.LogError(e, "outbox tick failed");
            return 0;
        }
    }

    private async Task PublishAsync(
        NpgsqlConnection conn, NpgsqlTransaction tx, OutboxRow row, CancellationToken ct)
    {
        var headers = new Headers
        {
            { "event_type", System.Text.Encoding.UTF8.GetBytes(row.EventType) },
            { "event_id", System.Text.Encoding.UTF8.GetBytes(row.Id.ToString()) },
        };

        // Partition key = aggregate_id to preserve ordering; awaiting the delivery
        // report means we only mark sent after the broker acked
        await kafka.ProduceAsync(
            TopicFor(row.AggregateType),
            new Message<string, string>
            {
                Key = row.AggregateId,
                Value = row.Payload,
                Headers = headers,
            },
            ct);

        await using var update = new NpgsqlCommand(
            "UPDATE outbox SET sent_at = NOW() WHERE id = @id", conn, tx);
        update.Parameters.AddWithValue("id", row.Id);
        await update.ExecuteNonQueryAsync(ct);
    }

    private static string TopicFor(string aggregateType) => aggregateType switch
    {
        "order" => "orders.events",
        "user" => "users.events",
        _ => throw new ArgumentException($"unknown aggregate: {aggregateType}"),
    };

    private sealed record OutboxRow(
        long Id, string AggregateType, string AggregateId, string EventType, string Payload);
}

import asyncio
import logging
from dataclasses import dataclass

import asyncpg
from aiokafka import AIOKafkaProducer

log = logging.getLogger(__name__)

BATCH_SIZE = 100
POLL_INTERVAL_S = 0.2

TOPIC_BY_AGGREGATE = {
    "order": "orders.events",
    "user": "users.events",
}

@dataclass(frozen=True, slots=True)
class OutboxRow:
    id: int
    aggregate_type: str
    aggregate_id: str
    event_type: str
    payload: str

class OutboxPublisher:
    """Polls pending rows, publishes to Kafka, marks sent.

    Runs as a long-lived worker (1+ replicas, coordinated via SKIP LOCKED).
    """

    def __init__(self, pool: asyncpg.Pool, producer: AIOKafkaProducer) -> None:
        self._pool = pool
        self._producer = producer

    async def run(self) -> None:
        while True:
            processed = await self._tick()
            if processed == 0:
                await asyncio.sleep(POLL_INTERVAL_S)

    async def _tick(self) -> int:
        try:
            async with self._pool.acquire() as conn:
                async with conn.transaction():
                    # SKIP LOCKED allows N parallel workers without duplication
                    records = await conn.fetch(
                        """
                        SELECT id, aggregate_type, aggregate_id, event_type, payload::text
                        FROM outbox
                        WHERE sent_at IS NULL
                        ORDER BY id ASC
                        LIMIT $1
                        FOR UPDATE SKIP LOCKED
                        """,
                        BATCH_SIZE,
                    )

                    for record in records:
                        await self._publish(conn, OutboxRow(*record))

                    return len(records)
        except Exception:
            # Transaction already rolled back; next tick re-reads the same rows
            log.exception("outbox tick failed")
            return 0

    async def _publish(self, conn: asyncpg.Connection, row: OutboxRow) -> None:
        topic = TOPIC_BY_AGGREGATE.get(row.aggregate_type)
        if topic is None:
            raise ValueError(f"unknown aggregate: {row.aggregate_type}")

        # Partition key = aggregate_id to preserve ordering per aggregate;
        # send_and_wait blocks until the broker acks, so sent_at is only set after
        await self._producer.send_and_wait(
            topic,
            key=row.aggregate_id.encode(),
            value=row.payload.encode(),
            headers=[
                ("event_type", row.event_type.encode()),
                ("event_id", str(row.id).encode()),
            ],
        )

        await conn.execute("UPDATE outbox SET sent_at = NOW() WHERE id = $1", row.id)

## Publisher worker: публикация батчами

package outbox

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	"github.com/segmentio/kafka-go"
)

const batchSize = 100

type Event struct {
	ID            int64
	AggregateType string
	AggregateID   string
	EventType     string
	Payload       json.RawMessage
	Headers       json.RawMessage
}

type Publisher struct {
	db      *sql.DB
	writer  *kafka.Writer
	log     *slog.Logger
	topicFn func(aggregateType string) string
}

// Run polls the outbox and publishes pending events.
// Graceful shutdown via ctx.
func (p *Publisher) Run(ctx context.Context) error {
	ticker := time.NewTicker(200 * time.Millisecond)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
			if err := p.tick(ctx); err != nil {
				p.log.Error("tick failed", "err", err)
			}
		}
	}
}

func (p *Publisher) tick(ctx context.Context) error {
	tx, err := p.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return err
	}
	defer tx.Rollback()

	rows, err := tx.QueryContext(ctx, `
		SELECT id, aggregate_type, aggregate_id, event_type, payload, headers
		FROM outbox
		WHERE sent_at IS NULL
		ORDER BY id ASC
		LIMIT $1
		FOR UPDATE SKIP LOCKED`, batchSize)
	if err != nil {
		return fmt.Errorf("select: %w", err)
	}

	var events []Event
	for rows.Next() {
		var e Event
		if err := rows.Scan(&e.ID, &e.AggregateType, &e.AggregateID, &e.EventType, &e.Payload, &e.Headers); err != nil {
			rows.Close()
			return err
		}
		events = append(events, e)
	}
	rows.Close()
	if len(events) == 0 {
		return nil
	}

	// Batch publish; partition key = aggregate_id for FIFO per aggregate
	msgs := make([]kafka.Message, 0, len(events))
	for _, e := range events {
		msgs = append(msgs, kafka.Message{
			Topic: p.topicFn(e.AggregateType),
			Key:   []byte(e.AggregateID),
			Value: e.Payload,
			Headers: []kafka.Header{
				{Key: "event_type", Value: []byte(e.EventType)},
				{Key: "event_id", Value: []byte(fmt.Sprintf("%d", e.ID))},
			},
		})
	}
	if err := p.writer.WriteMessages(ctx, msgs...); err != nil {
		return fmt.Errorf("kafka write: %w", err)
	}

	// Mark sent
	ids := make([]int64, len(events))
	for i, e := range events {
		ids[i] = e.ID
	}
	if _, err := tx.ExecContext(ctx,
		`UPDATE outbox SET sent_at = NOW() WHERE id = ANY($1)`, ids); err != nil {
		return fmt.Errorf("update: %w", err)
	}

	return tx.Commit()
}

using System.Text;
using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using Npgsql;

namespace App.Outbox;

public sealed record OutboxEvent(
    long Id, string AggregateType, string AggregateId, string EventType, string Payload);

/// Batched publisher: reads a batch, produces all messages, then marks the
/// whole batch sent in a single UPDATE.
public sealed class BatchOutboxPublisher(
    NpgsqlDataSource dataSource,
    IProducer<string, string> kafka,
    Func<string, string> topicFor,
    ILogger<BatchOutboxPublisher> log)
{
    private const int BatchSize = 100;

    /// Polls the outbox and publishes pending events.
    /// Graceful shutdown via the cancellation token.
    public async Task RunAsync(CancellationToken ct)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(200));
        while (await timer.WaitForNextTickAsync(ct))
        {
            try
            {
                await TickAsync(ct);
            }
            catch (Exception e)
            {
                log.LogError(e, "tick failed");
            }
        }
    }

    private async Task TickAsync(CancellationToken ct)
    {
        await using var conn = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await conn.BeginTransactionAsync(
            System.Data.IsolationLevel.ReadCommitted, ct);

        var events = new List<OutboxEvent>(BatchSize);
        await using (var select = new NpgsqlCommand(
            """
            SELECT id, aggregate_type, aggregate_id, event_type, payload::text
            FROM outbox
            WHERE sent_at IS NULL
            ORDER BY id ASC
            LIMIT @limit
            FOR UPDATE SKIP LOCKED
            """, conn, tx))
        {
            select.Parameters.AddWithValue("limit", BatchSize);
            await using var reader = await select.ExecuteReaderAsync(ct);
            while (await reader.ReadAsync(ct))
            {
                events.Add(new OutboxEvent(
                    reader.GetInt64(0), reader.GetString(1),
                    reader.GetString(2), reader.GetString(3), reader.GetString(4)));
            }
        }

        if (events.Count == 0)
        {
            return;
        }

        // Batch publish; partition key = aggregate_id for FIFO per aggregate
        foreach (var e in events)
        {
            await kafka.ProduceAsync(
                topicFor(e.AggregateType),
                new Message<string, string>
                {
                    Key = e.AggregateId,
                    Value = e.Payload,
                    Headers = new Headers
                    {
                        { "event_type", Encoding.UTF8.GetBytes(e.EventType) },
                        { "event_id", Encoding.UTF8.GetBytes(e.Id.ToString()) },
                    },
                },
                ct);
        }

        // Mark sent
        await using (var update = new NpgsqlCommand(
            "UPDATE outbox SET sent_at = NOW() WHERE id = ANY(@ids)", conn, tx))
        {
            update.Parameters.AddWithValue("ids", events.Select(e => e.Id).ToArray());
            await update.ExecuteNonQueryAsync(ct);
        }

        await tx.CommitAsync(ct);
    }
}

import asyncio
import logging
from collections.abc import Callable
from dataclasses import dataclass

import asyncpg
from aiokafka import AIOKafkaProducer

log = logging.getLogger(__name__)

BATCH_SIZE = 100
TICK_INTERVAL_S = 0.2

@dataclass(frozen=True, slots=True)
class OutboxEvent:
    id: int
    aggregate_type: str
    aggregate_id: str
    event_type: str
    payload: str

class BatchOutboxPublisher:
    """Reads a batch, produces all messages, marks the batch sent in one UPDATE."""

    def __init__(
        self,
        pool: asyncpg.Pool,
        producer: AIOKafkaProducer,
        topic_for: Callable[[str], str],
    ) -> None:
        self._pool = pool
        self._producer = producer
        self._topic_for = topic_for

    async def run(self) -> None:
        """Poll the outbox until cancelled (asyncio.CancelledError = graceful stop)."""
        while True:
            try:
                await self._tick()
            except asyncio.CancelledError:
                raise
            except Exception:
                log.exception("tick failed")
            await asyncio.sleep(TICK_INTERVAL_S)

    async def _tick(self) -> None:
        async with self._pool.acquire() as conn:
            async with conn.transaction(isolation="read_committed"):
                records = await conn.fetch(
                    """
                    SELECT id, aggregate_type, aggregate_id, event_type, payload::text
                    FROM outbox
                    WHERE sent_at IS NULL
                    ORDER BY id ASC
                    LIMIT $1
                    FOR UPDATE SKIP LOCKED
                    """,
                    BATCH_SIZE,
                )
                if not records:
                    return

                events = [OutboxEvent(*r) for r in records]

                # Batch publish; partition key = aggregate_id for FIFO per aggregate.
                # Gathering the futures keeps per-key ordering: the producer serialises
                # messages sharing a partition in submission order.
                futures = [
                    await self._producer.send(
                        self._topic_for(e.aggregate_type),
                        key=e.aggregate_id.encode(),
                        value=e.payload.encode(),
                        headers=[
                            ("event_type", e.event_type.encode()),
                            ("event_id", str(e.id).encode()),
                        ],
                    )
                    for e in events
                ]
                await asyncio.gather(*futures)

                # Mark sent
                await conn.execute(
                    "UPDATE outbox SET sent_at = NOW() WHERE id = ANY($1::bigint[])",
                    [e.id for e in events],
                )

**Ключевой момент:** `FOR UPDATE SKIP LOCKED` позволяет запустить несколько реплик publisher -- каждая возьмёт свой батч, не будет блокировок и дублей. Для гарантии FIFO per aggregate -- либо один worker, либо шардирование по `hash(aggregate_id) % N`.

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

Worker может опубликовать событие, но упасть до UPDATE sent_at = событие придёт в Kafka дважды. Consumer должен уметь это пережить.

CREATE TABLE inbox (
    event_id   TEXT PRIMARY KEY,    -- из headers (event_id publisher'а)
    topic      TEXT NOT NULL,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

<?php

declare(strict_types=1);

final class InventoryConsumer
{
    public function __construct(
        private readonly Connection $conn,
        private readonly InventoryService $service,
    ) {}

    public function handle(KafkaMessage $msg): void
    {
        $eventId = $msg->headers['event_id'];

        $this->conn->beginTransaction();
        try {
            // Atomic: insert inbox marker + business logic in one TX
            $inserted = $this->conn->executeStatement(
                'INSERT INTO inbox (event_id, topic) VALUES (:id, :t) ON CONFLICT DO NOTHING',
                ['id' => $eventId, 't' => $msg->topic],
            );

            if ($inserted === 0) {
                // Already processed -- idempotent skip
                $this->conn->rollBack();
                return;
            }

            $payload = json_decode($msg->value, true, flags: JSON_THROW_ON_ERROR);
            $this->service->reserve($payload['orderId']);

            $this->conn->commit();
        } catch (\Throwable $e) {
            $this->conn->rollBack();
            throw $e;
        }
    }
}

using System.Text;
using System.Text.Json;
using Confluent.Kafka;
using Npgsql;

namespace App.Inbox;

public sealed class InventoryConsumer(
    NpgsqlDataSource dataSource,
    IInventoryService service)
{
    public async Task HandleAsync(ConsumeResult<string, string> msg, CancellationToken ct)
    {
        var eventId = ReadHeader(msg.Message.Headers, "event_id")
            ?? throw new InvalidOperationException("message has no event_id header");

        await using var conn = await dataSource.OpenConnectionAsync(ct);
        await using var tx = await conn.BeginTransactionAsync(ct);

        // Atomic: insert inbox marker + business logic in one TX
        await using var insert = new NpgsqlCommand(
            "INSERT INTO inbox (event_id, topic) VALUES (@id, @topic) ON CONFLICT DO NOTHING",
            conn, tx);
        insert.Parameters.AddWithValue("id", eventId);
        insert.Parameters.AddWithValue("topic", msg.Topic);

        if (await insert.ExecuteNonQueryAsync(ct) == 0)
        {
            // Already processed -- idempotent skip
            await tx.RollbackAsync(ct);
            return;
        }

        var payload = JsonSerializer.Deserialize<OrderCreatedPayload>(msg.Message.Value)
            ?? throw new InvalidOperationException("empty payload");
        await service.ReserveAsync(payload.OrderId, conn, tx, ct);

        await tx.CommitAsync(ct);
    }

    private static string? ReadHeader(Headers? headers, string key) =>
        headers is not null && headers.TryGetLastBytes(key, out var raw)
            ? Encoding.UTF8.GetString(raw)
            : null;
}

import json

import asyncpg
from aiokafka import ConsumerRecord

class InventoryConsumer:
    def __init__(self, pool: asyncpg.Pool, service: InventoryService) -> None:
        self._pool = pool
        self._service = service

    async def handle(self, msg: ConsumerRecord) -> None:
        event_id = next(
            (v.decode() for k, v in (msg.headers or ()) if k == "event_id"), None
        )
        if event_id is None:
            raise ValueError("message has no event_id header")

        async with self._pool.acquire() as conn:
            async with conn.transaction():
                # Atomic: insert inbox marker + business logic in one TX.
                # execute() returns a tag like "INSERT 0 1"; "INSERT 0 0" == conflict.
                tag = await conn.execute(
                    "INSERT INTO inbox (event_id, topic) VALUES ($1, $2) ON CONFLICT DO NOTHING",
                    event_id,
                    msg.topic,
                )
                if tag.endswith(" 0"):
                    # Already processed -- idempotent skip
                    return

                payload = json.loads(msg.value)
                await self._service.reserve(conn, payload["orderId"])

## LISTEN/NOTIFY вместо polling (PostgreSQL)

Если важна латентность -- можно разбудить worker триггером:

CREATE FUNCTION outbox_notify() RETURNS TRIGGER AS $$
BEGIN
    PERFORM pg_notify('outbox_new', NEW.id::text);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER outbox_notify_trigger
AFTER INSERT ON outbox
FOR EACH ROW EXECUTE FUNCTION outbox_notify();

Worker делает LISTEN outbox_new, при каждом сигнале -- tick. Fallback на polling раз в 5 секунд на случай пропущенного notify (они best-effort).

CDC как альтернатива: Debezium

Вместо своего publisher'а можно читать WAL (Write-Ahead Log) PostgreSQL через Debezium. Он стримит изменения любой таблицы в Kafka автоматически.

Аспект	Outbox + Worker	Debezium CDC
Код	Ваш + миграция	Конфиг Debezium
Гибкость events	Полный контроль payload	Маппинг row -> event
Операционная нагрузка	Worker мониторить	Kafka Connect кластер
Latency	100-500ms	10-100ms
Схема событий	Бизнес-осмысленная	Отражает БД

Outbox лучше, если нужны бизнес-события с осмысленной схемой. CDC -- для репликации БД в поисковые/аналитические системы без написания кода.

Cleanup outbox

Outbox растёт линейно. Варианты уборки:

-- Keep 7 days history for audit/replay
DELETE FROM outbox
WHERE sent_at < NOW() - INTERVAL '7 days';

-- Or partition by day and DROP old partitions (faster)

Партицирование по дню -- быстрее, чем DELETE миллионов строк.

Мониторинг

Метрика	Назначение	Алерт
`outbox_pending` (gauge)	Количество неопубликованных	> 1000
`outbox_lag_seconds`	max(now - created_at) для pending	> 60
`outbox_published_total`	counter	-
`outbox_publish_errors_total`	counter	rate > 0.01/s
`outbox_worker_tick_duration`	histogram	p95 > 1s

-- Lag query
SELECT EXTRACT(EPOCH FROM (NOW() - MIN(created_at))) AS lag_seconds
FROM outbox
WHERE sent_at IS NULL;

Pitfalls

Забыли COMMIT outbox вместе с бизнес-данными -- обычный dual-write. Обязательно одна транзакция.
FIFO нарушается при N workers без шардинга -- order с id=1 опубликован worker B, id=2 -- worker A раньше. Решение: shard by hash(aggregate_id) % N.
Consumer не идемпотентен -- at-least-once превратится в обработано N раз. Inbox pattern обязателен.
Payload слишком большой (>1MB) -- Kafka откажется. Храните blob отдельно, в outbox -- ссылку.
outbox таблица распухла -- ленивые DELETE / партиции обязательны.
Worker публикует, но не UPDATE (упал) -- дубли. Консьюмеры обязаны быть идемпотентны.
Schema evolution: изменили структуру payload -- consumers старой версии сломаются. Используйте schema registry (Avro/Protobuf).

Выводы

Outbox решает dual-write проблему через атомарную запись события в ту же БД
Гарантия: at-least-once. Consumers обязаны быть идемпотентны (inbox pattern)
FOR UPDATE SKIP LOCKED позволяет масштабировать publisher'ы без дублей
FIFO per aggregate достигается через partition key = aggregate_id
CDC (Debezium) -- альтернатива без кода, но с менее осмысленной схемой событий
Мониторинг outbox_lag и outbox_pending критичен
Очистка таблицы через партицирование, не DELETE