午前2時の悪夢：ネットワークエラーで沈黙するスクリプト
多くのバックエンドエンジニアにとって、これは見覚えのある光景でしょう。提携先CRMからPostgreSQLへ50,000件の注文データを同期するワーカーを構築したとします。ローカル環境で100件のレコードを使ってテストした際は問題なく動作しました。しかし、本番サーバーにデプロイして深夜にバッチ処理を実行したところ、処理開始から3時間後に突如ワーカーが停止してしまいました。
ターミナルのログに残されていたのは、わずか1行のエラーでした。requests.exceptions.ConnectionError: ('Connection aborted.', RemoteDisconnected())、あるいは 502 Bad Gateway。パイプラインは途中で中断し、データは中途半端な状態のまま残されます。深夜2時にベッドから這い起き、再実行ボタンを押すためだけにPCを開かなければなりません。そして何よりやり切れないのは、手動で再実行すると何事もなかったかのように正常終了することです。
一時的障害（Transient Faults）の正体：一時的でも被害は甚大
分散システムやマイクロサービス環境において、ネットワーク接続が100%安定し続けることはありません。数ミリ秒から数秒程度発生する瞬間的な寸断はTransient Faults（一時的障害）と呼ばれます。代表的な原因には次の3つが挙げられます。
- ネットワーク層の不安定さ： 一時的なルーティングの再計算（BGPフラッピング）や瞬間的なパケットロス。
- 外部サービスのRate limit： 連携先サーバーの負荷が高まり、2〜5秒間
429 Too Many Requestsや503 Service Unavailableを返した後に自動回復するケース。 - Databaseのフェイルオーバー： PostgreSQLのプライマリからレプリカへの切り替えや、PgBouncerによる定期的な接続プールのリセット。
アプリケーションがあらゆる例外を一律で致命的なエラー（Fatal error）とみなして処理を即座に中断してしまうと、クラウド環境において極めて脆弱なシステムになってしまいます。
よくある2つの場当たり的な対処法（そして避けるべき理由）
アプローチ1：手動の try-except を while/for ループで囲む
Python初学者が最も直感的に書きがちな実装パターンです。
import time
import requests
def fetch_order_manual(order_id):
max_retries = 3
delay = 2 # 秒
for attempt in range(max_retries):
try:
res = requests.get(f"https://api.crm.com/orders/{order_id}", timeout=5)
res.raise_for_status()
return res.json()
except requests.RequestException as e:
if attempt == max_retries - 1:
raise
print(f"試行 {attempt + 1} 回目が失敗しました。{delay}秒待機します: {e}")
time.sleep(delay)
このコードでも動作自体はしますが、本番環境では以下の深刻な問題を引き起こします。
- ボイラープレートコードの氾濫： プロジェクト内にAPI呼び出しやDBクエリが30箇所あれば、この
for/whileブロックを30箇所にコピペすることになります。リトライロジックを変更する際も、30ファイルすべてを修正しなければなりません。 - 群れ効果（Thundering Herd問題）： たとえば500台のワーカーが12:00:00に同時にネットワークエラーに遭遇したとします。すべてのワーカーが一斉に2秒待機し、12:00:02に連携先サーバーへ同時に500リクエストを浴びせることになります。これにより、過負荷状態にあったサーバーが完全にダウンしてしまいます。
アプローチ2：独自のカスタムデコレータを作成する
再利用性を高めるためにカスタムデコレータを自作する開発者も多くいます。しかし、システムが拡張するにつれて自作実装はすぐに行き詰まります。指数バックオフの計算、待機時間のランダム化（Jitter）、ステータスコードのフィルタリング、async/await対応など、考慮すべき点は山積みです。車輪の再発明は避けましょう。
Tenacityライブラリによるリトライ処理の標準化
Tenacityは、メンテナンスが終了したretryingライブラリをベースに再設計された、Pythonエコシステムにおける標準的なリトライライブラリです。Tenacityの最大の強みは、デコレータによる宣言的（declarative）な構文にあります。開発者はリトライの条件を記述するだけでよく、フローの制御はライブラリ側に任せることができます。
1. インストール
pip install tenacity
2. 基本的なリトライ：試行回数の制限
ネットワーク通信を行う関数をラップするには、@retry デコレータに終了条件 stop_after_attempt を組み合わせて指定します。
from tenacity import retry, stop_after_attempt
import requests
@retry(stop=stop_after_attempt(3))
def sync_user_profile(user_id):
print(f"ユーザープロファイル {user_id} を取得中...")
response = requests.get(f"https://api.partner.com/users/{user_id}", timeout=3)
response.raise_for_status()
return response.json()
例外が発生した場合、Tenacityは自動でエラーを捕捉し、最大3回まで関数を再呼び出しします。この上限を超えると元の例外が再送出（re-raise）され、Sentryなどの監視システムが正しく検知できるようになります。
3. Exponential Backoff（指数バックオフ）とJitterの組み合わせ
エラー直後に即座にリトライしてはいけません。ベストプラクティスは、失敗するたびに待機時間を指数関数的に延ばす（1秒、2秒、4秒、8秒…）Exponential Backoffです。さらに、ワーカー同士の負荷を分散させるためにFull Jitter（待機時間のランダム化）を併用します。
from tenacity import retry, stop_after_attempt, wait_random_exponential
import psycopg2
@retry(
stop=stop_after_attempt(5),
wait=wait_random_exponential(multiplier=1, max=10)
)
def execute_db_query(query):
conn = psycopg2.connect("dbname=sales user=app password=secret host=db-cluster")
with conn.cursor() as cursor:
cursor.execute(query)
return cursor.fetchall()
上記の設定では、ワーカーは最大5回までリトライを試行します。試行間の待機時間は multiplier = 1 を基準として指数関数的に増加し、そこにランダムな揺らぎが加わり、最大でも10秒を超えないように制御されます。
4. 一時的エラーのみをリトライ：盲目的なリトライは厳禁
最も危険なアンチパターンは、ビジネスロジック上のエラーまでリトライしてしまうことです。APIが 401 Unauthorized（トークン期限切れ）や 422 Unprocessable Entity（不正なスキーマのペイロード）を返している場合、何千回リトライしても100%失敗します。さらに悪いことに、冪等性（Idempotency）が担保されていない決済処理をリトライすると、二重課金を引き起こす恐れもあります。
リトライの対象は、5xx 系のサーバーエラーやネットワーク寸断（ConnectionError、Timeout）などの一時的エラーに限定しましょう。
from tenacity import retry, stop_after_attempt, retry_if_exception_type
import requests
class UpstreamTransientError(Exception):
"""アップストリームのネットワークエラーまたは一時的な過負荷。"""
pass
def call_remote_service():
try:
res = requests.get("https://api.inventory.com/items", timeout=4)
# 502, 503, 504は一時的エラー、4xxは基本的にクライアントエラー
if res.status_code in (502, 503, 504):
raise UpstreamTransientError(f"アップストリーム混雑: HTTP {res.status_code}")
res.raise_for_status()
return res.json()
except (requests.ConnectionError, requests.Timeout) as exc:
raise UpstreamTransientError("アップストリームへの接続が切断されました") from exc
@retry(
stop=stop_after_attempt(4),
retry=retry_if_exception_type(UpstreamTransientError)
)
def get_inventory():
return call_remote_service()
実際の現場では、標準のHTTPステータスコードを返さず、HTTP 200とともに {"status": "FAILED_DB_LOCK_TIMEOUT"} のようなレスポンスボディを返すレガシーシステムも少なくありません。その場合、再送出するかリトライするかを判定する前に、正規表現（Regex）を使ってエラー文字列のパターンを抽出する必要があります。実際のログ文字列に対して正規表現パターンをすばやく検証したい場合は、何度も print デバッグを挟む代わりに、ブラウザ上でキャプチャグループを直接確認できる toolcraft.app/ja/tools/developer/regex-tester の正規表現チェッカーを活用するとスムーズです。
5. グレースフル・フォールバック：リトライ上限時の代替データ返却
規定回数のリトライを終過しても対象サーバーが応答しない場合、処理全体をクラッシュさせるのではなく、retry_error_callback パラメータを利用してフォールバック処理を起動します。
from tenacity import retry, stop_after_attempt, wait_fixed
def get_cached_rates(retry_state):
attempt = retry_state.attempt_number
print(f"[WARNING] API呼び出しが {attempt} 回失敗しました。Redisからフォールバックデータを取得します。")
# キャッシュされた古いデータまたは安全な代替データを返す
return {"USD": 25400, "EUR": 27100, "is_stale": True}
@retry(
stop=stop_after_attempt(3),
wait=wait_fixed(1),
retry_error_callback=get_cached_rates
)
def fetch_live_exchange_rates():
# レート取得APIの完全ダウンをシミュレート
raise ConnectionResetError("Gateway Timeout")
rates = fetch_live_exchange_rates()
print("使用データ:", rates)
まとめ
クラウド環境において、ネットワークエラーを完全にゼロにすることは不可能です。煩雑でメンテナンスの難しい try-except ループを自作する代わりに、Tenacityを採用することで、コードベースを簡潔かつ明瞭に保ちながら高い耐障害性を実現できます。適切なデコレータを付与し、Jitterを組み合わせたExponential Backoffを設定して、対象の例外を適切にフィルタリングするだけで、深夜に起こされる障害アラートの9割を防ぐことができるでしょう。
