⟐ Claude Code/2026-08-31中級

無音で半分になっていた定期実行と、欠落を捕まえる突き合わせ

1日2回動いているはずのバッチが、実際には1回しか起動していませんでした。エラーは1件も出ていません。期待起動回数と実行記録を突き合わせて、無音の欠落を検出する仕組みを作ります。

Claude Code²⁴¹ スケジュール実行⁴ cron³ 運用¹⁹ 自動化⁸³

✦ プレミアム記事

画像アセットの分類バッチが、1日2回動く設定になっていました。朝と夕方、1回あたり数百枚を処理する想定です。

処理量が想定の 50% ほどしか進んでいないことに気づいたのは、素材の滞留を見たときでした。エラーログは空です。失敗の通知も届いていません。動いた回のログはすべて正常終了で、内容にも異常はありませんでした。

夕方の回が、そもそも起動していなかったのです。

起動しなかった回は、失敗しません。失敗しないので、失敗を監視する仕組みには何も映りません。個人開発で自動処理を増やすほど、この空白は広がります。以下は、その空白を数えられるようにするまでの記録です。

書式は正しかった

設定はこう書いていました。

30 4,16 * * *

crontab の書式としては、これは1日2回です。念のため、自分で展開して数えることにしました。cron 式を受け取って、期間内に起動が期待される時刻を列挙するだけの小さなスクリプトです。

#!/usr/bin/env python3
"""cron 式を展開して、期間内の「起動が期待される時刻」を列挙する。"""
from datetime import datetime, timedelta
 
 
def _field(spec: str, lo: int, hi: int) -> set:
    out = set()
    for part in spec.split(","):
        step = 1
        if "/" in part:
            part, s = part.split("/", 1)
            step = int(s)
        if part in ("*", ""):
            start, end = lo, hi
        elif "-" in part:
            a, b = part.split("-", 1)
            start, end = int(a), int(b)
        else:
            start = end = int(part)
            if step != 1:
                end = hi
        out |= set(range(start, end + 1, step))
    return {v for v in out if lo <= v <= hi}
 
 
def expected(cron: str, start: datetime, end: datetime):
    mi, ho, dom, mon, dow = cron.split()
    M, H = _field(mi, 0, 59), _field(ho, 0, 23)
    DOM, MON = _field(dom, 1, 31), _field(mon, 1, 12)
    DOW = {d % 7 for d in _field(dow, 0, 7)}      # 日曜は 0 でも 7 でも同じ
    dom_restricted = dom.strip() != "*"
    dow_restricted = dow.strip() != "*"
 
    t = start.replace(second=0, microsecond=0)
    while t <= end:
        if t.minute in M and t.hour in H and t.month in MON:
            d_ok = t.day in DOM
            w_ok = ((t.weekday() + 1) % 7) in DOW
            # 日と曜日を両方指定したときは OR。片方だけなら AND として扱う
            ok = (d_ok or w_ok) if (dom_restricted and dow_restricted) else (d_ok and w_ok)
            if ok:
                yield t
        t += timedelta(minutes=1)

7日間で数えると 14 回。書式の解釈は間違っていませんでした。つまり欠落は、書き方ではなく実行基盤の側で起きていたことになります。

ここが最初の分かれ道でした。設定ファイルを読み返す作業をいくら続けても、原因には届きません。設定が正しいことを確認したなら、次に見るべきは設定ではなく実績です。

書いておくと、日と曜日を両方指定したときの OR 解釈は、自分で展開してみるまで自信が持てない部分でした。0 9 1 * 1 は「毎月1日」と「毎週月曜」の両方で起動します。手元で 9 月を展開すると 09-01（火）と各月曜の計 5 回になり、想定どおりでした。

失敗ログに映らない障害という種類

自動処理の監視は、ほとんどが「失敗を捕まえる」形をしています。終了コードを見る、例外を通知する、出力を検査する。どれも実行されたプロセスがあって初めて成り立ちます。

起動が消えるタイプの障害には、この形が届きません。

障害の種類	失敗ログ	気づき方
処理中の例外	出る	通知で即座に分かる
異常終了・タイムアウト	出る（終了コード）	通知で分かる
条件未達で何もしなかった	出ない	意図した挙動なので問題ない
起動しなかった	出ない	成果物の量を数えるまで分からない

私自身、この表の最後の行を長いあいだ持っていませんでした。監視の設計を「何が起きたか」で組み立てていて、「何が起きなかったか」を見る枠がなかったからです。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦エラーが出ていない自動処理について、本当に動いたのかを記録から確かめられるようになります

✦自分のスケジュール設定を展開して期待起動回数を数え、実行記録とのずれを毎朝1コマンドで検出できるようになります

✦起動そのものが消えるタイプの障害を、被害が積み上がる前に見つけられるようになります

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

起動した時点で、先に記録を書く

対策の一段目は、実行記録の取り方を変えることです。処理が終わってから結果を書くのではなく、起動した瞬間に「起動した」という行を確定させ、終了時に status だけを埋める形にします。こうすると、途中で強制終了された回も記録に残ります。

#!/usr/bin/env bash
# 起動した時点で先に記録を書き、終了時に status を確定させる。
# 途中で kill されても「起動はした」行が残る。
set -uo pipefail
 
LOG_DIR="${LOG_DIR:-$HOME/logs/batch}"; mkdir -p "$LOG_DIR"
LOG="$LOG_DIR/run.jsonl"
STARTED="$(date -u +%Y-%m-%dT%H:%M:%S)"
TMP="$(mktemp)"
 
finish() {
  code=$?
  case "$code" in
    0)  status=ok ;;
    64) status=skipped ;;   # 実行条件を満たさず、自主的に何もしなかった回
    *)  status=failed ;;
  esac
  printf '{"started_at":"%s","status":"%s","exit":%d,"tail":%s}\n' \
    "$STARTED" "$status" "$code" \
    "$(tail -c 200 "$TMP" | python3 -c 'import json,sys;print(json.dumps(sys.stdin.read()))')" \
    >> "$LOG"
  rm -f "$TMP"
}
trap finish EXIT
 
"$@" > "$TMP" 2>&1

手元で四通り流した結果です。正常終了、条件未達での自主スキップ、異常終了、そして SIGTERM による強制終了。

{"started_at":"2026-08-31T03:10:38","status":"ok","exit":0,"tail":"done\n"}
{"started_at":"2026-08-31T03:10:38","status":"skipped","exit":64,"tail":"条件未達\n"}
{"started_at":"2026-08-31T03:10:38","status":"failed","exit":3,"tail":"boom\n"}
{"started_at":"2026-08-31T03:10:38","status":"failed","exit":143,"tail":""}

最後の行の exit: 143 が SIGTERM です。処理本体は何も書けずに終わっていますが、trap が外側にあるおかげで行そのものは残りました。ここが trap を使う理由です。処理の内側に記録を書くと、処理が死んだときに記録も一緒に死にます。

終了コード 64 を「自主的に何もしなかった」に割り当てているのは、後で数えるときに欠落と区別するためです。上限に達した、対象が0件だった、前回の処理がまだ終わっていない。どれも正常な判断ですが、記録が無いと「起動しなかった回」と見分けがつきません。何もしなかったことは、必ず何かを書いて示します。

期待回数と実行記録を突き合わせる

二段目が本題です。展開した期待時刻と、実際に残った記録を突き合わせます。

#!/usr/bin/env python3
"""期待される起動時刻と、実際に残った実行記録を突き合わせる。"""
import json
import sys
from datetime import datetime, timedelta
from pathlib import Path
 
from occ import expected
 
TOLERANCE = timedelta(minutes=20)   # 起動遅延の許容幅
 
 
def load_records(log_dir: Path):
    """1実行につき1行の JSONL。status は ok / skipped / failed のいずれか。"""
    recs = []
    for f in sorted(log_dir.glob("*.jsonl")):
        for line in f.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line:
                continue
            try:
                r = json.loads(line)
                recs.append((datetime.fromisoformat(r["started_at"]), r))
            except (ValueError, KeyError):
                # 壊れた行で全体を落とさない。ただし黙って捨てもしない
                print(f"  ! 壊れた行を無視: {f.name}: {line[:60]}", file=sys.stderr)
    return recs
 
 
def reconcile(cron, log_dir, since, until):
    recs = load_records(Path(log_dir))
    unmatched = list(recs)
    missing = []
    for want in expected(cron, since, until):
        hit = next((r for r in unmatched if abs(r[0] - want) <= TOLERANCE), None)
        if hit:
            unmatched.remove(hit)      # 1件の記録を2つの期待に使い回さない
        else:
            missing.append(want)
    return missing, unmatched, recs
 
 
if __name__ == "__main__":
    cron, log_dir = sys.argv[1], sys.argv[2]
    until = datetime.fromisoformat(sys.argv[3]) if len(sys.argv) > 3 else datetime.now()
    since = until - timedelta(days=7)
 
    missing, extra, recs = reconcile(cron, log_dir, since, until)
    by = {}
    for _, r in recs:
        by[r.get("status", "?")] = by.get(r.get("status", "?"), 0) + 1
 
    print(f"期待 {len(missing) + len(recs) - len(extra)} 回 / 記録 {len(recs)} 件 {by}")
    print(f"欠落 {len(missing)} 回")
    for m in missing[:5]:
        print("  -", m.strftime("%Y-%m-%d %H:%M"))
    if len(missing) > 5:
        print(f"  ... 他 {len(missing) - 5} 回")
    sys.exit(1 if missing else 0)

朝の回だけが残っている1週間分を食わせると、こう出ます。

期待 14 回 / 記録 7 件 {'ok': 6, 'skipped': 1}
欠落 7 回
  - 2026-08-25 16:30
  - 2026-08-26 16:30
  - 2026-08-27 16:30
  - 2026-08-28 16:30
  - 2026-08-29 16:30
  ... 他 2 回

欠落した時刻がすべて 16:30 に揃っています。ここまで出れば、原因の見当は設定ファイルではなく実行基盤にあると分かります。ログを1件ずつ読んでいたときには見えなかった形です。

実装で気をつけた点が二つあります。ひとつは、照合済みの記録を候補から取り除くこと。取り除かないと、遅延した1件が複数の期待時刻にマッチして、欠落が過少に出ます。もうひとつは許容幅です。20分は、起動の混雑や処理待ちを吸収しつつ、隣の枠まで届かない幅として選びました。枠の間隔が狭い設定では、間隔の半分未満に縮めることを推奨します。この場合は、遅延の常態化を別の指標として拾う設計にしておくと安全です。

手元の定期処理を点検する手順

すでに動いている処理に後付けする場合、次の順で進めると既存の記録を捨てずに済みます。

1. まず数えるだけの状態を作る

記録の形式は変えずに、期待起動回数だけを先に数えます。既存のログファイル名やタイムスタンプが使えるなら、load_records をその形式に合わせて差し替えるだけで突き合わせは動きます。ここで大きなずれが出れば、記録の作り直しを待たずに調査へ入れます。

2. 記録の書き込み位置を外へ出す

次に trap 方式へ切り替えます。既存の処理本体には手を入れず、外側から包む形にすると差分が小さく、切り戻しも容易です。私はこの順序を守らずに両方を同時に変えたことがあり、ずれの原因が旧記録の欠損なのか新しい包み方の不備なのか分からなくなりました。

3. 突き合わせを毎朝の入口に置く

最後に、突き合わせスクリプトを1日1回動かします。欠落が0件なら終了コード0で静かに終わり、1件でもあれば非0で終わるようにしてあるので、通知の条件は「このコマンドが失敗したら」の1本で足ります。監視対象を増やさずに、監視の種類を1つ増やせます。

突き合わせで見えるのは欠落だけではありません

同じ突き合わせから、性質の違うずれが三種類とれます。

ずれ	出方	疑うべきところ
欠落	期待にマッチする記録が無い	スケジューラ側の起動、設定の反映漏れ、ホストの停止
余剰	どの期待にも紐づかない記録が残る	二重登録、手動実行の混入、時刻の扱いのずれ
遅延の常態化	マッチはするが差が許容幅に張り付く	実行時間の伸び、同時刻に集中した枠の詰まり