無停止データマイグレーション — Dual-Write、Backfill、CDC で移す

時刻 t0: source の row(id=42) = {amount: 1000}
時刻 t1: backfill が row(id=42) を target へコピー → target = {amount: 1000}
時刻 t2: source の row(id=42) が {amount: 2000} に UPDATE
結果:    source = 2000, target = 1000  (不一致!)

source の順序:  UPDATE amount=2000  →  UPDATE amount=3000
target の適用:  UPDATE amount=3000  →  UPDATE amount=2000  (逆転!)
結果:           source = 3000, target = 2000

                +-----------------+
   write  --->  |  Application    |
                +--------+--------+
                         |
              +----------+----------+
              |                     |
              v                     v
        +-----------+        +-----------+
        |  source   |        |  target   |
        |  (old DB) |        |  (new DB) |
        +-----------+        +-----------+

func CreateOrder(ctx context.Context, o Order) error {
    // 1) まず source へ書く(真実の源)
    if err := sourceDB.Insert(ctx, o); err != nil {
        return fmt.Errorf("source write failed: %w", err)
    }
    // 2) target へも書く
    if err := targetDB.Insert(ctx, o); err != nil {
        // target の失敗をどう扱うか?(以下で議論)
        log.Warn("target write failed", "order_id", o.ID, "err", err)
        metrics.Inc("dual_write.target_failure")
    }
    return nil
}

ケース A: source 成功、target 失敗 → target にデータ欠落
ケース B: source 成功、target 成功、しかしその間にプロセスが死ぬ → 部分反映
ケース C: 並行性 — 二つの要求が異なる順序で二つの DB に到達 → 順序逆転

BEGIN;
  INSERT INTO orders (id, amount, status)
  VALUES (42, 2000, 'CREATED');

  INSERT INTO outbox (aggregate_id, event_type, payload)
  VALUES (42, 'OrderCreated', '{"id":42,"amount":2000}');
COMMIT;

func Backfill(ctx context.Context, batchSize int) error {
    var lastID int64 = 0
    for {
        rows, err := sourceDB.Query(ctx, `
            SELECT id, amount, status, updated_at
            FROM orders
            WHERE id > $1
            ORDER BY id
            LIMIT $2`, lastID, batchSize)
        if err != nil {
            return err
        }
        batch := scanRows(rows)
        if len(batch) == 0 {
            break // 終了
        }
        // target へ UPSERT(冪等)
        if err := targetDB.UpsertBatch(ctx, batch); err != nil {
            return err
        }
        lastID = batch[len(batch)-1].ID
        time.Sleep(throttle) // source 負荷の調整
    }
    return nil
}

1) バージョン/タイムスタンプ比較: target.updated_at >= incoming.updated_at なら無視
2) Backfill は INSERT ... ON CONFLICT DO NOTHING を使う(既にあれば CDC 値を保持)
3) CDC は常に UPSERT(最新値で上書き)

   +-----------+    WAL/binlog    +-----------+    events    +-----------+
   |  source   | ---------------> | Debezium  | -----------> |  Kafka    |
   |  (DB)     |                  | connector |              |  topic    |
   +-----------+                  +-----------+              +-----+-----+
                                                                   |
                                                                   v
                                                            +-----------+
                                                            | consumer  |
                                                            | -> target |
                                                            +-----------+

-- 論理レプリケーションのための設定 (postgresql.conf)
-- wal_level = logical

-- パブリケーション(publication)の作成
CREATE PUBLICATION orders_pub FOR TABLE orders;

-- レプリケーションスロットの確認
SELECT slot_name, plugin, active
FROM pg_replication_slots;

-- レプリケーションスロットが保持している WAL サイズの確認
SELECT slot_name,
       pg_size_pretty(
         pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)
       ) AS retained_wal
FROM pg_replication_slots;

{
  "name": "orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "source-db",
    "database.port": "5432",
    "database.user": "debezium",
    "database.dbname": "shop",
    "topic.prefix": "cdc",
    "plugin.name": "pgoutput",
    "slot.name": "orders_slot",
    "publication.name": "orders_pub",
    "table.include.list": "public.orders",
    "snapshot.mode": "initial"
  }
}

{
  "op": "u",
  "before": { "id": 42, "amount": 1000, "status": "CREATED" },
  "after":  { "id": 42, "amount": 2000, "status": "PAID" },
  "source": { "lsn": 123456789, "ts_ms": 1718500000000 },
  "ts_ms": 1718500000123
}

func ApplyEvent(ctx context.Context, e CDCEvent) error {
    switch e.Op {
    case "c", "u", "r":
        // LSN 比較で古いイベントは無視
        _, err := targetDB.Exec(ctx, `
            INSERT INTO orders (id, amount, status, _lsn)
            VALUES ($1, $2, $3, $4)
            ON CONFLICT (id) DO UPDATE
              SET amount = EXCLUDED.amount,
                  status = EXCLUDED.status,
                  _lsn   = EXCLUDED._lsn
            WHERE orders._lsn < EXCLUDED._lsn`,
            e.After.ID, e.After.Amount, e.After.Status, e.LSN)
        return err
    case "d":
        _, err := targetDB.Exec(ctx,
            `DELETE FROM orders WHERE id = $1 AND _lsn < $2`,
            e.Before.ID, e.LSN)
        return err
    }
    return nil
}

[1] CDC スロット作成(この時点からの変更を捉え始める)
        |
        v
[2] Backfill 開始(過去データをバッチで target に埋める)
        |   同時に CDC は新規変更を target に適用
        v
[3] Backfill 完了 + CDC lag ~ 0  (source == target が収束)
        |
        v
[4] 検証(row count、checksum、サンプル比較)
        |
        v
[5] Shadow read(target を読むが結果は捨て、比較のみ)
        |
        v
[6] Read cutover(読み取りを target へ切り替え)
        |
        v
[7] Write cutover(書き込みを target へ切り替え、source を停止)
        |
        v
[8] 安定化後、source/CDC を整理

-- source
SELECT count(*) FROM orders;
-- target
SELECT count(*) FROM orders;

-- 範囲ごとのチェックサム (PostgreSQL)
SELECT
  (id / 100000) AS bucket,
  count(*) AS cnt,
  md5(string_agg(
    id || ':' || amount || ':' || status,
    ',' ORDER BY id
  )) AS checksum
FROM orders
GROUP BY (id / 100000)
ORDER BY bucket;

検証の優先順位:
1) 最も最近修正された行(CDC lag による欠落の検出)
2) 最も古い行(backfill 初期バグの検出)
3) 境界値 / 特異値(NULL、負数、空文字列)
4) 無作為 N 件

func GetOrder(ctx context.Context, id int64) (Order, error) {
    // 真実は依然として source
    o, err := sourceDB.GetOrder(ctx, id)
    if err != nil {
        return Order{}, err
    }
    // shadow: target も読んで比較(非同期、失敗は無視)
    go func() {
        shadow, serr := targetDB.GetOrder(context.Background(), id)
        if serr != nil {
            metrics.Inc("shadow.read_error")
            return
        }
        if !shadow.Equal(o) {
            metrics.Inc("shadow.mismatch")
            log.Warn("shadow mismatch", "id", id,
                "source", o, "target", shadow)
        }
    }()
    return o, nil
}

read_from_target = feature_flag("orders.read_target")  // 0% -> 1% -> 10% -> 50% -> 100%

1) 逆方向 CDC の準備: target -> source 同期をあらかじめ構成
   (ロールバック時に target の新規データを source へ戻すため)
2) 短い書き込み停止ウィンドウ(任意): 一瞬だけ書き込みを止め、
   CDC lag を 0 にしてから切り替えると最も安全
3) 冪等キーの保証: 切り替え直後の再試行が重複を作らないように

T-0   write を source で受けつつ CDC が target に反映中(lag ~0 を維持)
T-1   read を 100% target に切り替え、24 時間観察
T-2   逆方向 CDC(target->source)を有効化
T-3   (任意)5 秒間書き込みを一時停止、双方向 lag 0 を確認
T-4   write を target に切り替え
T-5   source への書き込みを遮断、target が唯一の真実
T-6   1 週間の安定化後、順方向 CDC、レプリケーションスロット、source を整理

1) UPSERT(ON CONFLICT): 存在すれば更新、なければ挿入 → 重複 INSERT を防止
2) バージョンガード: _lsn / updated_at / version カラムで古い変更を無視
3) 自然キーまたは決定的キー: 同じ入力は常に同じ PK へマッピング
4) 冪等トークン: メッセージ ID を処理記録テーブルに保存して重複を遮断

CREATE TABLE processed_events (
  event_id   TEXT PRIMARY KEY,
  processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- 適用トランザクションの中で
BEGIN;
  INSERT INTO processed_events (event_id)
  VALUES ('evt-abc-123')
  ON CONFLICT (event_id) DO NOTHING;
  -- 上の INSERT が 0 件なら既に処理したイベント → 本作業をスキップ
  -- ... 実際のデータ適用 ...
COMMIT;

[既存] orders(id, user_id, item_json, amount, ...)
[目標] orders(id, user_id, amount, ...)
       order_items(id, order_id, sku, qty, price)

[既存] monolith_db.payments  (モノリスが直接アクセス)
[目標] payment_service_db.payments  (決済サービスのみがアクセス)

[ 準備 ]
[ ] target スキーマの定義とインデックス設計を完了
[ ] 変換(transform)ロジックを backfill/CDC が共有する共通モジュールとして作成
[ ] CDC スロット/コネクタの構成、snapshot.mode を決定
[ ] 冪等な適用の保証(UPSERT + バージョンガード)
[ ] dedupe テーブルまたは冪等トークンを準備

[ 実行 ]
[ ] CDC スロットを backfill より先に作成
[ ] keyset ページネーション + スロットリングで backfill
[ ] backfill の checkpoint を保存(再起動に備える)
[ ] CDC lag、レプリケーションスロット lag、source 負荷を監視

[ 検証 ]
[ ] row count 比較(一次ゲート)
[ ] 範囲ごとの checksum 比較
[ ] 最近/最古/境界値のサンプル比較
[ ] shadow read で実際のクエリ結果を比較

[ カットオーバー ]
[ ] read cutover をパーセントで段階的に切り替え
[ ] 逆方向 CDC(target->source)を準備
[ ] (任意)短い書き込み停止ウィンドウで lag 0 を確保
[ ] write cutover 後に source の書き込みを遮断

[ ロールバック/整理 ]
[ ] ロールバック手順の文書化 + ステージングでのリハーサル
[ ] 安定化期間の間は逆方向 CDC を維持
[ ] 安全確認後にスロット/コネクタ/source を整理