TL;DR
-
try/exceptは失敗を捕まえる道具で、返ってこないを捕まえる道具ではない。外部ライブラリの C 拡張の中で固まると、例外は一度も発生しない - 無応答を測れるのは別スレッドだけ。そしてタイムアウトしたスレッドは kill できないので、タイムアウトしたら続行せずプロセスごと落とす
- 生存の根拠にログの最終行を使わない。ログは「出力に成功した」ことしか示さない。「1周終えた」だけを書くハートビートを別に持つ
- 監視結果を
OK / NGの2値にしない。「古い」と「見られなかった」は別物で、後者を PASS に丸めると「一度も起動していないプロセス」が正常に見える - 監視器自身も監視対象に入れる。検知器だけが古い版のまま動いていると、他を全部直しても何も通知されない
この記事のコードは標準ライブラリだけで動きます。全文を本文に置いてあり、最後に単体テストも付けてあります。コピーしてそのまま使えます。
この記事の監視コードと単体テストは GitHub に置いてあります（MIT）。
前提：何が起きたか
常駐プロセスが1つ、32時間止まっていました。プロセスは生きています。だから死活監視は全部「正常」と報告していました。
ログの最終行はこうです。
2026-09-06 22:00:13 [INFO] [WEEKEND] weekend_guard: Sunday 21:00 UTC < open 22:00
周期処理の中で、外部 SDK 越しに履歴を取りに行く関数がありました。ちょうど外部サービス側が再接続中のタイミングで、そこに当たっています。
厄介なのはここからで、この関数はすでに try/except Exception で囲まれていました。それでも32時間止まりました。
固まっているのが Python 側ではなく C 拡張の中だからです。例外は発生しません。ただ返ってきません。
try:
data = sdk.fetch_history() # ← C 拡張の中で無限に待つ
except Exception as e:
log.error(e) # ← ここには永久に来ない
例外ハンドラは「失敗」を捕まえる道具であって、「無応答」を捕まえる道具ではない、という一行に尽きます。
実装1：タイムアウト付き実行
無応答を測るには、待っている側とは別のスレッドが要ります。signal.alarm は使えません（メインスレッドでしか張れず、C 拡張の中にいる間はハンドラが走りません）。
import threading
def call_with_timeout(fn, timeout_sec, *args, **kwargs):
"""fn を別スレッドで実行し、timeout_sec 以内に返らなければ
(None, TimeoutError) を返す。
返り値: (result, error)
- 正常終了 : (戻り値, None)
- 例外 : (None, 例外インスタンス)
- 無応答 : (None, TimeoutError)
"""
box = {}
def runner():
try:
box["result"] = fn(*args, **kwargs)
except BaseException as e: # noqa: BLE001 - 呼び出し側へ渡す
box["error"] = e
th = threading.Thread(target=runner, daemon=True)
th.start()
th.join(timeout_sec)
if th.is_alive():
return None, TimeoutError(
f"{getattr(fn, '__name__', fn)} が {timeout_sec}s 以内に返りませんでした"
)
if "error" in box:
return None, box["error"]
return box.get("result"), None
ポイントを3つだけ。
daemon=True にする。 そうしないと、固まったスレッドがプロセスの終了を妨げます。落としたいときに落ちない監視は監視ではありません。
except BaseException にする。 Exception だけだと KeyboardInterrupt や SystemExit を取りこぼし、box が空のまま「正常終了して None を返した」と区別がつかなくなります。
結果を戻り値で返し、例外を再送出しない。 呼び出し側で「無応答」と「失敗」を別扱いしたいからです。同じ except に流れ込むと、また2値に潰れます。
タイムアウトしたら、続行してはいけない
ここが一番大事なところです。
result, err = call_with_timeout(run_once, RUN_ONCE_TIMEOUT_SEC)
if isinstance(err, TimeoutError):
log.error("run_once が無応答。プロセスを終了します")
os._exit(1) # ← 続行しない
タイムアウトしたスレッドは kill できません。 Python にスレッドを外から殺す API はありません。
放置して次のサイクルへ進むと、固まったスレッドが後から解けて、同じ処理が二重に走ります。副作用のある処理（発注・課金・メール送信・ファイル書き込み）なら、これは無応答そのものより高くつきます。
プロセスごと落として、スーパーバイザ（systemd / supervisor / Windows サービス / 自前の launcher）に起こさせるのが唯一安全な出口です。
sys.exit() ではなく os._exit() を使っているのは、sys.exit() が SystemExit を投げるだけで、固まっているスレッドがいる限りインタプリタが終了しないからです。
実装2：ハートビート
「生きているか」の根拠に、ログの最終行の時刻を使わないでください。
ログが示しているのは「出力に成功した」ことだけです。周回したことは示しません。逆に、周回しているのにログを出さない経路（何も無かったので早期 return する、など）では、正常なのに古く見えます。
「1周終えた」だけを書く専用のファイルを持たせます。
import os
import json
import time
from datetime import datetime, timezone
class Heartbeat:
def __init__(self, path, name):
self.path = path
self.name = name
def beat(self, **extra):
tmp = f"{self.path}.tmp"
payload = {
"name": self.name,
"pid": os.getpid(),
"at": datetime.now(timezone.utc).isoformat(),
"epoch": time.time(),
**extra,
}
with open(tmp, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False)
os.replace(tmp, self.path) # 原子的に差し替える
@staticmethod
def age_sec(path):
"""最後の beat から何秒経ったか。ファイルが無ければ None。"""
try:
with open(path, encoding="utf-8") as f:
return max(0.0, time.time() - float(json.load(f)["epoch"]))
except (OSError, ValueError, KeyError):
return None
使う側はループの末尾で1行呼ぶだけです。
hb = Heartbeat("/var/run/worker.hb.json", "worker")
while True:
result, err = call_with_timeout(run_once, 60)
if isinstance(err, TimeoutError):
os._exit(1)
hb.beat(loop=n, handled=len(result or []))
time.sleep(interval)
tmp に書いてから os.replace で差し替えているのは、読み手が書きかけの JSON を読むのを防ぐためです。os.replace は同一ファイルシステム上で原子的なので、読み手からは「古い内容」か「新しい内容」しか見えません。直接 open(path, "w") で書くと、その瞬間にファイルは0バイトになり、監視側が JSONDecodeError を踏みます。監視器が監視対象のせいで落ちるのは、いちばん間抜けな壊れ方です。
epoch（数値）と at（ISO 文字列）を両方書いているのは、前者が計算用、後者が人が目で読む用だからです。片方から片方を導けますが、障害対応中にタイムゾーンの変換をやりたくありません。
実装3：判定は3値にする
def check_heartbeats(specs, now=None):
"""specs: [(name, path, 許容秒), ...] → [(name, 状態, 経過秒 or None)]"""
out = []
for name, path, limit in specs:
age = Heartbeat.age_sec(path)
if age is None:
out.append((name, "UNKNOWN", None))
elif age > limit:
out.append((name, "STALE", age))
else:
out.append((name, "OK", age))
return out
OK / STALE / **UNKNOWN** の3値にしてください。
age_sec がファイルを読めなかったときに 0.0 を返す実装にすると、一度も起動していないプロセスが「たった今ビートした」ことになります。これは机上の話ではなく、実際にやりがちです。「読めなかったら安全側に倒して 0」と書いた瞬間、安全側の向きを間違えています。
「古い」と「見られなかった」は別の事実です。測定不能を PASS に丸めない、とだけ覚えておけば足ります。
実装4：監視器自身を監視する
上の3つを入れても、監視器だけが古い版のまま動いていたら何も通知されません。
これは実際に踏みました。対象プロセスの不具合が5日間と32時間それぞれ無通知だった根本原因が、これです。デプロイ手順の中で、監視器の再起動だけが標準の経路の外に置かれていました。一番後回しにされる再起動が、一番効く監視のものだった、という話です。
対策は単純で、監視器にも同じハートビートを打たせ、版を一緒に書くだけです。
def self_report(path, version):
Heartbeat(path, "watchdog").beat(version=version)
WATCHDOG_VERSION = "2026.09.23"
specs = [
("worker-a", "/var/run/a.hb.json", 120),
("worker-b", "/var/run/b.hb.json", 300),
("watchdog", "/var/run/watchdog.hb.json", 90), # ← 自分も入れる
]
self_report("/var/run/watchdog.hb.json", WATCHDOG_VERSION)
for name, state, age in check_heartbeats(specs):
...
監視器の版が、監視対象と同じ画面・同じ経路で見えることが肝心です。別の場所に書くと、結局それを見る習慣ができません。
その検査が本物かを確かめる
ここまで書いたら、最後に検査自身を検査します。
やり方は、わざと壊した版に当てて、ちゃんと落ちることを確認するだけです。通ることの確認より、落ちることの確認のほうが情報量があります。
# 🔴 壊れた版: 読めなかったときに 0 を返している（測定不能を PASS に丸めた）
def age_sec_BROKEN(path):
try:
with open(path, encoding="utf-8") as f:
return max(0.0, time.time() - float(json.load(f)["epoch"]))
except (OSError, ValueError, KeyError):
return 0.0 # ← ここだけが違う
存在しないハートビート（＝一度も起動していないプロセス）を両方に食わせます。
壊れた版 : [('worker', 'OK', 0.0)]
正しい版 : [('worker', 'UNKNOWN', None)]
壊れた版は、一度も起動していないプロセスを「正常」と言いました。この1行を出せて初めて、3値にしたことに意味があったと言えます。
単体テスト（全文）
import os, sys, time, tempfile
from watchdog import call_with_timeout, Heartbeat, check_heartbeats
def test_normal():
r, e = call_with_timeout(lambda: 1 + 1, 1.0)
assert (r, e) == (2, None), (r, e)
def test_exception():
def boom(): raise ValueError("x")
r, e = call_with_timeout(boom, 1.0)
assert r is None and isinstance(e, ValueError), (r, e)
def test_timeout():
t0 = time.time()
r, e = call_with_timeout(lambda: time.sleep(5), 0.3)
dt = time.time() - t0
assert r is None and isinstance(e, TimeoutError), (r, e)
assert dt < 1.0, dt # 5秒待たずに返ること
def test_args():
r, e = call_with_timeout(lambda a, b=0: a + b, 1.0, 3, b=4)
assert (r, e) == (7, None), (r, e)
def test_heartbeat_roundtrip():
d = tempfile.mkdtemp()
p = os.path.join(d, "hb.json")
assert Heartbeat.age_sec(p) is None # 未作成は UNKNOWN
Heartbeat(p, "worker").beat(loop=1)
age = Heartbeat.age_sec(p)
assert age is not None and age < 1.0, age
res = check_heartbeats([("worker", p, 60), ("missing", p + "x", 60)])
assert res[0][1] == "OK", res
assert res[1][1] == "UNKNOWN" and res[1][2] is None, res
def test_heartbeat_stale():
d = tempfile.mkdtemp(); p = os.path.join(d, "hb.json")
Heartbeat(p, "w").beat()
time.sleep(0.2)
assert check_heartbeats([("w", p, 0.1)])[0][1] == "STALE"
def test_atomic_no_tmp_left():
d = tempfile.mkdtemp(); p = os.path.join(d, "hb.json")
Heartbeat(p, "w").beat()
assert os.listdir(d) == ["hb.json"], os.listdir(d)
for name, fn in sorted(list(globals().items())):
if name.startswith("test_"):
fn(); print(f" [OK] {name}")
print("\n全て通過")
$ python3 test_watchdog.py
[OK] test_args
[OK] test_atomic_no_tmp_left
[OK] test_exception
[OK] test_heartbeat_roundtrip
[OK] test_heartbeat_stale
[OK] test_normal
[OK] test_timeout
全て通過
test_timeout が dt < 1.0 を見ているのが地味に効きます。「TimeoutError が返った」だけを見ていると、5秒待ってから返す実装（つまりタイムアウトが効いていない実装）を通してしまいます。
チェックリスト
コピーして、CLAUDE.md / AGENTS.md のような常時読ませる指示書や、レビュー項目に足してください。物語のままだと次の事故を防ぎませんが、チェックリストの一行にすると効きます。
## 無応答まわりの検査
- [ ] 「無応答」を検知できるか。try/except は *失敗* を捕まえる道具であって、
*返ってこない* を捕まえる道具ではない。外部ライブラリの C 拡張は
例外を出さずに固まる。タイムアウトは別スレッドでしか測れない
- [ ] タイムアウト後に続行していないか。固まったスレッドは kill できないので、
後から解けて二重に実行される。副作用のある処理なら落として起こし直すこと
- [ ] 生存の根拠にログの最終行を使っていないか。ログは「出力に成功した」ことしか
示さない。「1周終えた」だけを書くハートビートを別に持つこと
- [ ] 測定不能を PASS に丸めていないか。「古い」と「見られなかった」を
同じ値で返していないか（OK/NG の2値にしない）
- [ ] 監視器自身が監視対象に入っているか。版も一緒に出ているか。
検知器だけが古いと、他を全部直しても何も通知されない
- [ ] ハートビートの書き込みは原子的か。直接上書きすると読み手が
書きかけを読む。監視器が監視対象のせいで落ちるのは避けること
- [ ] その検査が本物か確かめたか。わざと壊した版に当てて、
ちゃんと同じ場所で落ちることを確認したか
まとめ
死活監視で「プロセスが存在するか」を見ているうちは、生きたまま黙っている状態を拾えません。拾うには、
- 外側から時間を測る（別スレッド）
- タイムアウトしたら続行せず落とす
- 「1周した」だけを原子的に記録する
- 「見られなかった」を PASS に丸めない
- 監視器自身も同じ経路で見る
の5つで足ります。全部、標準ライブラリだけで書けます。
いちばん効いたのは 5 でした。1〜4 を入れた後も、しばらく何も通知が来なくて「直った」と思っていたのですが、実際には監視器が古い版のままだっただけでした。
実コードと単体テストは、この記事用の着地ページにまとめてあります。
この監視がどこから来たか
この無応答検知は、「戦略のアイデアを7日で検定して、通らなければ捨てる」 手順を実際に回すための運用側の道具です。検定そのものを自動で回し続けるには、止まったことに気づける ことが先に要りました。32時間黙っていたのは、その検定を回していたプロセスです。
検定の7日ぶん（戦略を検証できる文に翻訳する → コストを実測する → 事前宣言で検定条件を固定する → 判定する）は有料noteにまとめました。題材に選んだ戦略は Day5 で死んでいます。 うまくいった話ではなく、落とすまでの手順です。
その「勝てるんじゃない?」、1週間でEAになります。 — ¥1,980