自動化が途中で止まったら何を保存するか――チェックポイントと再開判断を分ける

Posted at 2026-11-04

2026-11-05、架空の処理を途中終了させ、再起動後に残る状態をローカルで確認しました。対象は、外部への書込みを含む小さな自動化を設計・保守するエンジニアです。

止まったステップ番号だけを保存しても、「その書込みは終わったのか」が分からないことがあります。保存したいのは処理位置に加えて、何が確定し、何がまだ不明かです。

再開できる状態と、照合が必要な状態を分ける

単一ワーカーの設計例として、状態を3つに絞ります。

保存状態	意味	再起動後の判断
`prepared`	入力を固定し、まだ送信準備の記録へ進んでいない	入力と処理版を確認して開始
`dispatching`	書込みを試みる直前の記録を確定済み。結果は未確定	自動再送を保留し、対象側と照合
`confirmed`	結果の参照IDと成功状態を同時に確定済み	同じ書込みを飛ばし、次の処理へ

dispatching は「送信済み」を意味しません。記録を確定した直後、実際に送信する前にも止まるからです。ここを正確に区別するため状態を増やしても、別システムへの書込みと手元の保存が一つのトランザクションにならなければ、どこかに結果不明の区間が残ります。

次の図は、この区間を隠さないための設計案です。中央の点線枠が、人による照合へ戻す境界です。

入力や処理版が違えば、図の先へ進めません。確定済みの結果も、新しい入力の成功として流用しない扱いにします。

状態DBを増やす前に、既存の記録で足りるかを見る

選択肢	選ぶ条件	足りないところ
最初から再実行	読取り中心、または繰返しの副作用が許容される	書込み済みの区間を復元できない
既存ワークフローの実行履歴	保存粒度・保持期間・再開仕様が要件に合う	対象側の結果まで確定したとは限らない
手動の照合記録	件数が少なく、担当者が確認できる	未処理の滞留と引継ぎを管理する必要
業務用チェックポイント	実行履歴とは別に、長く残す判断根拠が必要	状態更新・保持・復旧の実装と保守が増える

たとえばn8nには実行データの保存設定と削除機能があります。保存するデータや保持条件を確認せず、実行履歴を恒久的な業務台帳として使うのは避けます。公式の実行データ管理資料で設定を確認し、既存の仕組みで満たせるなら別DBは増やしません。

保存順序を変えずに、プロセスを途中終了させる

今回の検証は、Python 3.11.8 / SQLite 3.51.0 / macOSで行いました。state.db は進行状態、別接続の mock.db は架空の外部書込み先です。ネットワーク送信、課金API、実際の業務データは使っていません。Codexが作成・実行した個人用の検証で、顧客システムの復旧実績ではありません。

処理の順序は次のとおりです。

入力のSHA-256と処理版を確認する。
dispatching を保存し、commit() する。
模擬書込み先で1件を追加し、そちらの commit() を終える。
結果IDと confirmed を同じトランザクションで保存する。

Pythonのsqlite3公式資料に沿って確定点を明示しました。SQLiteのトランザクションで守る範囲と、模擬書込み先の確定は別です。SQLiteのatomic commitの説明を、外部APIまで含めた一括確定と解釈しないようにします。

指定した位置で子プロセスを os._exit(17) により終了し、親プロセスからDBを読み直して再起動しました。finally や通常の終了処理を通らないことはPython公式仕様で確認しています。電源断やディスク故障を再現した試験ではありません。

終了位置	再起動前の状態	模擬先の件数	再起動後の結果
送信準備の記録前	`prepared`	0	処理して `confirmed`、1件
送信準備の確定直後	`dispatching`	0	保留、0件のまま
模擬先の確定直後	`dispatching`	1	保留、1件のまま
成功状態を更新したが未commit	`dispatching`	1	保留、1件のまま
成功状態と結果IDのcommit直後	`confirmed`	1	同じ書込みを飛ばす、1件のまま
記録前に終了し、入力ハッシュを変更	`prepared`	0	不一致で保留、0件のまま

2行目と3行目は、手元の状態が同じでも模擬先の結果が違いました。 さらに4行目では、成功へのUPDATEを実行していてもcommit前だったため、再接続時には dispatching が残りました。メモリ上の進捗や成功ログだけを再開根拠にしない理由です。

全6ケースで保存状態・参照ID・模擬先件数をassertしています。ただし、保留後の照合と状態変更はこのコードでは実装していません。「保留すれば自動復旧できる」という結果ではありません。

再現コード：checkpoint.pyとして保存し、python3 checkpoint.pyで実行

標準ライブラリだけを使います。一時ディレクトリに架空のDBを作り、試験後に削除します。通常実行で6ケースのJSONが出力され、期待値と異なればassertが失敗します。

"""Synthetic crash/restart experiment; local SQLite files only."""
import hashlib
import json
import os
from pathlib import Path
import sqlite3
import subprocess
import sys
import tempfile

PAYLOAD = '{"record":"demo-01","operation":"mark"}'
DIGEST = hashlib.sha256(PAYLOAD.encode()).hexdigest()


def run_worker(folder, stop):
    db = sqlite3.connect(folder / "state.db")
    remote = sqlite3.connect(folder / "mock.db")
    state, digest, version = db.execute(
        "SELECT state, digest, version FROM jobs WHERE id=1"
    ).fetchone()
    if digest != DIGEST or version != "v1":
        print("hold: input/version mismatch", flush=True)
        return
    if state == "confirmed":
        print("advance: result already saved", flush=True)
        return
    if state == "dispatching":
        print("hold: reconcile before resume", flush=True)
        return
    if state != "prepared":
        raise ValueError("unknown state")
    # Only prepared may start a write. Commit the intent BEFORE that write.
    if stop == "before_intent":
        os._exit(17)
    db.execute("UPDATE jobs SET state='dispatching' WHERE id=1")
    db.commit()
    if stop == "after_intent":
        os._exit(17)
    remote.execute("INSERT INTO effects VALUES ('demo-01')")
    remote.commit()
    if stop == "after_effect":
        os._exit(17)
    db.execute("UPDATE jobs SET state='confirmed', receipt='mock-01' WHERE id=1")
    if stop == "before_commit":
        os._exit(17)
    db.commit()
    if stop == "after_commit":
        os._exit(17)
    print("confirmed", flush=True)


def inspect(folder):
    with sqlite3.connect(folder / "state.db") as db:
        state, receipt = db.execute("SELECT state, receipt FROM jobs").fetchone()
    with sqlite3.connect(folder / "mock.db") as db:
        count = db.execute("SELECT count(*) FROM effects").fetchone()[0]
    return [state, receipt, count]


def experiment():
    expected = {
        "before_intent": (["prepared", None, 0], ["confirmed", "mock-01", 1]),
        "after_intent": (["dispatching", None, 0], ["dispatching", None, 0]),
        "after_effect": (["dispatching", None, 1], ["dispatching", None, 1]),
        "before_commit": (["dispatching", None, 1], ["dispatching", None, 1]),
        "after_commit": (["confirmed", "mock-01", 1], ["confirmed", "mock-01", 1]),
        "changed_input": (["prepared", None, 0], ["prepared", None, 0]),
    }
    results = []
    with tempfile.TemporaryDirectory() as tmp:
        for stop, (before, after) in expected.items():
            folder = Path(tmp) / stop
            folder.mkdir()
            with sqlite3.connect(folder / "state.db") as db:
                db.execute("CREATE TABLE jobs(id INTEGER PRIMARY KEY, state TEXT, digest TEXT, version TEXT, receipt TEXT)")
                db.execute("INSERT INTO jobs VALUES (1,'prepared',?,'v1',NULL)", (DIGEST,))
            with sqlite3.connect(folder / "mock.db") as db:
                db.execute("CREATE TABLE effects(record TEXT)")
            child = [sys.executable, __file__, str(folder)]
            crashed = subprocess.run(child + ["before_intent" if stop == "changed_input" else stop], check=False)
            assert crashed.returncode == 17
            first = inspect(folder)
            assert first == before, (stop, first, before)
            if stop == "changed_input":
                with sqlite3.connect(folder / "state.db") as db:
                    db.execute("UPDATE jobs SET digest='changed' WHERE id=1")
            resumed = subprocess.run(child + ["none"], capture_output=True, text=True, check=True)
            final = inspect(folder)
            assert final == after, (stop, final, after)
            if stop == "changed_input":
                assert "mismatch" in resumed.stdout
            results.append({"stop": stop, "after_crash": first, "resume": resumed.stdout.strip(), "after_resume": final})
    print(json.dumps({"python": sys.version.split()[0], "sqlite": sqlite3.sqlite_version, "cases": results}, indent=2))


if __name__ == "__main__":
    if len(sys.argv) == 3:
        run_worker(Path(sys.argv[1]), sys.argv[2])
    else:
        experiment()

引継ぎに必要な項目は、入力全文ではない

このfixtureは状態・入力ハッシュ・処理版・結果IDに絞っています。実運用での追加項目は、次の用途から決めるのがよいと考えます。以下は未実装の設計案です。

項目	残す理由
業務操作ID・実行ID・ステップ	同じ業務を追い、試行を区別する
入力参照・版・ハッシュ	再開時に同じ入力か確認する
処理版・承認対象版	新しいロジックや改稿版への承認流用を防ぐ
試行開始時刻・期限・保留理由	長時間止まった処理を見つける
対象側の参照ID・確認時刻	照合の証拠と次の確認先を残す
次の操作・担当者・判断履歴	別の担当者が再開条件を理解する

秘密鍵や入力全文を進行ログへ複製しません。参照先にはアクセス制御と保持期間が必要です。ハッシュも匿名化の代わりにはならず、低エントロピーな値から元の情報が推測される場合があります。

保留した処理は、担当者が対象側の結果と入力を照合します。一致する成功を確認できれば証拠を付けて確定へ進め、未処理だと信頼できる方法で確認できれば再試行を判断します。どちらとも言えないときは保留を続けます。古い承認の期限や対象版が違う場合も、そのまま実行しません。

この例を採用しない条件

このコードは単一ワーカー専用です。複数ワーカーが同時に prepared を読めば、両方が書込みへ進む競合を防げません。並行実行が必要なら、排他的な取得、条件付き状態更新、所有者や期限切れ時の扱いまで検証する必要があります。fixtureにはそれらを実装していません。

また、confirmed の後に別の副作用があるなら、その操作にもチェックポイントが要ります。今回の「次へ進む」という出力は、後続処理の成功を示していません。ディスク容量不足、DB破損、バックアップからの復元、認証失効、実APIの照合方法、長期運用での保留件数は未検証です。

小規模な単一ワーカーなら、まずこの3状態で止まり方を説明できるかを試す。そのうえで、既存ワークフローの履歴だけで要件を満たせるなら追加実装は見送る、という判断を勧めます。運用を引き受ける前には、保留の確認担当と確認期限を決め、結果不明のケースから再開試験を始めたいところです。保留が増える分の照合作業と保存データの保守費用も必要で、工数削減の効果は今回測っていません。

参考資料

確認日：2026-11-05。公式仕様、今回のローカル実測、未実装の設計案を分けています。

この記事を書いた人✏️@YushiYamamoto
個人事業ITPRODX.comの山本勇志です。PHPサイトの改修・保守、Web・CMS開発の担当経験を基に、既存システムの切り分けや設計判断を発信しています。
古いホームページやPHPサイトの調査・限定改修は、対応できる範囲を確認のうえご相談を承ります。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up