分散マイクロサービスにおける二重書き込み（Dual-Write）の罠
一般的なチェックアウト（決済）処理の流れを想像してみてください。APIがPostgreSQL内の注文データを更新し、KafkaやRabbitMQへメッセージを送信します。一見すると非常にシンプルに思えます。後続の配送サービスや請求書発行サービスは、そのイベントを受け取って注文を処理します。
多くのエンジニアは、最初に以下のようなコードを書くのではないでしょうか。
// アンチパターン：安全ではない二重書き込み（Dual-Write）
_, err := db.ExecContext(ctx, "UPDATE orders SET status = 'PAID' WHERE id = $1", orderID)
if err != nil {
return err
}
// まさにこのタイミングでネットワークが切断されたり、Podがクラッシュしたらどうなるでしょうか？
err = messageBroker.Publish(ctx, "order.paid", eventPayload)
if err != nil {
// データベース上は注文がPAIDになりますが、後続サービスはその変更を検知できません。
return err
}
これが「二重書き込み（Dual-Write）の罠」です。リレーショナルデータベースとメッセージブローカーの間を繋ぐ分散トランザクションは存在しません。データベースへのコミットとブローカーへの送信の間でアプリケーションがクラッシュした場合、イベントは永遠に失われてしまいます。顧客は支払いを済ませたのに、倉庫では一向に梱包が行われないという事態に陥ります。
実行順序を逆にしても問題は解決しません。先にイベントを送信して後からコミットを行うと、「ゴーストイベント」が発生するリスクがあります。もしデータベースのトランザクションがロールバックされた場合、後続サービスは未払いの注文を処理してしまうことになります。
理論上は2相コミット（2PC: Two-Phase Commit）によって分散トランザクションを調整できます。しかし実際には、2PCは50〜100msもの深刻なレイテンシのスパイクを引き起こし、単一障害点（SPOF）を生み出すうえ、Kafkaなどのモダンなメッセージブローカーではネイティブにサポートされていません。求められているのは、高いスループットを維持しながら確実にイベントを配信する仕組みです。
Transactional Outboxパターンの仕組み
Transactional Outboxパターンは、分散トランザクションを完全に回避します。その代わりに、ローカルデータベースが誇る実績十分なACID特性の保証を活用します。ドメインデータの変更とそれに対応する送信イベントを、単一のアトミックなデータベーストランザクション内で同時に処理します。
3ステップのワークフロー
- 単一のアトミックな書き込み（Single Atomic Write）： アプリケーションは
ordersテーブルを更新する際、同時にoutbox_eventsテーブルへレコードを書き込みます。データベーストランザクションがコミットされれば両方のレコードがディスクに永続化され、失敗した場合は両方ともクリーンにロールバックされます。ゴーストイベントの発生はゼロになります。 - 非同期メッセージリレー（Asynchronous Message Relay）： 独立したバックグラウンドワーカーが
outbox_eventsから未処理（pending）レコードをポーリングし、ブローカーへ配信したうえで処理済みフラグを付与します。 - 少なくとも1回の配信保証（At-Least-Once Delivery）： ワーカーがKafkaへイベントを発行した直後、PostgreSQLの更新前にクラッシュする可能性があります。そのため後続のコンシューマーは稀に重複イベントを受信することがあるため、冪等（べきとう）性を保つよう設計してください。
DebeziumのようなChange Data Capture（CDC）エンジンは、大規模な分散アーキテクチャで真価を発揮します。しかし、秒間500〜5,000トランザクション規模のシステムであれば、PostgreSQLのFOR UPDATE SKIP LOCKEDを活用したGoのインプロセスワーカーを構築することで、運用オーバーヘッドをほぼかけることなく極めて堅牢な信頼性を実現できます。
Goによるステップバイステップの実装
1. PostgreSQLスキーマの設計
まずは、ビジネスドメインとOutboxイベントログ用のスキーマを定義します。
CREATE TABLE orders (
id UUID PRIMARY KEY,
customer_id UUID NOT NULL,
amount_cents BIGINT NOT NULL,
status VARCHAR(32) NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE outbox_events (
id UUID PRIMARY KEY,
aggregate_type VARCHAR(64) NOT NULL,
aggregate_id UUID NOT NULL,
event_type VARCHAR(64) NOT NULL,
payload JSONB NOT NULL,
status VARCHAR(16) NOT NULL DEFAULT 'PENDING',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
processed_at TIMESTAMPTZ
);
-- 部分インデックスにより、数百万件の履歴レコードが蓄積されてもミリ秒未満のクエリ速度を維持
CREATE INDEX idx_outbox_pending ON outbox_events (created_at) WHERE status = 'PENDING';
status = 'PENDING'に対する部分インデックス（Partial Index）に注目してください。これにより、テーブルに数百万件の履歴レコードが蓄積された後でも、インデックス検索を2ms未満に維持できます。
2. Goにおけるアトミックな書き込み
ドメインデータの変更とOutboxレコードの挿入を、単一のデータベーストランザクションでラップします。
package main
import (
"context"
"database/sql"
"encoding/json"
"time"
"github.com/google/uuid"
)
type OrderPaidEvent struct {
OrderID uuid.UUID `json:"order_id"`
AmountCents int64 `json:"amount_cents"`
Timestamp time.Time `json:"timestamp"`
}
func CompleteOrder(ctx context.Context, db *sql.DB, orderID uuid.UUID, amountCents int64) error {
tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
if err != nil {
return err
}
defer tx.Rollback()
// 1. ドメインの状態を変更
updateQuery := `UPDATE orders SET status = 'PAID' WHERE id = $1`
if _, err := tx.ExecContext(ctx, updateQuery, orderID); err != nil {
return err
}
// 2. イベントペイロードのシリアライズ
payload, err := json.Marshal(OrderPaidEvent{
OrderID: orderID,
AmountCents: amountCents,
Timestamp: time.Now().UTC(),
})
if err != nil {
return err
}
// 3. 同一トランザクション内でイベントをステージング
outboxQuery := `
INSERT INTO outbox_events (id, aggregate_type, aggregate_id, event_type, payload, status)
VALUES ($1, $2, $3, $4, $5, 'PENDING')
`
eventID := uuid.New()
if _, err := tx.ExecContext(ctx, outboxQuery, eventID, "Order", orderID, "OrderPaid", payload); err != nil {
return err
}
return tx.Commit()
}
3. マルチレプリカセーフなバックグラウンドワーカー
複数のワーカーインスタンスを実行することでボトルネックを防ぐことができますが、単純なポーリングではロックの競合が発生します。PostgreSQLはFOR UPDATE SKIP LOCKEDによってこの問題を解決します。各ワーカーが未割り当てのバッチに対して排他ロックを取得し、他のワーカーは即座に次の利用可能な行へとスキップします。
package main
import (
"context"
"database/sql"
"log"
"time"
"github.com/google/uuid"
)
type OutboxRecord struct {
ID uuid.UUID
EventType string
Payload []byte
}
type MessagePublisher interface {
Publish(ctx context.Context, topic string, data []byte) error
}
func StartOutboxWorker(ctx context.Context, db *sql.DB, publisher MessagePublisher, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := processBatch(ctx, db, publisher); err != nil {
log.Printf("Outboxリレーエラー: %v", err)
}
}
}
}
func processBatch(ctx context.Context, db *sql.DB, publisher MessagePublisher) error {
// リレーバッチの実行時間を5秒に制限
batchCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
tx, err := db.BeginTx(batchCtx, nil)
if err != nil {
return err
}
defer tx.Rollback()
query := `
SELECT id, event_type, payload
FROM outbox_events
WHERE status = 'PENDING'
ORDER BY created_at ASC
LIMIT 50
FOR UPDATE SKIP LOCKED
`
rows, err := tx.QueryContext(batchCtx, query)
if err != nil {
return err
}
defer rows.Close()
var batch []OutboxRecord
for rows.Next() {
var rec OutboxRecord
if err := rows.Scan(&rec.ID, &rec.EventType, &rec.Payload); err != nil {
return err
}
batch = append(batch, rec)
}
if err := rows.Err(); err != nil {
return err
}
if len(batch) == 0 {
return nil
}
for _, event := range batch {
if err := publisher.Publish(batchCtx, event.EventType, event.Payload); err != nil {
// ロールバックによりPENDING状態を維持し、次回のtickでリトライ可能にする
return err
}
updateQuery := `
UPDATE outbox_events
SET status = 'PROCESSED', processed_at = $1
WHERE id = $2
`
if _, err := tx.ExecContext(batchCtx, updateQuery, time.Now().UTC(), event.ID); err != nil {
return err
}
}
return tx.Commit()
}
まとめと本番運用のTips
Transactional Outboxパターンは、ドメインの書き込み処理を外部ブローカーの稼働状態から切り離します。これにより、基盤となるデータベースの状態が変更された場合にのみ、確実にイベントを届けることができます。
本番環境へ導入する前に、以下の2つの運用上の保護対策を念頭に置いておきましょう。
- 配信タイムアウトを短く保つ：
publisher.Publishの呼び出しには200〜500msの上限を設定します。発行処理はアクティブなトランザクション内で行われるため、ブローカーの呼び出しが遅延するとデータベース接続とテーブルロックが長時間保持されてしまいます。 - 処理済みレコードを積極的に削除する： 毎日10万件のイベントが追加されるテーブルは急速に肥大化します。
status = 'PROCESSED'かつprocessed_at < NOW() - INTERVAL '7 days'となるレコードを削除またはパーティショニングする日次cronジョブを設定してください。
