50件の模擬バッチを、41件目で止めました。
問題は、もう一度実行するときです。最初の40件まで処理し直してよいのでしょうか。今回は、完了した処理を記録する方法と、「送ったか分からない処理」を止めておく方法を、Python標準ライブラリの sqlite3 だけで試しました。2026年10月30日、手元のPC（macOS）で、AI支援でコードを作成・実行した模擬実験の記録です。実在のAPIや業務データは使っていません。
困るのは「送ったか分からない」1件
今回の実験では、先に完了した40件分を記録からスキップできました。
難しいのは41件目です。外部へ通知を送った直後、手元に「完了」と書く前に止まると、手元の記録からは送ったのかどうかが分かりません。
「完了」だけを記録する方法では、この1件は記録なしに見えます。再実行すると未送信と判断して送るので、送信先がすでに受け付けていれば二重送信になります。
そこで、送信の前に「送信中」（IN_FLIGHT）と書くことにしました。再実行時にこの記録が残っていれば、少なくとも「送ったか分からない」ことは分かります。
記録する状態と、再実行時の扱い
状態は3つだけです。
| 記録 | 書くタイミング | 再実行したときの扱い |
|---|---|---|
IN_FLIGHT |
各処理の直前 | 前処理なら再実行。外部送信なら照合し、確認できなければ止める |
SUCCEEDED |
処理が終わった後 | スキップ |
FAILED |
例外で止まったとき | 前処理なら再実行。外部送信なら IN_FLIGHT と同じく照合する |
実装を見直したのは FAILED の扱いです。最初の版では、外部送信が例外で止まった行も、再実行時に自動で送り直していました。しかし、タイムアウトのように「送信先は受け付けたが、応答だけ届かなかった」場合も例外になります。例外が出たことは、送っていない証拠になりません。そこで、外部送信の FAILED も IN_FLIGHT と同じく「送ったか分からない」として扱うようにしました。
前処理（入力の検証や変換）は外部に影響しないので、どちらの状態でもそのまま再実行します。
もう1つ、入力のハッシュ（payload_digest）も記録しています。前回と入力が変わった行は、スキップも再送もせずに止めます。前回送った内容と今回の内容が違う場合、どちらを正とするかはプログラムでは決められないためです。
重要な処理だけ抜き出すと
全体のコードは末尾に置きました。判断の中心は次の2か所です。
送信前に記録する部分です。状態の書き換えは、1行ずつ with conn: のトランザクションで確定させています。
_mark(conn, key, "IN_FLIGHT", digest) # 送信の「前」に記録する
try:
receipt = send_external(item_id, digest)
_mark(conn, key, "SUCCEEDED", digest, receipt=receipt)
except Exception as exc:
# 例外メッセージにはトークンや宛先が含まれ得るため、型名だけを残す
_mark(conn, key, "FAILED", digest, error_type=type(exc).__name__)
raise
再実行時に、外部送信の行が SUCCEEDED 以外で残っていた場合の扱いです。
if step in SIDE_EFFECT_STEPS:
# IN_FLIGHT も FAILED も「送ったか分からない」。再送せず、まず照合する
receipt = verify_sent(job_id, item_id, step) if verify_sent else None
if receipt:
_mark(conn, key, "SUCCEEDED", digest, receipt=receipt)
continue
held.append(f"{item_id}:{step}:unknown")
break
verify_sent は、送信先に受付記録があるかを問い合わせる関数です。job_id・item_id・処理名の3つで照合します。item_id だけでは、別の実行や別の処理の受付と取り違えるおそれがあるためです。照合の手段がない、または確認できなかった場合は、その件を止めて人の判断に回します。
現在のコードで測り直した結果
50件・各3処理のバッチで、41件目の外部送信を一度だけ失敗させました。待ち時間は1件あたり7msの模擬値です（前処理1.5ms×2、送信4ms）。失敗のさせ方を2通り用意し、それぞれ5回ずつ実行しました。
- 受付前に失敗：送信先は受け付けていない
- 受付後、応答前に失敗：送信先は受け付けたが、手元には応答が届かない
比較対象は、記録を持たずに最初から全件を再実行する方法です。
| 全件を再実行 | 記録を使って再開 | |
|---|---|---|
| 再実行で処理した数 | 150 | 2回目：27（42〜50件目） 照合・確認後の3回目：受付前に失敗なら1、受付後なら0 |
| スキップした数 | 0 | 2回目：122（1〜40件目と、41件目の前処理2つ） |
| 二重送信 | 受付前に失敗：40件 受付後に失敗：41件 |
どちらも0件 |
| 41件目の扱い | 区別せず送り直す | 2回目で41件目だけを保留し、照合結果に応じて送る／送らない |
2回目の実行では、41件目だけを保留し、42〜50件目は処理を続けました。アイテム同士に依存がない前提です。受付前に失敗した場合は、送信先に記録がないことを確認してから3回目で1回だけ送りました。受付後に失敗した場合は、照合で受付記録が見つかったので、送らずに完了扱いにしました。どちらも、50件すべてがちょうど1回ずつ送信先に届いています。
所要時間も記録しましたが、これは模擬の待ち時間をほぼそのまま反映した値です。受付前失敗では、全件再実行の5回平均が432.44ms、記録を使った再開（2回目と3回目の合計）が95.47msでした。受付後失敗では、それぞれ434.36msと86.85msでした。この差は「処理する件数が減った」ことによるもので、実際のAPIの速度や、人が照合する時間は含んでいません。この記事で確かめたかったのは、時間よりも再実行で何が二重に起きるかです。
止まった1件をどう解くか
止まった行は、送信先の記録を確認してから解きます。完全版には、確認した結果を記録するための resolve_unknown_step を入れました。
| 確認した結果 | すること |
|---|---|
| 送信先に受付記録があった |
delivered=True と受付の識別子を渡し、完了扱いにする（送らない） |
| 送信先に記録がないことを確認できた |
delivered=False を渡して行を消し、次の実行で送る |
| 確認できない | 止めたままにする。送るかどうかは送信先の担当者と決める |
ここでいう「記録がないことを確認」は、単に検索結果が0件だったという意味ではありません。記録の反映待ちや処理途中の可能性もあります。送信先の仕様に従って未受付を確定できなければ、delivered=False で解除せず保留します。
避けたい操作が2つあります。
1つは、確認しないまま行を消すことです。消した行は「未実行」に見えるので、次の実行で送られます。送信先が受け付けていれば二重送信です。
もう1つは、新しい job_id で流し直すことです。新しい job_id には過去の記録が1件もないので、完了済みの40件も含めて全件が送られます。復旧のつもりで使うと、記録を持たない方法と同じ結果になります。新しい job_id は、全件を改めて送ると決めたときだけ使います。
送信先が冪等キー（同じキーの要求を1件として扱う仕組み）に対応しているなら、job_id・item_id・処理名から作ったキーを付けて送ると、照合できない場合の安全網になります。対応の有無や保持期間はサービスごとに違い、今回は検証していません。
この方法が合わない場面
- 1件の処理が非常に軽い場合：処理ごとに2回書き込むので、マイクロ秒単位の処理では書き込みの方が重くなります。外部送信の前後だけ記録する、といった粒度の調整が必要です。
-
電源断やOS自体の障害まで扱う場合：サンプルは
WALとsynchronous=NORMALを使っています。この設定では、電源断などでコミット済みの記録が失われる可能性があります。今回はPythonの例外と中断状態を検証したもので、停電や実プロセスの強制終了試験はしていません。外部送信前の記録を電源断からも守る用途では、同期設定と実行環境を別に検証します（SQLite公式）。 - 複数のプロセスが同時に同じジョブを処理する場合：今回の実装は1プロセスで順に処理する前提です。取り合いを防ぐ仕組みは入れていません。
- ジャーナルをネットワーク上やクラウド同期フォルダに置く場合：SQLiteの公式資料は、WALモードがネットワークファイルシステム上では動作しないことを明記しています。ローカルディスクに置きます。
実行方法と完全版コード
Python 3.10以降、標準ライブラリのみで動きます。
python3 resumable_runner.py --scenario both --trials 5 # 2通りの失敗で模擬実験
resumable_runner.py（完全版）
#!/usr/bin/env python3
"""途中で止まったバッチを、送信済みの処理を繰り返さずに再開するための最小構成（標準ライブラリのみ）。
状態:
- IN_FLIGHT : 外部送信の直前に記録。応答前に止まると、この状態のまま残る
- SUCCEEDED : 応答を受け取って完了
- FAILED : 例外で止まった
再開時の扱い:
- SUCCEEDED（入力が同じ） → スキップ
- 前処理の FAILED / IN_FLIGHT → 再実行（外部への副作用がないため）
- 外部送信の FAILED / IN_FLIGHT → 「送ったか分からない」。照合で受付を確認できれば SUCCEEDED、
確認できなければ止めて人が判断する
- 入力が前回と違う行 → 止めて人が判断する
"""
from __future__ import annotations
import argparse
import hashlib
import json
import sqlite3
import tempfile
import time
from pathlib import Path
from typing import Callable
STEPS: tuple[str, ...] = ("validate", "transform", "notify_external")
SIDE_EFFECT_STEPS: frozenset[str] = frozenset({"notify_external"})
Verify = Callable[[str, str, str], "str | None"]
def init_db(db_path: Path) -> sqlite3.Connection:
"""ジャーナルを初期化する。保存先はローカルディスクに限る。"""
conn = sqlite3.connect(str(db_path))
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("PRAGMA synchronous=NORMAL;")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS step_journal (
job_id TEXT NOT NULL,
item_id TEXT NOT NULL,
step_name TEXT NOT NULL,
status TEXT NOT NULL CHECK(status IN ('IN_FLIGHT', 'SUCCEEDED', 'FAILED')),
payload_digest TEXT NOT NULL,
receipt TEXT,
error_type TEXT,
updated_at REAL NOT NULL,
PRIMARY KEY (job_id, item_id, step_name)
);
"""
)
conn.commit()
return conn
def _digest(item: dict[str, str]) -> str:
return hashlib.sha256(json.dumps(item, sort_keys=True).encode("utf-8")).hexdigest()
def _mark(conn: sqlite3.Connection, key: tuple[str, str, str], status: str, digest: str,
receipt: str | None = None, error_type: str | None = None) -> None:
"""1行の状態を1トランザクションで書き換える。"""
with conn:
conn.execute(
"""
INSERT INTO step_journal (job_id, item_id, step_name, status, payload_digest,
receipt, error_type, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(job_id, item_id, step_name) DO UPDATE SET
status = excluded.status,
payload_digest = excluded.payload_digest,
receipt = excluded.receipt,
error_type = excluded.error_type,
updated_at = excluded.updated_at
""",
(*key, status, digest, receipt, error_type, time.time()),
)
def run_resumable_batch(
conn: sqlite3.Connection,
job_id: str,
items: list[dict[str, str]],
send_external: Callable[[str, str], str],
verify_sent: Verify | None = None,
local_step: Callable[[str, dict[str, str]], None] | None = None,
) -> dict[str, object]:
"""完了済みを飛ばし、送ったか分からない外部送信は照合できるまで止める。"""
if not job_id.strip():
raise ValueError("job_id は空にできません")
executed = skipped = 0
held: list[str] = []
for item in items:
item_id = item.get("id", "").strip()
if not item_id:
raise ValueError("item.id が空です")
digest = _digest(item)
for step in STEPS:
key = (job_id, item_id, step)
row = conn.execute(
"SELECT status, payload_digest FROM step_journal "
"WHERE job_id = ? AND item_id = ? AND step_name = ?",
key,
).fetchone()
if row is not None:
status, stored_digest = row
if stored_digest != digest:
held.append(f"{item_id}:{step}:input_changed")
break
if status == "SUCCEEDED":
skipped += 1
continue
if step in SIDE_EFFECT_STEPS:
# IN_FLIGHT も FAILED も「送ったか分からない」。再送せず、まず照合する
receipt = verify_sent(*key) if verify_sent else None
if receipt:
_mark(conn, key, "SUCCEEDED", digest, receipt=receipt)
skipped += 1
continue
held.append(f"{item_id}:{step}:unknown")
break
# 前処理の FAILED / IN_FLIGHT は外部へ影響しないので、そのまま再実行する
_mark(conn, key, "IN_FLIGHT", digest) # 送信の「前」に記録する
executed += 1
try:
if step in SIDE_EFFECT_STEPS:
receipt = send_external(item_id, digest)
else:
if local_step:
local_step(step, item)
receipt = None
_mark(conn, key, "SUCCEEDED", digest, receipt=receipt)
except Exception as exc:
# 例外メッセージにはトークンや宛先が含まれ得るため、型名だけを残す
_mark(conn, key, "FAILED", digest, error_type=type(exc).__name__)
raise
return {
"status": "needs_review" if held else "completed",
"executed_steps": executed,
"skipped_steps": skipped,
"held": held,
}
def resolve_unknown_step(
conn: sqlite3.Connection,
job_id: str,
item_id: str,
step_name: str,
*,
delivered: bool,
receipt: str | None = None,
) -> None:
"""送信先の記録を人が確認した後に呼ぶ。確認していない状態では呼ばない。
delivered=True : 受付済みだった → SUCCEEDED にして、以後は再送しない
delivered=False : 送信先に記録がないことを確認した → 行を消し、次回の実行で送る
"""
key = (job_id, item_id, step_name)
row = conn.execute(
"SELECT status FROM step_journal WHERE job_id = ? AND item_id = ? AND step_name = ?", key
).fetchone()
if row is None or row[0] == "SUCCEEDED":
raise ValueError("未解決の行ではありません")
with conn:
if delivered:
if not receipt:
raise ValueError("受付済みと判断する根拠（receipt）が必要です")
conn.execute(
"UPDATE step_journal SET status = 'SUCCEEDED', receipt = ?, error_type = NULL, "
"updated_at = ? WHERE job_id = ? AND item_id = ? AND step_name = ?",
(receipt, time.time(), *key),
)
else:
conn.execute(
"DELETE FROM step_journal WHERE job_id = ? AND item_id = ? AND step_name = ?", key
)
# ---------------------------------------------------------------- 模擬実験
class TransientError(RuntimeError):
"""模擬の一時エラー。"""
class MockReceiver:
"""模擬の送信先。受付記録を持ち、照合に答える。"""
def __init__(self, latency_sec: float, fail_item: str | None, fail_after_accept: bool, job_id: str = "job-1") -> None:
self.latency_sec = latency_sec
self.fail_item = fail_item
self.fail_after_accept = fail_after_accept
self.accepted: dict[str, int] = {}
self.job_id = job_id
self.receipts: dict[tuple[str, str, str], str] = {}
def send(self, item_id: str, digest: str) -> str:
time.sleep(self.latency_sec)
if item_id == self.fail_item and not self.fail_after_accept:
raise TransientError("受付前に失敗")
self.accepted[item_id] = self.accepted.get(item_id, 0) + 1
self.receipts[(self.job_id, item_id, "notify_external")] = f"rcpt-{item_id}-{digest[:8]}"
if item_id == self.fail_item and self.fail_after_accept:
raise TransientError("受付後、応答前に失敗")
return f"rcpt-{item_id}-{digest[:8]}"
def lookup(self, job_id: str, item_id: str, step_name: str) -> str | None:
return self.receipts.get((job_id, item_id, step_name))
@property
def duplicates(self) -> int:
return sum(n - 1 for n in self.accepted.values() if n > 1)
def _naive(items: list[dict[str, str]], rx: MockReceiver, local_sec: float) -> int:
steps = 0
for item in items:
time.sleep(local_sec * 2)
steps += 3
rx.send(item["id"], _digest(item))
return steps
def run_experiment(scenario: str, trials: int = 5, total: int = 50, fail_at: int = 41) -> dict[str, object]:
"""scenario: before_accept（受付前に失敗）/ after_accept（受付後、応答前に失敗）"""
if scenario not in ("before_accept", "after_accept"):
raise ValueError("scenario が不正です")
after = scenario == "after_accept"
items = [{"id": f"item-{i:03d}", "body": f"payload-{i}"} for i in range(1, total + 1)]
fail_item = f"item-{fail_at:03d}"
local_sec, send_sec = 0.0015, 0.004
naive_ms: list[float] = []
journal_ms: list[float] = []
naive_steps = naive_dup = 0
journal_record: dict[str, object] = {}
trial_records: list[dict[str, object]] = []
for _ in range(trials):
rx = MockReceiver(send_sec, fail_item, after)
try:
_naive(items, rx, local_sec)
except TransientError:
pass
rx.fail_item = None
t0 = time.perf_counter()
naive_steps = _naive(items, rx, local_sec)
naive_ms.append((time.perf_counter() - t0) * 1000)
naive_dup = rx.duplicates
with tempfile.TemporaryDirectory() as tmp:
conn = init_db(Path(tmp) / "journal.sqlite3")
rx = MockReceiver(send_sec, fail_item, after)
local = lambda _s, _i: time.sleep(local_sec) # noqa: E731
try:
run_resumable_batch(conn, "job-1", items, rx.send, rx.lookup, local)
except TransientError:
pass
rx.fail_item = None
t0 = time.perf_counter()
r2 = run_resumable_batch(conn, "job-1", items, rx.send, None, local) # 照合なしで再開
r3: dict[str, object] | None = None
if r2["status"] == "needs_review":
# 人が送信先の記録を確認した想定（模擬の送信先に問い合わせる）
found = rx.lookup("job-1", fail_item, "notify_external")
resolve_unknown_step(conn, "job-1", fail_item, "notify_external",
delivered=bool(found), receipt=found)
r3 = run_resumable_batch(conn, "job-1", items, rx.send, None, local)
journal_ms.append((time.perf_counter() - t0) * 1000)
journal_record = {
"second_run": {k: r2[k] for k in ("status", "executed_steps", "skipped_steps", "held")},
"after_resolve_run": (
{k: r3[k] for k in ("status", "executed_steps", "skipped_steps")} if r3 else None
),
"duplicates": rx.duplicates,
"sent_once_items": sum(1 for n in rx.accepted.values() if n == 1),
}
if rx.duplicates != 0 or journal_record["sent_once_items"] != total:
raise RuntimeError("再開結果が期待と異なります")
trial_records.append(journal_record)
conn.close()
avg = lambda xs: round(sum(xs) / len(xs), 2) # noqa: E731
return {
"scenario": scenario,
"conditions": {"trials": trials, "items": total, "fail_at": fail_item,
"simulated_wait_ms_per_item": (local_sec * 2 + send_sec) * 1000},
"rerun_all": {"steps": naive_steps, "duplicates": naive_dup,
"ms": [round(x, 2) for x in naive_ms], "avg_ms": avg(naive_ms)},
"trial_outcomes": trial_records,
"journal": {**journal_record, "ms": [round(x, 2) for x in journal_ms], "avg_ms": avg(journal_ms)},
}
def main() -> None:
ap = argparse.ArgumentParser(description="途中再開の模擬実験")
ap.add_argument("--scenario", choices=["before_accept", "after_accept", "both"], default="both")
ap.add_argument("--trials", type=int, default=5)
a = ap.parse_args()
if a.trials < 1:
ap.error("--trials は1以上")
names = ["before_accept", "after_accept"] if a.scenario == "both" else [a.scenario]
print(json.dumps([run_experiment(n, a.trials) for n in names], ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
確認できたことと、確認していないこと
| 内容 | 根拠 | 確認していない範囲 |
|---|---|---|
| 再開時のスキップ数・停止・二重送信0件 | 手元の模擬実験（2通り×5回）とテスト8件、2026-10-30 | 実際のAPI、件数が数万件以上の場合、複数プロセスでの同時実行 |
| 照合による完了扱い | 単一ジョブの模擬受付記録で照合。別ジョブ・別処理を取り違えないテストも実施 | 照合手段のない送信先での運用 |
with conn: によるトランザクションの確定 |
Python公式資料（2026-10-30確認） | ― |
| WALモードとネットワークファイルシステムの制約 | SQLite公式資料（2026-10-30確認） | ― |
参考リンク
- SQLite: Write-Ahead Logging（2026-10-30確認）
https://www.sqlite.org/wal.html - SQLite: Transaction（2026-10-30確認）
https://www.sqlite.org/lang_transaction.html - Python
sqlite3：接続をコンテキストマネージャーとして使う（2026-10-30確認）
https://docs.python.org/ja/3/library/sqlite3.html#how-to-use-the-connection-context-manager
この記事を書いた人✏️@YushiYamamoto
個人事業ITPRODX.comの山本勇志です。PHPサイトの改修・保守、Web・CMS開発の担当経験を基に、既存システムの切り分けや設計判断を発信しています。
古いホームページやPHPサイトの調査・限定改修は、対応できる範囲を確認のうえご相談を承ります。