再送を「もう一件の申請」にしない

モバイルで配送申請を送った直後に通信が切れると、利用者には登録できたか分かりません。ボタンを無効にするだけでは、再読み込みや通信の再試行を止められません。サーバー側で「同じ操作の再送」を区別する必要があります。

筆者は韓国向け配送代行サービスKimfastの関係者です。この記事は配送申請という題材を使った独立した学習用コードです。Kimfastの本番実装、障害事例、性能測定を公開するものではありません。顧客データも使いません。

先に決める3つの動作

入力	返す結果
まだ保存されていないキー	新しい申請ID
同じアカウント・同じキー・同じ内容	前回と同じ申請ID
同じアカウント・同じキー・違う内容	競合エラー

キーは送信を押すたびではなく、一つの申請操作について生成し、結果が確定するまで再利用します。新しい申請には新しいキーを使います。アカウントIDは認証済みセッションからサーバーが決めます。クライアントに自由に指定させません。

最小構成：SQLiteに申請とキーを同時に保存する

次を idempotency_demo.py として保存します。標準ライブラリだけで動きます。例では保存された行そのものを申請とみなし、外部APIや決済は呼びません。

"""Educational example only; no connection to Kimfast production."""
import hashlib
import json
import sqlite3
import uuid
from contextlib import closing


def initialize(path):
    with closing(sqlite3.connect(path)) as db:
        db.execute("""CREATE TABLE IF NOT EXISTS requests (
            account TEXT NOT NULL, request_key TEXT NOT NULL,
            payload_hash TEXT NOT NULL, application_id TEXT NOT NULL,
            PRIMARY KEY (account, request_key))""")
        db.commit()


def submit(path, account, request_key, payload):
    if not account or not request_key:
        raise ValueError("account and request_key are required")
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"),
                           ensure_ascii=False, allow_nan=False)
    digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    with closing(sqlite3.connect(path, isolation_level=None, timeout=5)) as db:
        db.execute("BEGIN IMMEDIATE")
        try:
            existing = db.execute(
                "SELECT payload_hash, application_id FROM requests "
                "WHERE account=? AND request_key=?",
                (account, request_key)).fetchone()
            if existing:
                if existing[0] != digest:
                    raise ValueError("same key used with a different payload")
                application_id, created = existing[1], False
            else:
                application_id, created = str(uuid.uuid4()), True
                db.execute("INSERT INTO requests VALUES (?, ?, ?, ?)",
                           (account, request_key, digest, application_id))
            db.commit()
            return application_id, created
        except Exception:
            db.rollback()
            raise

PRIMARY KEY (account, request_key) が一意性の境界です。BEGIN IMMEDIATE で書き込みトランザクションを開始し、確認と作成を同じトランザクションに入れます。この短い処理の中でネットワーク通信はしません。

JSONのキー順はそろえますが、これは業務上の同値性を完全に判定する仕組みではありません。たとえば 1 と 1.0、配列の順序は別扱いです。実サービスでは、入力の型・単位・必須項目を検証し、正規化した申請データを渡します。

まず再送と競合を試す

from idempotency_demo import initialize, submit

initialize("demo.db")
payload = {"origin": "AU", "destination": "KR", "parcels": 1}
first = submit("demo.db", "demo-account", "action-001", payload)
retry = submit("demo.db", "demo-account", "action-001", payload)
assert first[0] == retry[0]
assert first[1] is True and retry[1] is False

try:
    submit("demo.db", "demo-account", "action-001", {**payload, "parcels": 2})
except ValueError as error:
    print(error)  # same key used with a different payload

同じデータでも別のキーなら別の申請を作ります。内容が似ているだけで勝手に統合すると、利用者が意図した2回目の注文を消してしまうためです。

同時再送も確認する

テスト用の新しいDBで、別々の接続から同じ操作を8回呼びます。

from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from tempfile import TemporaryDirectory
import sqlite3

with TemporaryDirectory() as tmp:
    path = str(Path(tmp) / "concurrency.db")
    initialize(path)
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(
            lambda _: submit(path, "demo-account", "shared", payload),
            range(8),
        ))
    assert len({result[0] for result in results}) == 1
    assert sum(result[1] for result in results) == 1
    with sqlite3.connect(path) as db:
        assert db.execute("SELECT COUNT(*) FROM requests").fetchone()[0] == 1

2026年9月28日のローカル実行では、再送、JSONキー順、内容変更、アカウント分離、新規キー、空キー・NaN、同時再送の7つのunittestが通過しました。これは性能や本番可用性の保証ではありません。

本番に持ち込む前に残ること

認証・認可、CSRF対策、入力スキーマ、キー長とリクエストサイズの上限。
キーの保存期間と再利用ルール。期限後に再送された場合の扱い。
DBロックやタイムアウトを、成功・削除・未登録と即断しないこと。結果照会の導線を用意します。
決済や送り状APIなど外部処理との整合性。DBの冪等性だけでは外部処理の二重実行を防げません。必要に応じてoutbox、外部サービス側の冪等キー、照合処理を別に設計します。
申請IDの保存と実際の申請作成が別テーブルなら、両方を同じトランザクションで扱います。

UIでも「申請を受け付けた」と「発送が完了した」を分けて表示することが大切です。配送代行には申請、倉庫での確認、料金確認など異なる段階があります。題材の背景はKimfastのオーストラリア配送案内で確認できます。これは韓国向けサービスの案内であり、日本向け配送を提供するという意味ではありません。

参考資料・制作について

Python公式 sqlite3 ドキュメント：接続、トランザクション、パラメータを使ったSQL。
図は本記事の処理と実行結果を説明するために作成したものです。実際の倉庫や顧客の写真ではありません。
本文と例の作成にAIを使用しました。コードは上記の範囲でローカル実行しました。