Redisリスナースレッドが死亡した場合に再起動するよう修正

Action Cable の Redis アダプタは、ブロードキャスト用に常駐スレッドを生成して接続を維持しますが、再接続試行が尽きたときにスレッドが終了すると、以降の subscribe がキューに詰め込まれたまま放置され、プロセスを再起動するまで放送が失われます。本PR はその状態を検知してスレッドを再生成し、再接続カウンタもリセットすることで、死んだリスナースレッドからの復旧を自動化します。

背景

Action Cable の Redis リスナースレッドは ensure_listener_running メソッドで @thread ||= Thread.new とメモ化されているため、一度生成されたスレッドが死んでも @thread が nil にならず再生成されません。この挙動は #57690 で修正されたサブスクリプション数が 0 になるケースを防ぎますが、再接続試行回数が枯渇した場合 にはスレッドが終了したまま残り、以降の購読がすべて失われるという新たな障害を引き起こします。

技術的な変更

actioncable/lib/action_cable/subscription_adapter/redis.rb の Listener#ensure_listener_running に、死んだスレッドを検知してリセットするロジックが追加されました。具体的には以下のコードが挿入されています。

def ensure_listener_running
  # The internal sentinel subscription keeps the listener alive in
  # normal operation, but the thread can still die for other reasons
  # (e.g. exhausting the Redis reconnect attempts). Since this method
  # memoizes with `||=`, a dead thread would never be replaced and
  # every later subscribe would queue forever. Drop a dead thread so
  # the next subscribe spawns a fresh listener.
  if @thread && !@thread.alive?
    @thread = nil
    @reconnect_attempt = 0
  end

  @thread ||= Thread.new do
    Thread.current.abort_on_exception = true
    # … existing listener loop …
  end
end

このチェックは @thread が存在しかつ alive? が false のときに、スレッド参照を nil に戻し同時に @reconnect_attempt カウンタを 0 にリセットします。その後の @thread ||= Thread.new が新しいスレッドを生成できるようになります。コード変更は追加だけで既存ロジックを削除せず、既存の true/false 設定やサブスクリプションの動作に影響を与えません。

テスト側では actioncable/test/subscription_adapter/redis_test.rb にリスナースレッドが死んだ後に再起動することを検証するテストが追加されました。reconnect_attempts: 0 で即座に再接続を諦めさせ、drop_pubsub_connections 後にスレッドが死亡したことを wait_for ヘルパで確認し、続く新規チャンネル購読でスレッドが再生成されブロードキャストが正常に届くことをチェックします。

# 省略されたテスト本体は上記記事本文で示した通り

さらに、ポーリング用の wait_for ヘルパが actioncable/test/test_helper.rb に実装され、非同期状態の確認を固定スリープではなく条件が満たされるまで繰り返しチェックする形に改善されました。

def wait_for(message: "condition not met", timeout: 5, interval: 0.01)
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
  loop do
    return if yield
    if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
      raise Timeout::Error, "#{message} after #{timeout} seconds"
    end
    sleep interval
  end
end

設計判断

この変更は スレッドのリサイクルを防ぎつつ再生成を許容する という設計判断に基づきます。@thread のメモ化はパフォーマンス上有用ですが、死んだスレッドが再利用されるリスクを伴います。そこで、死んだスレッドを明示的に破棄しカウンタをリセット する最小限の分岐を導入することで、既存コードへの侵入度を抑えつつ信頼性を向上させました。代替案としては Thread.new を毎回呼び出す方式や、別キーでスレッド管理を行う案が考えられましたが、後者は大規模なリファクタリングが必要になるため採用されませんでした。

まとめ

ensure_listener_running に死んだスレッド検知とリセット処理を追加したことで、Redis 接続が再接続できずにスレッドが終了したケースでも自動的にリスナーが再起動します。これにより Action Cable のブロードキャストがプロセス再起動まで失われるリスクが除去され、既存の設定や動作に影響を与えない形で信頼性が向上しました。