무중단 데이터 마이그레이션 — 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 비교(1차 게이트)
[ ] 범위별 checksum 비교
[ ] 최근/오래된/경계값 샘플 비교
[ ] shadow read로 실제 쿼리 결과 비교

[ 컷오버 ]
[ ] read cutover를 퍼센트로 점진 전환
[ ] 역방향 CDC(target->source) 준비
[ ] (선택) 짧은 쓰기 정지 윈도우로 lag 0 확보
[ ] write cutover 후 source 쓰기 차단

[ 롤백/정리 ]
[ ] 롤백 절차 문서화 + 스테이징 리허설
[ ] 안정화 기간 동안 역방향 CDC 유지
[ ] 안전 확인 후 슬롯/커넥터/source 정리