Here’s a piece of code I’ve written, in some form, at three different companies:
func (s *ProductService) CreateProduct(ctx context.Context, cmd CreateProduct) error {
product := entities.NewProduct(cmd.Name, cmd.Price, seller)
if err := s.repo.Create(ctx, product); err != nil {
return err
}
// Tell the rest of the world
return s.kafka.Publish(ctx, "product.created", toEvent(product))
}
Enter fullscreen mode Exit fullscreen mode
Save to Postgres, then publish to Kafka. It works in every demo, every test, and roughly 99.9% of the time in production. The remaining 0.1% is where the fun is.
The database and the broker are two separate systems. There is no transaction spanning both. So:
- The process crashes after the DB commit but before the Kafka publish. The product exists, but downstream services (search index, pricing, notifications) never hear about it. Silent data drift.
- You flip the order — publish first, then save — and now a failed DB insert produces a ghost event: consumers react to a product that was never created.
- Kafka is briefly unavailable. Do you retry? For how long? Do you fail the HTTP request even though the product is already committed?
You can’t “just be careful” your way out of this. Wrapping both calls in a retry loop doesn’t create atomicity, it just shrinks the window. I know because I shipped variations of that retry loop for years before admitting the design was wrong.
The fix is old and boring, which is exactly what you want: the transactional outbox. Don’t publish the event. Store it — in the same database, in the same transaction as the state change. A separate process reads the stored events and does the actual publishing. One transaction, one system of record, no dual write.
I recently rebuilt this end-to-end in my Go DDD template repo, and I want to walk through the actual implementation, because most outbox articles stop at the diagram.
Events belong to the aggregate
The first design decision has nothing to do with infrastructure. Who creates the event?
In a lot of codebases the service layer builds the event, right next to the publish call. That’s backwards. “A product was created” is a domain fact, and the aggregate is the thing that knows when its own state changed. So the aggregate records events as part of the state change:
func NewProduct(name string, price Money, seller ValidatedSeller) *Product {
product := &Product{
Id: uuid.Must(uuid.NewV7()),
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
Name: name,
Price: price,
SellerId: seller.Id,
}
product.recordEvent(events.NewProductCreated(
product.Id, name, price.Cents(), string(price.Currency()), seller.Id))
return product
}
func (p *Product) recordEvent(event events.DomainEvent) {
p.domainEvents = append(p.domainEvents, event)
}
// PullEvents returns the recorded domain events and clears them.
func (p *Product) PullEvents() []events.DomainEvent {
pulled := p.domainEvents
p.domainEvents = nil
return pulled
}
Enter fullscreen mode Exit fullscreen mode
The events themselves are dumb structs. Past-tense names, immutable, no behavior:
type DomainEvent interface {
EventId() uuid.UUID
EventName() string
OccurredAt() time.Time
AggregateId() uuid.UUID
}
type ProductCreated struct {
BaseEvent
Name string
PriceCents int64
Currency string
SellerId uuid.UUID
}
func (e ProductCreated) EventName() string { return "product.created" }
Enter fullscreen mode Exit fullscreen mode
Two details worth calling out. The event Id is a UUIDv7 — time-ordered, so it sorts nicely and doubles as a deduplication key for consumers later. And PullEvents clears the slice, so the repository pulls exactly once per save and a retried save can’t double-insert the same events.
One transaction or it didn’t happen
The outbox table is deliberately simple:
CREATE TABLE outbox_events (
id UUID PRIMARY KEY,
aggregate_id UUID NOT NULL,
event_name TEXT NOT NULL,
payload JSONB NOT NULL,
occurred_at TIMESTAMP WITH TIME ZONE NOT NULL,
published_at TIMESTAMP WITH TIME ZONE
);
CREATE INDEX idx_outbox_events_unpublished
ON outbox_events(occurred_at) WHERE published_at IS NULL;
Enter fullscreen mode Exit fullscreen mode
Note the partial index. The relay only ever asks one question — “give me unpublished events, oldest first” — and the outbox table grows forever (or until you archive it). A partial index on WHERE published_at IS NULL stays tiny no matter how many millions of published rows accumulate, because rows drop out of the index the moment they’re marked published. A full index on occurred_at would keep indexing dead rows for no reason.
The repository is where the pattern actually pays off. Aggregate insert and outbox insert share one pgx transaction:
func (repo *SqlcProductRepository) Create(ctx context.Context, product *entities.ValidatedProduct) (*entities.Product, error) {
tx, err := repo.pool.Begin(ctx)
if err != nil {
return nil, err
}
defer func() { _ = tx.Rollback(ctx) }()
qtx := repo.queries.WithTx(tx)
if _, err := qtx.CreateProduct(ctx, db.CreateProductParams{
ID: product.Id,
Name: product.Name,
PriceCents: product.Price.Cents(),
// ...
}); err != nil {
return nil, err
}
if err := insertOutboxEvents(ctx, qtx, product.PullEvents()); err != nil {
return nil, err
}
row, err := qtx.GetProductById(ctx, product.Id)
if err != nil {
return nil, err
}
created, err := productFromRow(row.ID, row.Name, /* ... */)
if err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, err
}
return created, nil
}
Enter fullscreen mode Exit fullscreen mode
Either the product row and its events commit, or neither does. Crash anywhere in between and the transaction rolls back — no orphaned product, no ghost event. Even the read-after-write happens inside the transaction, so a transient failure can’t surface after the commit already succeeded.
The outbox insert itself just serializes each event to JSONB:
func insertOutboxEvents(ctx context.Context, queries *db.Queries, domainEvents []events.DomainEvent) error {
for _, event := range domainEvents {
payload, err := json.Marshal(event)
if err != nil {
return err
}
if err := queries.InsertOutboxEvent(ctx, db.InsertOutboxEventParams{
ID: event.EventId(),
AggregateID: event.AggregateId(),
EventName: event.EventName(),
Payload: payload,
OccurredAt: timestamptzFromTime(event.OccurredAt()),
}); err != nil {
return err
}
}
return nil
}
Enter fullscreen mode Exit fullscreen mode
Notice the service layer knows nothing about any of this. It calls repo.Create and the events ride along. You can’t forget to publish, because there is no publish step to forget.
The relay: dumb on purpose
Something still has to get the events out of Postgres and into the broker. That’s the relay — a loop that polls unpublished rows and hands them to a Publisher:
type Publisher interface {
Publish(ctx context.Context, eventName string, payload []byte) error
}
func (r *Relay) relayBatch(ctx context.Context) error {
events, err := r.queries.GetUnpublishedOutboxEvents(ctx, r.batchSize)
if err != nil {
return err
}
for _, event := range events {
if err := r.publisher.Publish(ctx, event.EventName, event.Payload); err != nil {
// Stop the batch; unpublished events are retried next tick.
return err
}
if err := r.queries.MarkOutboxEventPublished(ctx, event.ID); err != nil {
return err
}
}
return nil
}
Enter fullscreen mode Exit fullscreen mode
In the template the publisher just logs via slog; in a real deployment you swap in Kafka, NATS, SQS, whatever. The interesting property is the failure mode: publish succeeds, then MarkOutboxEventPublished fails — crash, network blip, deploy. Next tick, the row is still unpublished, so it gets published again.
This is not a bug. It’s the contract: the outbox gives you at-least-once delivery, never exactly-once. Which means every consumer must be idempotent. Handle product.created twice and end up in the same state as handling it once. The event Id is your dedup key — consumers keep a small table of processed event Ids, or use a natural idempotent operation (upsert into the search index instead of insert).
If that sounds like a burden: you needed idempotent consumers anyway. Kafka itself will redeliver on consumer group rebalances. At-least-once is the honest default of distributed messaging; the outbox just makes it explicit instead of pretending otherwise.
The caveats nobody puts in the diagram
Ordering. The poll query orders by occurred_at, and UUIDv7 event Ids are time-ordered too, so events for one aggregate come out in the order they were recorded — as long as there’s a single relay. That’s per-relay ordering, not global ordering. If you publish to a partitioned topic, partition by aggregate_id so consumers see each aggregate’s events in order. Cross-aggregate ordering is a promise you should never make.
Scaling the relay. Run one relay instance and life is simple. Run two naive ones and both grab the same batch and double-publish everything (idempotent consumers save you, but it’s wasteful). The standard fix is FOR UPDATE SKIP LOCKED in the poll query:
SELECT id, event_name, payload
FROM outbox_events
WHERE published_at IS NULL
ORDER BY occurred_at
LIMIT $1
FOR UPDATE SKIP LOCKED;
Enter fullscreen mode Exit fullscreen mode
Each relay locks the rows it’s working on; competing relays skip locked rows and grab the next ones. You trade a bit of ordering (batches interleave across relays) for horizontal scale. My template ships the single-instance version because that’s the right default — I’d rather people add SKIP LOCKED when they measure a need than start with complexity they don’t have.
Polling vs. CDC. Polling every second or so is fine for a huge range of workloads and requires zero extra infrastructure. If you need lower latency or your outbox writes are heavy, log-tailing CDC (Debezium reading the WAL) publishes without polling. It’s better and it’s a lot more moving parts. Start with polling.
Cleanup. Published rows pile up. A nightly DELETE FROM outbox_events WHERE published_at < now() - interval '30 days' keeps the table sane, and thanks to the partial index the relay’s query doesn’t care either way.
When to skip all of this
Honest answer: often.
If you’re a single service and the “event handler” lives in the same process — say, creating a product should also warm a cache — you don’t need an outbox. Call the function. In-process, in-transaction, done.
The outbox earns its keep exactly when a state change in your database must reliably reach another system: a broker, a search index, a webhook, another service. No cross-system consumer, no outbox. I’ve seen teams cargo-cult the pattern into a monolith with zero integration events and end up maintaining a relay that publishes to nobody.
Also: if you can tolerate losing the occasional event (analytics pings, best-effort notifications), fire-and-forget with a retry is genuinely fine. The outbox is for events where “we lost it” means data drift or a support ticket.
But the moment someone says “when X happens here, Y must happen over there” — reach for it. The pattern is maybe 200 lines including the migration, and it turns a distributed-systems problem into a table and a for loop.
The full implementation — aggregate, repository, relay, migration, and testcontainers-based integration tests — lives in my Go DDD template at https://github.com/sklinkert/go-ddd, alongside the rest of the patterns I keep rebuilding at every job: validated entities, value objects like Money, and race-safe idempotency keys. I also extracted a broker-agnostic version of the outbox into a standalone library, https://github.com/sklinkert/go-outbox, if you want the pattern without the template.
답글 남기기