第172回では複数のPython CLIを順番に実行し、第173回ではワークフローの二重起動を排他ロックで防ぎました。それでも、一時的な資源不足や接続失敗で子コマンドが止まることはあります。
ここで、すべての失敗を同じように再実行するのは危険です。外部サービスへの登録が完了した直後に応答だけ受け取れなかった場合、再実行によって同じ登録や通知が繰り返される可能性があります。
今回は、子コマンド側で「一時失敗」と定義した終了コードだけを、上限回数と待機時間を決めて再試行するラッパーを作ります。対象はPython 3.10以上のLinuxです。外部ライブラリは使いません。
排他制御と再試行は解決する問題が異なる
| 仕組み | 防ぐ問題 | 防げない問題 |
|---|---|---|
| 順次実行 | 処理順の乱れ、失敗後の後続実行 | 同じ処理の二重起動 |
| 排他ロック | 同じホスト上での同時実行 | 時間をずらした重複処理 |
| 今回の再試行 | 明示された一時失敗からの再実行 | 副作用の重複、処理の冪等性 |
サンプルでは終了コード75を一時失敗として扱います。ただし、75がすべてのCLIで共通して一時失敗を意味するわけではありません。再試行するコードは、子コマンドとの入出力契約で先に決めます。
今回の実行契約
| 項目 | 仕様 |
|---|---|
| 子コマンド | --より後ろの引数配列を、シェルを介さず実行する |
| 再試行対象 | --retry-codeで明示した1～124の終了コードだけ |
| 実行回数 | 初回を含め1～10回。既定値は3回 |
| 待機 | 初期待機時間に係数を掛け、上限値で打ち止めにする |
| 標準入力 | DEVNULLへ接続し、対話入力を待たせない |
| 子の出力 | 試行ごとに一時ファイルへ保存し、実行中は表示しない |
| 最終成功 | 成功した試行の標準出力と標準エラー出力だけを返す |
| 最終失敗 | 標準出力は捨て、各失敗のメタデータと標準エラー出力を返す |
| 環境変数 | 試行番号と最大回数を子へ渡す |
途中の失敗後に成功した場合、以前の失敗ログは残しません。最終成功時の出力を後続処理へそのまま渡すための仕様です。全試行の監査記録が必要な業務では、この実装とは別に、アクセス制御したログ保存先を用意してください。
再試行ラッパーを作る
次のコードをBOMなしUTF-8のretry_command.pyとして保存します。
#!/usr/bin/env python3
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
def bounded_int(label, low, high):
def parse(text):
try:
value = int(text, 10)
except ValueError as exc:
raise argparse.ArgumentTypeError(
f"{label}は整数で指定してください"
) from exc
if not low <= value <= high:
raise argparse.ArgumentTypeError(
f"{label}は{low}以上{high}以下で指定してください"
)
return value
return parse
def parse_args():
parser = argparse.ArgumentParser(
description="指定した終了コードだけを再試行する"
)
parser.add_argument(
"--retry-code",
action="append",
required=True,
type=bounded_int("retry-code", 1, 124),
)
parser.add_argument(
"--max-attempts",
type=bounded_int("max-attempts", 1, 10),
default=3,
)
parser.add_argument(
"--initial-delay-ms",
type=bounded_int("initial-delay-ms", 0, 60_000),
default=1_000,
)
parser.add_argument(
"--backoff-factor",
type=bounded_int("backoff-factor", 1, 10),
default=2,
)
parser.add_argument(
"--max-delay-ms",
type=bounded_int("max-delay-ms", 0, 60_000),
default=30_000,
)
parser.add_argument("command", nargs=argparse.REMAINDER)
args = parser.parse_args()
if not args.command or args.command[0] != "--":
parser.error("子コマンドの前に -- を指定してください")
args.command = args.command[1:]
if not args.command:
parser.error("-- の後に子コマンドが必要です")
if args.initial_delay_ms > args.max_delay_ms:
parser.error("initial-delay-msはmax-delay-ms以下にしてください")
args.retry_codes = frozenset(args.retry_code)
return args
def append_record(history, payload, child_stderr=None):
line = json.dumps(
payload, ensure_ascii=False, separators=(",", ":")
)
history.write((line + "\n").encode("utf-8"))
if child_stderr is None:
return
child_stderr.seek(0, os.SEEK_END)
if child_stderr.tell() == 0:
return
child_stderr.seek(0)
shutil.copyfileobj(child_stderr, history, length=1024 * 1024)
child_stderr.seek(-1, os.SEEK_END)
if child_stderr.read(1) != b"\n":
history.write(b"\n")
def replay(source, target):
source.seek(0)
shutil.copyfileobj(source, target, length=1024 * 1024)
target.flush()
def normalized_status(returncode):
if returncode >= 0:
return min(returncode, 255)
return min(128 + (-returncode), 255)
def run(args):
delay_ms = args.initial_delay_ms
with tempfile.TemporaryFile() as history:
for attempt in range(1, args.max_attempts + 1):
environment = os.environ.copy()
environment["RETRY_WRAPPER_ATTEMPT"] = str(attempt)
environment["RETRY_WRAPPER_MAX_ATTEMPTS"] = str(
args.max_attempts
)
with (
tempfile.TemporaryFile() as child_stdout,
tempfile.TemporaryFile() as child_stderr,
):
try:
completed = subprocess.run(
args.command,
stdin=subprocess.DEVNULL,
stdout=child_stdout,
stderr=child_stderr,
shell=False,
check=False,
env=environment,
)
except OSError as exc:
if isinstance(exc, FileNotFoundError):
kind, status = "not_found", 127
elif isinstance(exc, PermissionError):
kind, status = "permission_denied", 126
else:
kind, status = "os_error", 125
append_record(
history,
{
"source": "retry_wrapper",
"event": "launch_error",
"attempt": attempt,
"max_attempts": args.max_attempts,
"kind": kind,
"retrying": False,
},
)
replay(history, sys.stderr.buffer)
return status
returncode = completed.returncode
if returncode == 0:
replay(child_stdout, sys.stdout.buffer)
replay(child_stderr, sys.stderr.buffer)
return 0
retrying = (
returncode in args.retry_codes
and attempt < args.max_attempts
)
event = {
"source": "retry_wrapper",
"event": "attempt_failed",
"attempt": attempt,
"max_attempts": args.max_attempts,
"returncode": returncode,
"retrying": retrying,
}
if retrying:
event["next_delay_ms"] = delay_ms
append_record(history, event, child_stderr)
if not retrying:
replay(history, sys.stderr.buffer)
return normalized_status(returncode)
time.sleep(delay_ms / 1000)
delay_ms = min(
args.max_delay_ms,
delay_ms * args.backoff_factor,
)
raise AssertionError("到達しない分岐です")
def wrapper_event(event, kind):
payload = {
"source": "retry_wrapper",
"event": event,
"kind": kind,
}
print(
json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
file=sys.stderr,
flush=True,
)
def main():
args = parse_args()
try:
return run(args)
except KeyboardInterrupt:
wrapper_event("interrupted", "keyboard_interrupt")
return 130
except BrokenPipeError:
try:
sys.stdout.close()
except BrokenPipeError:
pass
return 141
except OSError as exc:
wrapper_event("wrapper_error", type(exc).__name__)
return 125
if __name__ == "__main__":
raise SystemExit(main())
処理の順序を確認する
- 子コマンドを一時ファイルへ接続して起動する
- 終了コード0なら、その試行の出力だけを呼び出し元へ返す
- 非ゼロなら、終了コードが再試行リストにあるか確認する
- 再試行する場合は待機し、次の待機時間を係数倍する
- 対象外の終了コードまたは上限到達なら、失敗履歴を標準エラー出力へ返す
- シグナルで終了した子の負の戻り値は、このラッパーの契約として
128 + シグナル番号へ変換する
Pythonの公式文書でも、POSIXでシグナル終了した子プロセスの戻り値は負のシグナル番号になると説明されています。詳細はPython 3.10のsubprocess文書で確認できます。
再試行のたびに、子コマンドへRETRY_WRAPPER_ATTEMPTとRETRY_WRAPPER_MAX_ATTEMPTSを渡します。試行番号は診断には使えますが、処理の重複を防ぐ識別子ではありません。副作用のある処理には、全試行で同じ値を使う業務上の処理IDが別に必要です。
3回目に成功する子コマンドで確認する
次のコードをdemo_job.pyとして保存します。実運用の一時障害を再現するものではなく、試行番号に応じて終了コードを変える動作確認用コードです。
#!/usr/bin/env python3
import json
import os
import sys
attempt = int(os.environ["RETRY_WRAPPER_ATTEMPT"])
if attempt < 3:
print(f"一時失敗: attempt={attempt}", file=sys.stderr)
raise SystemExit(75)
print(
json.dumps(
{"status": "ok", "attempt": attempt},
ensure_ascii=False,
separators=(",", ":"),
)
)
同じディレクトリで次の順に実行します。
python3 retry_command.py --retry-code 75 --max-attempts 3 --initial-delay-ms 10 --backoff-factor 2 --max-delay-ms 20 -- python3 demo_job.py > result.json 2> retry.err
status=$?
printf 'exit=%s\n' "$status"
cat result.json
wc -c < retry.err
結果は次のとおりです。
exit=0
{"status":"ok","attempt":3}
0
1回目の後に10ミリ秒、2回目の後に20ミリ秒待機します。3回目が成功すると、それ以前の標準出力、標準エラー出力、失敗イベントは破棄されます。
恒久エラーは1回で止める
終了コード65を再試行リストへ含めていないため、次のコマンドは1回で停止します。終了コードを確認する行は、コマンドの直後に実行してください。
python3 retry_command.py --retry-code 75 --initial-delay-ms 0 --max-delay-ms 0 -- python3 -c 'import sys; print("設定エラー", file=sys.stderr); raise SystemExit(65)' > permanent.out 2> permanent.err
status=$?
printf 'exit=%s\n' "$status"
cat permanent.err
permanent.outは空です。終了コードと標準エラー出力は次のようになります。
exit=65
{"source":"retry_wrapper","event":"attempt_failed","attempt":1,"max_attempts":3,"returncode":65,"retrying":false}
設定エラー
終了コードを確認する
| 終了経路 | ラッパーの終了コード | 再試行 |
|---|---|---|
| 子が成功 | 0 | 終了する |
| 対象外の子終了コード65 | 65 | しない |
| 対象コード75で上限到達 | 75 | それ以上しない |
| 実行ファイルが存在しない | 127 | しない |
| 実行権限がない | 126 | しない |
| ラッパー内のOSエラー | 125 | しない |
| 子がSIGTERMで終了 | 143 | しない |
| Ctrl+C | 130 | しない |
| 出力先のパイプが閉じられた | 141 | しない |
| 引数エラー | 2 | 起動前に停止する |
子がSIGTERMで終了すると、subprocessの戻り値は-15です。失敗イベントには-15を記録し、ラッパー自体は143で終了します。シグナル終了は通常の一時失敗として再試行しません。
再試行の前に冪等性を確認する
再試行できるのは、同じ処理を繰り返しても業務上の結果が重複しない場合です。少なくとも、次のいずれかが必要です。
- 処理が読み取り専用である
- 同じ処理IDによる登録を保存先が拒否する
- 再実行前に、すでに完了しているか照会できる
- 一時ファイルへ出力し、成功時にだけ確定できる
メール送信、請求登録、外部APIの更新、追記型ファイルへの書き込みなどは、排他ロックだけでは重複を防げません。外部処理が完了した後で応答だけ失われた場合、終了コードからは「未実行」と「実行済み」を区別できないためです。
複数ステップからなるワークフロー全体を再試行すると、失敗前に完了したステップまで繰り返されます。再試行ラッパーは、再実行可能だと確認した個別ステップへ適用します。第173回の排他ロックと組み合わせる場合は、ロックの内側で全試行を実行し、待機中も同じロックを保持します。
この小さな実装の制約
- タイムアウトは実装していないため、終了しない子コマンドを待ち続ける
- 待機時間にランダムな揺らぎがなく、多数の端末が同時実行すると再試行時刻も重なり得る
- 子が起動した孫プロセスの終了や、ラッパー自身がSIGTERMを受けた場合の転送は管理しない
- 書き込み可能で十分な空き容量のあるOSの一時領域が必要になる
- 子の出力サイズを制限しておらず、システムの一時領域を消費する
- 出力は子の終了後に再生されるため、実行中のログをリアルタイムには確認できない
- 標準出力と標準エラー出力を別々に保存するため、両者の細かな出力順序は再現しない
- 最終成功時には、それ以前の失敗履歴を保存しない
この実装の対象は、出力サイズと実行時間を別の運用ルールで制限できる、ローカルの小規模なバッチ処理です。まず既存CLIを一つ選び、成功、恒久失敗、一時失敗の終了コードを定義します。そのうえで、副作用の重複を防げる処理だけを再試行対象にしてください。