互換性を保つschema migrationを段階実行する

到達目標

1. schema changeを互換なexpand-contract phaseと明示的な遷移条件へ分解できる

   • expand-contract段階、観測、停止、ロールバックを含む移行計画
   • database change managementとPostgreSQL制約を互換性へ結ぶ5分発表

2. dual write、backfill、dual readの観測から停止とrollbackを判断できる

   • expand-contract段階、観測、停止、ロールバックを含む移行計画
   • 互換性、観測、停止、rollbackが欠けるmigrationを診断する回答

3. backfill error rateだけが変わる条件で同じ計画の継続判断を再評価できる

   • 互換性、観測、停止、rollbackが欠けるmigrationを診断する回答
   • backfill error rateだけを変えて継続・停止・rollbackを再評価した記録

能力の進行

1. recognize

   expand、dual write、backfill、dual read、contractの責務を区別できる

   証拠: expand-contract段階、観測、停止、ロールバックを含む移行計画

2. explain

   old readerとnew readerを同時に支える期間が必要な理由を説明できる

   証拠: database change managementとPostgreSQL制約を互換性へ結ぶ5分発表

3. apply

   各phaseへ観測、停止条件、rollback、compatibilityを割り当てられる

   証拠: expand-contract段階、観測、停止、ロールバックを含む移行計画

4. diagnose

   backfill成功件数だけに隠れたerror、drift、reader incompatibilityを反証できる

   証拠: 互換性、観測、停止、rollbackが欠けるmigrationを診断する回答

5. lead

   application、database、operationsのownerと停止権限を含む段階移行を主導できる

   証拠: backfill error rateだけを変えて継続・停止・rollbackを再評価した記録

なぜ重要か

database schemaはapplication version、batch、replica、運用手順が共有する契約である。破壊的変更を一度に行うと、DDL command自体が成功してもold reader、遅延したwriter、backfill、logical replicationのいずれかが壊れる。安全なmigrationは「実行できたか」ではなく、混在versionでsystem outcomeを保てるかを段階ごとに検証する。

DORAのdatabase change managementは小さく頻繁でapplicationと協調した変更を重視する。PostgreSQL 18のALTER TABLEは操作ごとのlockと制約を確認する一次資料であり、logical replicationのRestrictionsはschema定義が自動複製されない等の境界を示す。一般論をそのまま適用せず、対象version、table規模、replication構成で検証する。

メンタルモデル

migrationを五つのstateへ分ける。expandで互換な構造を追加し、dual writeで新旧表現を同時更新し、backfillで既存rowを移す。dual readで新旧読取結果を比較し、十分な観測後にだけcontractで旧構造を除く。次phaseへの遷移は時刻ではなく検証済みのguardから導く。

rollbackは必ずしもDDLの逆操作ではない。feature flagでold readへ戻す、dual writeを止める、backfill checkpointへ戻る、contractを延期するなど、各phaseで安全な回復形が異なる。不可逆なcontractには、旧利用者がいないことを示す観測gateが必要である。

動く例で考える

顧客名fieldを五phaseで移行する

前提

lesson-defined simulated migrationであり、実database、実customer row、production lock時間を観測したものではない。PostgreSQL固有の実行計画を保証せず、phase判断の因果だけを検査する。

入力

expand、dual write、backfill、dual read、contractの固定plan、old/new reader成功、dual write parity、backfill error rateとthresholdを与える。

操作

各phaseのcompatibility、observation、stop condition、rollbackを入力から構成する。baselineは低いbackfill error rate、transferはその値だけを高くし、同じplanへ判断関数を適用する。

観測

baselineはcontinue、transferはrollbackになる。判断を常にcontinueへ差し替えるとmigration-causal-invariantが検知する。

結論

command successはsystem outcomeではない。互換性とerror thresholdを満たす観測がcontinueを支え、違反時に実行されたrollback結果が安全性の証拠になる。

python3.13 - <<'PY'
import json

HARNESS = "migration_state_machine_lab_v1"
TRANSFER_TASK = (
    "backfill error rateだけを変え、同じ移行計画の継続・停止・"
    "ロールバック判断を再評価する"
)
CANONICAL_PHASE_NAMES = (
    "expand",
    "dual-write",
    "backfill",
    "dual-read",
    "contract",
)
PHASE_INPUTS = [
    {
        "name": "expand",
        "compatibility": "nullable customer_name_v2 keeps old reader valid",
        "observation": "schema exists and old reader fixture succeeds",
        "stop_condition": "old reader failure or blocking lock exceeds budget",
        "rollback": "stop rollout and remove unused nullable field",
    },
    {
        "name": "dual-write",
        "compatibility": "old and new names are written from one input",
        "observation": "write parity and write error rate",
        "stop_condition": "parity mismatch or write error exceeds threshold",
        "rollback": "disable new-field write and retain old-field write",
    },
    {
        "name": "backfill",
        "compatibility": "old field remains source for old reader",
        "observation": "processed rows, error rate, checkpoint and replica lag",
        "stop_condition": "backfill error rate exceeds threshold",
        "rollback": "pause at checkpoint and clear unverified new values",
    },
    {
        "name": "dual-read",
        "compatibility": "new reader falls back to old field",
        "observation": "old and new read result parity",
        "stop_condition": "read mismatch or fallback rate exceeds threshold",
        "rollback": "route all reads to old field",
    },
    {
        "name": "contract",
        "compatibility": "old reader usage is zero before removal",
        "observation": "old-field read and write counters remain zero",
        "stop_condition": "any old reader or writer is observed",
        "rollback": "defer removal and restore compatibility window",
    },
]
THRESHOLDS = {
    "max_backfill_error_rate": 0.01,
}
BASELINE_OBSERVATION = {
    "old_reader_successes": 200,
    "old_reader_attempts": 200,
    "new_reader_successes": 200,
    "new_reader_attempts": 200,
    "dual_write_matches": 200,
    "dual_write_attempts": 200,
    "backfill_rows": 500,
    "backfill_errors": 2,
}
TRANSFER_OBSERVATION = {
    "old_reader_successes": 200,
    "old_reader_attempts": 200,
    "new_reader_successes": 200,
    "new_reader_attempts": 200,
    "dual_write_matches": 200,
    "dual_write_attempts": 200,
    "backfill_rows": 500,
    "backfill_errors": 20,
}
ROLLBACK_OBSERVATION = {
    "executed": True,
    "observed": True,
    "expected_state": "backfill-paused-at-checkpoint",
    "actual_state": "backfill-paused-at-checkpoint",
}
OBSERVATION_FIELDS = {
    "old_reader_successes",
    "old_reader_attempts",
    "new_reader_successes",
    "new_reader_attempts",
    "dual_write_matches",
    "dual_write_attempts",
    "backfill_rows",
    "backfill_errors",
}
PHASE_FIELDS = {
    "name",
    "compatibility",
    "observation",
    "stop_condition",
    "rollback",
}
ROLLBACK_FIELDS = {
    "executed",
    "observed",
    "expected_state",
    "actual_state",
}

def validate_migration_inputs(baseline, transferred, thresholds):
    for observation in (baseline, transferred):
        if (
            type(observation) is not dict
            or set(observation) != OBSERVATION_FIELDS
            or not all(
                type(value) is int and value >= 0
                for value in observation.values()
            )
            or observation["old_reader_attempts"] <= 0
            or observation["new_reader_attempts"] <= 0
            or observation["dual_write_attempts"] <= 0
            or observation["backfill_rows"] <= 0
        ):
            raise AssertionError(
                "migration-transfer-invariant: invalid observation schema"
            )
    if (
        type(thresholds) is not dict
        or set(thresholds) != {"max_backfill_error_rate"}
        or type(thresholds["max_backfill_error_rate"])
        not in {int, float}
        or type(thresholds["max_backfill_error_rate"]) is bool
        or thresholds["max_backfill_error_rate"] < 0
    ):
        raise AssertionError(
            "migration-transfer-invariant: invalid threshold schema"
        )
    changed = [
        field
        for field in sorted(set(baseline) | set(transferred))
        if (
            field not in baseline
            or field not in transferred
            or baseline[field] != transferred[field]
        )
    ]
    if changed != ["backfill_errors"]:
        raise AssertionError(
            "migration-transfer-invariant: transfer changed wrong fields"
        )
    return changed

def ratio(numerator, denominator):
    if denominator <= 0:
        raise AssertionError(
            "migration-causal-invariant: denominator must be positive"
        )
    return numerator / denominator

def derive_observation(source):
    return {
        "old_reader_supported": (
            source["old_reader_successes"] == source["old_reader_attempts"]
        ),
        "new_reader_supported": (
            source["new_reader_successes"] == source["new_reader_attempts"]
        ),
        "dual_write_verified": (
            source["dual_write_matches"] == source["dual_write_attempts"]
        ),
        "backfill_error_rate": ratio(
            source["backfill_errors"],
            source["backfill_rows"],
        ),
    }

def decide_migration(observation, thresholds):
    compatible = (
        observation["old_reader_supported"]
        and observation["new_reader_supported"]
        and observation["dual_write_verified"]
    )
    within_error_budget = (
        observation["backfill_error_rate"]
        <= thresholds["max_backfill_error_rate"]
    )
    return "continue" if compatible and within_error_budget else "rollback"

def evaluate(source, thresholds):
    observation = derive_observation(source)
    decision = decide_migration(observation, thresholds)
    # decisionの置換を見逃さないため、入力から独立に導いたguardと
    # state machineの出力を照合してfail closedにする。
    guard_allows_continue = (
        observation["old_reader_supported"]
        and observation["new_reader_supported"]
        and observation["dual_write_verified"]
        and observation["backfill_error_rate"]
        <= thresholds["max_backfill_error_rate"]
    )
    expected_decision = (
        "continue" if guard_allows_continue else "rollback"
    )
    if decision != expected_decision:
        raise AssertionError(
            "migration-causal-invariant: decision bypassed observed guard"
        )
    return {
        "observation": observation,
        "decision": decision,
    }

def build_phases(phase_inputs):
    # phaseごとの停止と回復を同じ構造へ必須化し、happy pathだけの
    # 手順がcontractまで流れることを防ぐ。
    if (
        type(phase_inputs) is not list
        or not phase_inputs
        or any(
            type(phase) is not dict
            or set(phase) != PHASE_FIELDS
            or not all(
                type(value) is str and value
                for value in phase.values()
            )
            for phase in phase_inputs
        )
    ):
        raise AssertionError(
            "migration-plan-invariant: invalid phase schema"
        )
    if (
        tuple(phase["name"] for phase in phase_inputs)
        != CANONICAL_PHASE_NAMES
    ):
        raise AssertionError(
            "migration-plan-invariant: noncanonical phase sequence"
        )
    phases = [
        {
            "name": phase["name"],
            "compatibility": phase["compatibility"],
            "observation": phase["observation"],
            "stop_condition": phase["stop_condition"],
            "rollback": phase["rollback"],
        }
        for phase in phase_inputs
    ]
    if not all(
        phase["compatibility"]
        and phase["observation"]
        and phase["stop_condition"]
        and phase["rollback"]
        for phase in phases
    ):
        raise AssertionError(
            "migration-causal-invariant: incomplete phase"
        )
    return phases

def verify_rollback(decision, observation):
    if (
        type(observation) is not dict
        or set(observation) != ROLLBACK_FIELDS
        or type(observation["executed"]) is not bool
        or type(observation["observed"]) is not bool
        or type(observation["expected_state"]) is not str
        or not observation["expected_state"]
        or type(observation["actual_state"]) is not str
        or not observation["actual_state"]
    ):
        raise AssertionError(
            "migration-rollback-outcome-invariant: invalid schema"
        )
    system_outcome = (
        decision == "rollback"
        and observation["executed"]
        and observation["observed"]
        and observation["actual_state"]
        == observation["expected_state"]
    )
    if decision == "rollback" and not system_outcome:
        raise AssertionError(
            "migration-rollback-outcome-invariant: "
            "rollback outcome was not observed"
        )
    return {
        "executed": observation["executed"],
        "observed": observation["observed"],
        "expected_state": observation["expected_state"],
        "actual_state": observation["actual_state"],
        "system_outcome": system_outcome,
    }

def main():
    phases = build_phases(PHASE_INPUTS)
    transferred_plan = build_phases(PHASE_INPUTS)
    if phases != transferred_plan:
        raise AssertionError(
            "migration-plan-invariant: plan snapshots drifted"
        )
    changed_input_fields = validate_migration_inputs(
        BASELINE_OBSERVATION,
        TRANSFER_OBSERVATION,
        THRESHOLDS,
    )
    baseline = evaluate(BASELINE_OBSERVATION, THRESHOLDS)
    transferred = evaluate(TRANSFER_OBSERVATION, THRESHOLDS)
    if (
        baseline["decision"] != "continue"
        or transferred["decision"] != "rollback"
    ):
        raise AssertionError(
            "migration-causal-invariant: transfer changed wrong cause"
        )
    rollback_outcome = verify_rollback(
        transferred["decision"],
        ROLLBACK_OBSERVATION,
    )
    compatibility = {
        "old_reader_supported": (
            baseline["observation"]["old_reader_supported"]
        ),
        "new_reader_supported": (
            baseline["observation"]["new_reader_supported"]
        ),
        "dual_write_verified": (
            baseline["observation"]["dual_write_verified"]
        ),
    }
    report = {
        "harness": HARNESS,
        "fixture_metadata": {
            "kind": "simulated",
            "provenance": "lesson-defined customer-name migration inputs",
            "limitations": (
                "simulated countsでありproduction lock、traffic、"
                "replication lag、PostgreSQL実行結果を表さない"
            ),
            "synthetic_or_observed_explicit": (
                "値はすべてsimulated入力またはその計算結果"
            ),
        },
        "runtime_bound": {
            "records": (
                len(PHASE_INPUTS)
                + len(BASELINE_OBSERVATION)
                + len(TRANSFER_OBSERVATION)
            ),
            "subprocesses": 0,
            "maximum_iterations": (
                len(PHASE_INPUTS) + len(BASELINE_OBSERVATION)
            ),
        },
        "external_network_used": False,
        "migration_plan": {
            "phases": phases,
        },
        "compatibility": compatibility,
        "backfill_error_rate_transfer": {
            "changed_assumption": "backfill-error-rate",
            "changed_fields": ["backfill_error_rate"],
            "same_migration_plan": phases == transferred_plan,
            "baseline_plan": phases,
            "transferred_plan": transferred_plan,
            "baseline_error_rate": (
                baseline["observation"]["backfill_error_rate"]
            ),
            "transferred_error_rate": (
                transferred["observation"]["backfill_error_rate"]
            ),
            "baseline_decision": baseline["decision"],
            "transferred_decision": transferred["decision"],
            "rollback_outcome": rollback_outcome,
        },
        "command_success_distinction": {
            "command_completed": True,
            "system_outcome_checked": (
                all(compatibility.values())
                and baseline["decision"] == "continue"
                and transferred["decision"] == "rollback"
                and rollback_outcome["system_outcome"]
            ),
            "command_success_equals_system_outcome": False,
            "outcome_evidence": (
                "reader compatibility、write parity、backfill error、"
                "観測済みrollback system outcomeを検査"
            ),
        },
        "mastery_evidence": {
            "lab_steps": [
                {"step": 1, "evidence": phases},
                {"step": 2, "evidence": baseline["observation"]},
                {"step": 3, "evidence": rollback_outcome},
            ],
            "assessments": [
                {"assessment": 1, "evidence": compatibility},
                {
                    "assessment": 2,
                    "evidence": transferred["observation"],
                },
            ],
            "rubric_dimensions": [
                "technical-correctness",
                "judgment",
                "evidence",
                "communication",
            ],
            "transfer": {
                "task": TRANSFER_TASK,
                "changed_assumption": "backfill-error-rate",
                "evidence": transferred["decision"],
            },
        },
    }
    print(json.dumps(report, ensure_ascii=False, sort_keys=True))

try:
    main()
except AssertionError as error:
    raise SystemExit(str(error)) from None
PY

トレードオフと失敗モード

• 誤診: ALTER TABLE commandが成功したのでmigrationは完了した。反証: command successはreader compatibility、dual write parity、backfill error、replica、利用者可視のsystem outcomeを検証していない。
• 誤診: dual write期間を短くすれば不整合riskも必ず減る。反証: 観測sampleが不足すれば低頻度のmismatchを見逃し、contract後にrollback不能な破壊へ進む。
• 誤診: rollbackは逆向きDDLを用意することである。反証: phaseごとにold readへ戻す、writeを止める、checkpointで停止するなど回復形が異なり、回復後のsystem outcome確認まで必要である。

知識チェック

1. expand、dual write、backfill、dual read、contractの順序が互換性を作る理由を説明せよ。
2. backfill processed rowsだけではcontinueを判断できない理由は何か。
3. contract前にold readerを一件観測した時のstopとrollbackを定義せよ。
4. backfill error rateだけを変えるtransferで固定すべきplan証拠は何か。

出典と次の学習

DORA Capability: Database change management（accessed 2026-07-31）をdelivery practiceの根拠に、PostgreSQL 18 Documentation: ALTER TABLEをDDLとlockの確認に、PostgreSQL 18 Documentation: Restrictionsをlogical replication境界の確認に使う。完全な書誌情報と参照先はlesson metadataに分離している。

次はcore-23で、migrationを含むincidentのevidence timeline、意思決定、contributing factor、検証可能なactionをnon-blaming reviewへ変換する。

実践ラボ

顧客名columnのexpand-contract migrationを設計する

提出成果物: expand-contract段階、観測、停止、ロールバックを含む移行計画

1. old readerとnew readerを同時に支える五つのphase、観測、停止条件、rollbackを定義する
2. fixed simulated observationからcompatibilityとbackfill error rateを検査しcontinue判断を導く
3. backfill error rateだけを上げ、同じmigration planがrollbackへ遷移することを検証する

説明して理解を確かめる

5分で、DORA database change management、PostgreSQL 18 ALTER TABLE、logical replication Restrictionsを、具体的なexpand-contract順序と停止判断へどう変換するか説明する。

アセスメント

1. 問い: new columnを追加して全rowを一括更新し、直後にold columnを削除する計画を評価せよ。

   期待する証拠: lock、old reader、new reader、dual write、backfill、観測、contract gate、rollback

2. 問い: migration commandが成功したので配信完了とする判断を評価せよ。

   期待する証拠: command successとsystem outcomeの区別、reader compatibility、write parity、backfill error、rollback outcome

別問題へ転用する

backfill error rateだけを変え、同じ移行計画の継続・停止・ロールバック判断を再評価する

復習スケジュール

1. 1日後

   expandとcontractの間にold readerを支える理由は何か

2. 7日後

   backfillで件数以外に観測すべき結果は何か

3. 30日後

   command successとmigration system outcomeをどう分けるか

4. 90日後

   expandとcontractの間にold readerを支える理由は何か

評価ルーブリック

出典

以下の外部資料は利用者が選択したときだけ開きます。

• DORA Capability: Database change management (accessed 2026-07-31) (primary)
• PostgreSQL 18 Documentation: ALTER TABLE (primary)
• PostgreSQL 18 Documentation: Restrictions (primary)