API契約を失敗、再送、進化まで設計する

到達目標

1. HTTP意味論、schema、業務不変条件、認可、失敗形式を分離したAPI契約を記述できる

   • version、冪等key、状態遷移、Problem Details、互換性判定を含む実行可能API契約
   • 冪等性と同一応答、schema妥当性と業務妥当性、認可を混同しない5分説明

2. 応答喪失後の再送で同じ副作用を一回に保ちつつ、同じ応答やexactly-onceを誤って保証しない

   • version、冪等key、状態遷移、Problem Details、互換性判定を含む実行可能API契約
   • 再送と契約進化の障害シナリオを観測で切り分ける回答

3. source、wire、意味の互換性を区別し、長時間オフラインの利用者へ契約変更を移せる

   • 再送と契約進化の障害シナリオを観測で切り分ける回答
   • 長時間オフライン端末に対する互換な同期と失敗契約

能力の進行

1. recognize

   HTTP method、表現schema、業務意味、認証、認可、失敗形式を別の契約面として分類できる

   証拠: version、冪等key、状態遷移、Problem Details、互換性判定を含む実行可能API契約

2. explain

   同じ冪等keyの再送が同じ効果を保っても同じresponseやexactly-onceを意味しない理由を説明できる

   証拠: 冪等性と同一応答、schema妥当性と業務妥当性、認可を混同しない5分説明

3. apply

   version、再送、状態遷移、Problem Detailsを含むローカル契約fixtureを実装できる

   証拠: version、冪等key、状態遷移、Problem Details、互換性判定を含む実行可能API契約

4. diagnose

   schema違反、意味違反、認可拒否、version非互換を証拠から切り分けられる

   証拠: 再送と契約進化の障害シナリオを観測で切り分ける回答

5. lead

   利用者の更新遅延と撤回困難性を考慮し、互換性方針と廃止条件をレビューできる

   証拠: 長時間オフライン端末に対する互換な同期と失敗契約

なぜ重要か

APIはfieldとendpointの一覧ではない。consumerはHTTP methodとstatus、表現の形、fieldの意味、認可、状態遷移、失敗後の次の操作へ依存する。生成したOpenAPI文書やJSON Schemaが妥当でも、別tenantの操作を許したり、同じ再送で副作用を二重化したりすれば契約は壊れている。

長時間オフラインになる端末では、providerが新versionを公開しても旧clientが数週間残る。応答を受け取れなかったclientは、requestが未到達なのか、処理済みで応答だけ失われたのかを区別できない。互換性、冪等性、構造化した失敗を一つの進化方針として設計する必要がある。

メンタルモデル

API契約を六つの面へ分ける。transportはHTTP methodとstatus、structureはJSON Schema、semanticsはfieldと状態遷移の意味、identityは認証主体、authorityはその主体が対象へ行える操作、failureは失敗の分類と回復操作である。一面の検証結果を別の面の保証へ拡張しない。

RFC 9110のmethod意味論は共通の土台だが、業務操作の副作用境界はapplicationが定義する。RFC 9457のProblem Detailsはtype、title、status、detail、instanceという標準memberを定義するが、RFC上これらは一律の必須memberではない。このAPIはtype、title、statusをprofileとして必須にし、detailとinstanceを任意にする。type URIを識別子として使う場合も、本文取得に成功しなければ処理できない設計にはしない。

動く例で考える

現場端末の作業完了を同じkeyで再送する

前提

field-7の端末は最大14日offlineになる。v1とv2は同じCompleteWorkコマンドを受け、tenant-aの主体だけがtenant-aの作業を完了できる。

入力

operation-42をkey-42で初回送信し、response喪失後に同じscopeとpayloadで再送する。同じscopeとkeyでpayloadだけが異なる衝突、同じraw keyを使う別tenant、3種類の契約変更、schema上正しい別tenant requestも与える。

操作

serverはtenant、認証済みprincipal、route、keyとpayload SHA-256を業務効果と共に保存する。同じscopeでfingerprintが違えば409にし、別tenantなら独立した効果として処理する。注入したRFC 3339時計でresponse時刻を生成し、構造検査と認可判定を独立に記録する。

観測

初回と再送のeffect countはともに1、response時刻は異なる。payload衝突ではeffectを増やさず、別tenantでは2件目を作る。v1とv2、3状態遷移、API profileのProblem Details、三軸の互換性分類、schema validだがunauthorizedの反例をJSONへ出力する。

結論

同じ効果を再現できても同じresponseやexactly-once配送を証明したことにはならない。raw keyだけを共有せず認証scopeを分離し、payload取り違えを409にする。旧clientが未知enumを拒否するなら、schema上の追加でも意味的riskである。

python3.13 - <<'PY'
import hashlib
import json

HARNESS = "api_contract_lab_v1"
SUPPORTED_VERSIONS = ["v1", "v2"]
ROUTE = "POST /work-items/{id}:complete"

def structurally_valid(request):
    required = {
        "version",
        "operation_id",
        "idempotency_key",
        "tenant",
        "work_item_id",
        "result",
    }
    return (
        required.issubset(request)
        and request["version"] in SUPPORTED_VERSIONS
        and all(isinstance(request[name], str) and request[name] for name in required)
    )

def authorized(request, principal):
    return request["tenant"] == principal["tenant"]

def request_fingerprint(request):
    payload = {
        name: request[name]
        for name in (
            "version",
            "operation_id",
            "tenant",
            "work_item_id",
            "result",
        )
    }
    canonical = json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()

class FixedClock:
    def __init__(self, values):
        self.values = list(values)
        self.index = 0

    def __call__(self):
        if self.index >= len(self.values):
            raise AssertionError("fixed clock exhausted")
        value = self.values[self.index]
        self.index += 1
        return value

class WorkService:
    def __init__(self, clock):
        self.results = {}
        self.effect_count = 0
        self.transitions = []
        self.clock = clock

    def complete(self, request, principal, route):
        if not structurally_valid(request) or not authorized(request, principal):
            raise ValueError("request failed validation or authorization")
        storage_key = (
            principal["tenant"],
            principal["subject"],
            route,
            request["idempotency_key"],
        )
        scope = {
            "tenant": principal["tenant"],
            "principal": principal["subject"],
            "route": route,
            "idempotency_key": request["idempotency_key"],
        }
        fingerprint = request_fingerprint(request)
        response_generated_at = self.clock()
        stored = self.results.get(storage_key)
        if stored is not None:
            if stored["request_fingerprint"] != fingerprint:
                return {
                    "status": 409,
                    "effect_applied": False,
                    "effect_count": self.effect_count,
                    "scope": scope,
                    "request_fingerprint": fingerprint,
                    "response_generated_at": response_generated_at,
                    "problem_details": {
                        "type": "urn:problem:idempotency-key-conflict",
                        "title": "Idempotency key payload conflict",
                        "status": 409,
                    },
                }
            return {
                "status": 200,
                "effect_applied": False,
                "effect": dict(stored["effect"]),
                "effect_count": self.effect_count,
                "scope": scope,
                "request_fingerprint": fingerprint,
                "response_generated_at": response_generated_at,
            }

        self.transitions.extend(
            [
                {"from": "assigned", "to": "in-progress"},
                {"from": "in-progress", "to": "completed"},
                {"from": "completed", "to": "synced"},
            ]
        )
        self.effect_count += 1
        effect = {
                "operation_id": request["operation_id"],
                "work_item_id": request["work_item_id"],
                "state": "completed",
        }
        self.results[storage_key] = {
            "request_fingerprint": fingerprint,
            "effect": effect,
        }
        return {
            "status": 200,
            "effect_applied": True,
            "effect": dict(effect),
            "effect_count": self.effect_count,
            "scope": scope,
            "request_fingerprint": fingerprint,
            "response_generated_at": response_generated_at,
        }

def classify_change(change):
    # source・wire・semanticを潰さず、旧client traceへ根拠を戻す。
    outcomes = {
        "add_optional_field": {
            "source": {
                "compatible": True,
                "evidence": "旧clientは未知fieldを読み飛ばす",
            },
            "wire": {
                "compatible": True,
                "evidence": "既存required fieldと型を維持する",
            },
            "semantic": {
                "compatible": True,
                "evidence": "省略時の既存意味を維持する",
            },
            "trace_id": "trace-add-optional",
        },
        "remove_required_field": {
            "source": {
                "compatible": False,
                "evidence": "生成済みclientのaccessorが参照できない",
            },
            "wire": {
                "compatible": False,
                "evidence": "旧schemaがrequired field欠落を拒否する",
            },
            "semantic": {
                "compatible": False,
                "evidence": "旧clientが作業完了を判定できない",
            },
            "trace_id": "trace-remove-required",
        },
        "expand_enum": {
            "source": {
                "compatible": False,
                "evidence": "網羅switchの旧clientが未知値を扱えない",
            },
            "wire": {
                "compatible": True,
                "evidence": "JSON stringとしては配送できる",
            },
            "semantic": {
                "compatible": False,
                "evidence": "未知状態を既定値へ誤変換する",
            },
            "trace_id": "trace-expand-enum",
        },
    }
    return outcomes[change]

def main():
    request = {
        "version": "v1",
        "operation_id": "operation-42",
        "idempotency_key": "key-42",
        "tenant": "tenant-a",
        "work_item_id": "work-42",
        "result": "completed",
    }
    principal = {"subject": "field-7", "tenant": "tenant-a"}
    assert structurally_valid(request)
    assert authorized(request, principal)

    clock = FixedClock(
        [
            "2026-08-01T09:00:00Z",
            "2026-08-01T09:00:01Z",
            "2026-08-01T09:00:02Z",
            "2026-08-01T09:00:03Z",
        ]
    )
    service = WorkService(clock)
    initial = service.complete(request, principal, ROUTE)
    replay = service.complete(request, principal, ROUTE)
    same_effect = initial["effect"] == replay["effect"]
    same_response = initial == replay
    assert initial["effect_count"] == replay["effect_count"] == 1
    assert same_effect and not same_response

    conflicting_request = dict(request)
    conflicting_request["operation_id"] = "operation-conflict"
    conflicting_request["result"] = "cancelled"
    conflict = service.complete(conflicting_request, principal, ROUTE)
    assert conflict["status"] == 409
    assert not conflict["effect_applied"]
    assert conflict["effect_count"] == 1
    assert conflict["request_fingerprint"] != initial["request_fingerprint"]

    other_tenant_request = dict(request)
    other_tenant_request["tenant"] = "tenant-b"
    other_tenant_request["operation_id"] = "operation-tenant-b"
    other_tenant_principal = {"subject": "field-7", "tenant": "tenant-b"}
    other_tenant = service.complete(
        other_tenant_request,
        other_tenant_principal,
        ROUTE,
    )
    assert other_tenant["effect_applied"]
    assert other_tenant["effect_count"] == 2
    assert other_tenant["effect"]["operation_id"] != initial["effect"]["operation_id"]

    cross_tenant = dict(request)
    cross_tenant["tenant"] = "tenant-b"
    boundary = {
        "schema_valid": structurally_valid(cross_tenant),
        "authorized": authorized(cross_tenant, principal),
        "reason": "principal-tenant-mismatch",
    }
    assert boundary["schema_valid"] and not boundary["authorized"]

    compatibility = {
        change: classify_change(change)
        for change in (
            "add_optional_field",
            "remove_required_field",
            "expand_enum",
        )
    }
    offline_client_trace = [
        {
            "id": "trace-add-optional",
            "client": "v1-field-device",
            "observed": "unknown field ignored",
        },
        {
            "id": "trace-remove-required",
            "client": "v1-field-device",
            "observed": "required result missing",
        },
        {
            "id": "trace-expand-enum",
            "client": "v1-field-device",
            "observed": "unknown enum rejected",
        },
    ]
    assert {
        change["trace_id"] for change in compatibility.values()
    } == {trace["id"] for trace in offline_client_trace}
    assert initial["scope"] == replay["scope"]
    assert initial["request_fingerprint"] == replay["request_fingerprint"]
    assert initial["response_generated_at"] != replay["response_generated_at"]
    return {
        "harness": HARNESS,
        "fixture": "offline-field-client-v1",
        "supported_versions": SUPPORTED_VERSIONS,
        "idempotency_evidence": {
            "initial": initial,
            "replay": replay,
            "same_effect": same_effect,
            "same_response": same_response,
            "exactly_once_claimed": False,
            "payload_conflict": conflict,
            "other_tenant": other_tenant,
        },
        "compatibility_cases": compatibility,
        "offline_client_trace": offline_client_trace,
        "problem_contract": {
            "rfc_standard_members": [
                "type",
                "title",
                "status",
                "detail",
                "instance",
            ],
            "rfc_members_are_optional": True,
            "required_by_this_api": ["type", "title", "status"],
            "optional_by_this_api": ["detail", "instance"],
        },
        "authorization_boundary": boundary,
        "state_transitions": service.transitions,
        "offline_window_days": 14,
        "external_network_used": False,
    }

print(json.dumps(main(), ensure_ascii=False, indent=2))
PY

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

• 誤診: 初回と再送でresponseが違うので冪等ではない。反証: 冪等性が比較するのは意図した効果である。時刻やtrace IDを含む表現は変わり得る。副作用回数と対象状態を観測し、同じresponseやexactly-onceを別の保証として扱う。
• 誤診: OpenAPIとJSON Schemaに通ったrequestなので正しく認可され、業務上も実行可能だ。反証: schemaは主に表現の構造を検証する。主体が別tenantなら認可で拒否し、状態遷移が不変条件を破るなら意味検証で拒否する。
• 失敗モード: idempotency keyだけ保存し、業務更新と別transactionにすると、crash境界で効果とkeyが不一致になる。
• 失敗モード: enum追加を常に安全とみなすと、未知値を網羅switchで拒否する旧clientや、既定値へ誤変換するclientを壊す。

知識チェック

1. 同じ冪等keyの二回目に、初回と異なるstatusまたはresponseを返してよい条件は何か。
2. schema validだが意味的にinvalidなrequestと、意味的にvalidだがunauthorizedなrequestを一つずつ示せ。
3. 任意field追加、必須field削除、enum拡張をsource、wire、意味の三面で評価せよ。
4. Problem Detailsのtype、title、status、detail、instanceをconsumerはどう使い分けるか。

出典と次の学習

HTTP method、status、表現の意味はRFC 9110を基準にする。interface記述はOpenAPI Specification v3.2.0、構造検証はJSON Schema Draft 2020-12のvalidation vocabulary、後方互換の変更分類はAIP-180、共通の失敗表現はRFC 9457へ照合する。Problem Detailsの標準memberはRFC上一律必須ではなく、必要な必須性はAPI profileで明示する。いずれも業務意味と認可の設計を自動化するものではない。

次はcore-08で、契約を実現するmoduleの依存方向を変更理由から決め、代替案と撤回条件をADRへ残す。API versionを増やす前に、変更を局所化できるarchitectureかを検査する。

実践ラボ

オフライン現場端末の再送と契約進化を検証する

提出成果物: 互換性、冪等性、失敗形式を含むAPI契約

1. v1とv2で共有する作業完了コマンド、状態遷移、認可境界を宣言する
2. tenant、認証済みprincipal、route、冪等keyをscopeとし、同じpayloadの再送と異なるpayload fingerprintの衝突を検証する
3. 注入した固定RFC 3339時計で、副作用回数が一のままresponse生成時刻が異なる証拠を記録する
4. 任意field追加、必須field削除、enum拡張をsource、wire、semanticの三軸で分類しoffline client traceへ結ぶ
5. RFC 9457の標準memberの任意性と、このAPI profileが必須にするmemberを分けたProblem Detailsを生成する
6. schema上正しい別tenantのrequestを認可で拒否し、外部networkなしで三つ以上の状態遷移を検査する

説明して理解を確かめる

5分で、HTTPの冪等methodと業務上の冪等key、同じ効果と同じresponse、at-least-once再送とexactly-once主張を分け、schema妥当性が業務意味と認可を保証しない理由を説明する。

アセスメント

1. 問い: 端末が作業完了POSTの応答を失い、同じkeyで再送した。初回と異なるresponse時刻を返したら冪等でないか。

   期待する証拠: 対象resourceの意図した効果、副作用回数、response表現の違い、保存済み結果、結果照会、exactly-once非保証の区別

2. 問い: v2でenum値を追加した。JSON Schemaを通る旧clientなら互換と断定できるか。

   期待する証拠: 旧clientの未知値処理、sourceとwireと意味の互換性、長期offline端末、段階的提供と観測

別問題へ転用する

長時間オフラインになる現場端末へ互換な再送・同期API契約を設計する

復習スケジュール

1. 1日後

   同じ効果だが同じresponseでない冪等再送の例を示す

2. 7日後

   schema妥当性だけでは検出できない業務違反と認可違反は何か

3. 30日後

   enum追加が意味的に破壊的となる旧clientの挙動を示す

4. 90日後

   同じ効果だが同じresponseでない冪等再送の例を示す

評価ルーブリック

出典

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

• RFC 9110: HTTP Semantics (standard)
• OpenAPI Specification v3.2.0 (standard)
• JSON Schema Validation: A Vocabulary for Structural Validation of JSON (standard)
• AIP-180: Backwards compatibility (primary)
• RFC 9457: Problem Details for HTTP APIs (standard)