I/Oは必ず失敗する — リトライと縮退で制御ループを守る

安全機能を足したら、逆にロボットが倒れやすくなった。

自分が作っているロボットの制御ソフトに「2秒ごとにシリアルバス経由でサーボの温度を読み、過熱していたら止める」温度監視を実装したときの話です。機能そのものは数十行で書けました。温度を読んで、しきい値を超えたら停止命令を送る。シンプルです。しかし動かしてみると、温度が特に高くもない状態で制御ループがまるごと落ちる現象が発生しました。原因はサーボの過熱ではなく、温度を読もうとした瞬間のシリアルバスの通信失敗でした。

ハッピーパスだけのI/O呼び出しは未完成

最初に書いたコードを単純化すると、こういう構造でした。

# 悪い例：ハッピーパスだけのI/O呼び出し
def check_temperature_loop():
    while True:
        temps = read_temperatures()   # 失敗したら例外を投げる
        if max(temps) > TEMP_LIMIT:
            emergency_stop()
        time.sleep(2)

read_temperatures() が使うシリアルバス（複数のデバイスを1本の配線でつなぎ順番に通信する方式）のライブラリは、デフォルトでリトライ回数ゼロ、通信失敗時に例外を投げる仕様でした。CRCエラー（送受信データのビット誤りを検出する仕組みで、1ビットでも化けていれば検知できる）が1回起きれば即座に例外が上がります。そしてその例外を誰も捕まえていなかった。温度監視の呼び出し元も、制御ループも、です。

結果として、例外は制御ループ全体まで伝播して、ロボットが丸ごと落ちていました。「過熱で止める」ための安全機能が、「温度読みの単発の失敗でロボットを落とす」という新しい故障点になってしまったわけです。

物理バスの「失敗して当たり前」という故障モデル

USBで接続したシリアルバスには、一過性の通信失敗が日常的に起きます。ケーブルを伝わるノイズ、複数デバイスが同時に返信するタイミングのずれ、OSのスケジューラによる遅延でタイムアウトになること、CRCビットが化けること——これらは異常ではなく、物理的な伝送媒体を使う以上、ある確率で必ず起きる現象です。

自分はソフトウェア設計の観点では「失敗を前提に書く」ことを知っているつもりでした。しかし実際には、物理バスの故障モデルをまだ染み込ませていなかった。Webの外部API呼び出しでタイムアウトやネットワークエラーを考慮するのと全く同じように、シリアルバスの読み取りにも「失敗して当たり前」という前提が必要でした。

ここで問題になるのが、失敗への対応の粒度です。

あなたなら、I/O呼び出しが失敗したとき、どこまで「許す」設計にしますか？

三段の守り：リトライ・縮退・段階的エスカレーション

修正の方針は「一過性の失敗か、恒久的な失敗か」を段階的に判断することです。1回失敗しただけではロボットを止めない。しかし失敗が続くなら止める。この区別を実装に落とし込んだのが次の三段構造です。

第一段：リトライで一過性の失敗を吸収する

バスの読み取り関数に num_retry パラメータを渡し、同じ読み取りを数回試みます。CRCエラーやタイムアウトの多くは、1〜3回リトライするだけで成功します。一過性の障害をこの層で吸収するのが最初の守りです。

第二段：縮退（graceful degradation）で1回の失敗を無害化する

リトライをすべて使い切っても失敗した場合、例外を握りつぶして「今回は読めなかった」という状態を返します。ロボットを止めるのではなく、ログを残して次の周期まで待ちます。1回の失敗をシステム全体に伝播させない、これが縮退（graceful degradation）の考え方です。

第三段：連続失敗カウンタで段階的にエスカレーションする

縮退し続けていると「本物の故障」を見逃すリスクがあります。そこで連続失敗の回数を数え、N回（たとえば3回）を超えたら「温度が継続的に読めない＝安全を保証できない」と判断し、安全側——すなわち停止——に倒します。

この三段構造をコードで示すとこうなります。

# 良い例：リトライ・縮退・段階的エスカレーション
MAX_CONSECUTIVE_FAILURES = 3
consecutive_failures = 0

def check_temperature_loop():
    global consecutive_failures

    while True:
        result = _read_temperatures_safe(num_retry=3)

        if result is None:
            consecutive_failures += 1
            logger.warning(
                f"温度読み取り失敗 ({consecutive_failures}/{MAX_CONSECUTIVE_FAILURES})"
            )
            if consecutive_failures >= MAX_CONSECUTIVE_FAILURES:
                logger.error("温度監視が継続的に失敗。安全停止します。")
                emergency_stop()
                break
        else:
            consecutive_failures = 0   # 成功したらリセット
            if max(result) > TEMP_LIMIT:
                logger.error(f"過熱検知: {max(result)}°C。停止します。")
                emergency_stop()
                break

        time.sleep(2)


def _read_temperatures_safe(num_retry: int) -> list[float] | None:
    """失敗時はNoneを返す。例外は外に出さない。"""
    try:
        return read_temperatures(num_retry=num_retry)
    except CommunicationError as e:
        logger.debug(f"温度読み取り例外を吸収: {e}")
        return None

consecutive_failures というカウンタが肝です。成功するたびにゼロにリセットし、失敗が連続したときだけカウントが積み上がります。「1回でも失敗したら止める」のも「失敗を永遠に無視する」のも、どちらも危険です。前者は一過性のノイズで誤停止を繰り返し、後者は本物の断線や過熱を見逃します。連続失敗カウンタが「一過性（transient）か恒久（permanent）か」を判断する境界線になります。

WebのタイムアウトとハードI/Oは同じ構造

この設計は、Web開発で外部APIを呼び出すときの定石と同じ構造をしています。タイムアウトを設定して、タイムアウトしたら数回リトライして、それでも失敗したらキャッシュ値で縮退して、閾値を超えたらアラートを飛ばす——まさに同じです。

この分野には既存のパターンが整備されています。リトライ間隔を1回目→2秒、2回目→4秒と指数的に伸ばす exponential backoff は、サーバーへの集中リクエストを分散させる効果があります。連続失敗が続く相手への呼び出しを一時的に遮断し、回復を待ってから再試行する circuit breaker パターンは、障害を隔離してシステム全体への伝播を防ぎます。

どちらのパターンも「失敗は起きる。どう付き合うか」という発想から生まれています。制御ループの外部I/Oも全く同じ発想が必要でした。

「I/Oは必ず失敗しうる。失敗時の挙動が定義されていないI/O呼び出しは未完成」——これを自分なりに言語化できたのが今回の収穫です。

設計のときに最初に決めること

ハードウェアI/Oを書くとき、自分はこれからこの順番で考えます。

• この呼び出しはどの種類の失敗をするか？ タイムアウト、CRCエラー、デバイス未検出——可能性を列挙する
• 一過性か恒久かの閾値 N をいくつにするか？ 制御周期と安全要件から逆算する。2秒周期で3回失敗なら6秒間データがない状態を許容することになる
• N 回失敗したときの安全側はどちらか？ 停止か、縮退継続か。安全機能なら「停止」が基本
• 成功したときにカウンタをリセットするか？ 一時的な回復を「完全回復」と見なすかどうかの設計判断

これを各I/O呼び出しに1行で添えておくだけで、「誰も捕まえていない例外」は生まれなくなります。

自分が足りていなかったこと

今回の問題の根は、抽象的なソフト設計と物理バスの故障モデルの間に隙間があったことです。関数の責務分離や依存性の注入は普段から意識していても、「物理的な伝送経路はノイズが乗る」「USBホストドライバのスケジューリングでタイムアウトが起きる」という前提を、コードの構造に組み込めていませんでした。

シリアルバスの読み取りは、見た目は普通の関数呼び出しです。だからこそ、ファイルI/OやAPIコールと同じように「失敗は例外として上がってくる」という前提で書いてしまう。WebフレームワークやORMが吸収してくれていた失敗を、ここでは自分で設計しなければならない。

コードを見直したとき、気づいたのは一点でした。「この例外、誰が捕まえるんですか？」——この問いを各I/O呼び出しに当てたとき、設計の穴がはっきり見えました。制御ループを書くときは、各I/O呼び出しに「失敗したらどうする」を必ず一行で決めておく——その場リトライ、前回値で縮退、N回で停止のどれかです。それが今回、自分が書き直したことの全てです。