お疲れ様です！IT業界で働くアライグマです！
個人開発している技術記事キュレーションサービス「DevPick」に、利用者が好きな外部RSSフィードを自由に追加して購読できる機能を作りました。自分が普段読んでいる技術ブログや会社のエンジニアブログを、DevPickのAIスコアリングフィードにまとめて流し込める機能です。
ところが、このカスタムRSSを裏側で定期的に巡回して新着記事を取り込むバッチを動かし始めたところ、すぐに2つの壁にぶつかりました。「並行するプロセスが同じフィードを二重に取得しにいってしまう問題」と、「リンク切れやサーバダウンで応答しない壊れたフィードのせいで、巡回バッチ全体が詰まってしまう問題」です。今回は、その2つの問題に対処するためにリース方式と自動停止を入れた経緯を書きます。
外部のRSSを定期巡回させると、壊れたフィードに足を取られる
DevPickで従来扱っていたニュースソースは、私が運用側であらかじめ動作確認した信頼できる主要メディアだけでした。配信形式やレスポンス速度も安定しており、定期巡回で大きく詰まることは滅多にありません。
しかし、利用者が登録するカスタムRSSとなると前提がまったく変わります。個人ブログ、技術コミュニティ、企業のテックブログなど、URLの形式もホスティング環境もバラバラです。
当然ながら、中にはURLの入力ミスで最初から404エラーになるものや、すでに閉鎖されて応答しないサイト、ドメインの有効期限が切れて接続タイムアウトになるものが紛れ込みます。悪意がなくても、相手先のサーバメンテナンスや障害で一時的に接続できなくなることもしょっちゅうあります。以前定期ジョブの実行管理で詰まった話でも触れましたが、スケジューラで動かすジョブには実行時間の枠に限りがあります。
これらを無防備に毎回巡回対象に含めていると、1件のフィード取得にタイムアウトの十数秒を消費してしまい、バッチ全体の処理時間が雪だるま式に伸びていきます。その結果、巡回間隔に全体の処理が収まらなくなり、正常に動いている他の大半のフィードが何時間も巡回されない「飢餓（Starvation）」が起きる危険がありました。
「相手のある外部通信は、いつか必ず失敗する」という前提で巡回網を組み立てないと、壊れた一部のフィードに引きずられてシステム全体が機能不全に陥ります。
IT女子 アラ美
ITアライグマ複数プロセスで同じフィードを二重取得しないための「リース方式（Lease）」
DevPickのバックエンドは可用性と負荷分散のために複数プロセスで稼働しています。このとき最初にぶつかったのが、定期巡回バッチが並行して起動した際に「同じフィードを複数のプロセスが同時に取得しにいってしまう」問題でした。
1つのフィードに対して同時に2つのプロセスが外部アクセスを行い、新着記事の重複判定や保存処理を同時に走らせてしまいます。これは相手先サーバーに対しても迷惑ですし、自サーバーのリソースやDB接続も無駄に食い合います。
最初は、DBから巡回対象を引いてきた上で、処理直前に「最終フェッチ日時が取得時点と同じなら更新する」という楽観ロック（CAS: Compare-And-Swap）を入れました。しかし、これだけでは「処理中にプロセスが再起動した場合」や「ジョブが少し遅延して重複起動した場合」に、直前の実行が完了する前に次の実行が割り込んでしまい、排他が崩れるリスクが残りました。
そこで採用したのが、有効期限を持たせた「リース方式（Lease）」です。
巡回対象を選ぶ際、「前回のフェッチ日時が一定時間（例: 10分）以上前のもの、または一度もフェッチされていないもの」だけに絞り込みます。そして各フィードの処理を開始する直前に、DBの条件付き更新（UPDATE）を使って原子的に処理権（リース）を確保します。
単純化した処理の流れは次のようになります。
# リース期間（例: 10分）を定義
LEASE_DURATION = timedelta(minutes=10)
cutoff = datetime.now(timezone.utc) - LEASE_DURATION
# 1. 巡回対象の候補を古い順に取得（直近10分以内に処理されたものは除外）
candidates = db.query(FeedSource.id).filter(
FeedSource.is_active == True,
or_(
FeedSource.last_fetched_at.is_(None),
FeedSource.last_fetched_at <= cutoff,
),
).order_by(FeedSource.last_fetched_at.asc()).limit(50).all()
for source_id in candidates:
# 2. 条件付きUPDATEで処理権（リース）を原子的（atomic）に確保する
result = db.execute(
update(FeedSource)
.where(
FeedSource.id == source_id,
FeedSource.is_active == True,
or_(
FeedSource.last_fetched_at.is_(None),
FeedSource.last_fetched_at <= cutoff,
),
)
.values(last_fetched_at=datetime.now(timezone.utc))
)
db.commit()
# 更新行数が0なら、別のプロセスが先にリースを確保したと判断してスキップ
if result.rowcount == 0:
continue
# 3. 処理権を確保できたプロセスだけが実際のRSSフェッチを実行
fetch_and_store(source_id)
条件付きUPDATEの戻り値（影響を受けた行数）が 1 であれば、そのプロセスが確実にそのフィードの処理権（リース）を獲得したことになります。もし他プロセスがコンマ数秒先に確保していれば、条件が不一致となり更新行数は 0 になるため、安全にスキップされます。
さらに、成功・失敗に関わらずフェッチ開始時点で last_fetched_at が更新されるため、上限件数（50件）で積み残された他のフィードが次回巡回時に優先され、特定のフィードだけが取り残されることも防げるようになりました。
IT女子 アラ美ITアライグマ死んだエンドポイントへの無駄撃ちを止めるサーキットブレーカー
リース方式によって複数プロセス間での処理権の衝突は解消できましたが、もう1つの問題である「壊れた外部フィードへの無駄撃ち」は依然として残っていました。
サイトが閉鎖されていたり、URLの指定が間違っていたりするフィードは、何度巡回しても毎回タイムアウトやHTTPエラーになります。1回の巡回で50件の上限枠を設けている中で、こうした死んだエンドポイントが何十件も居座り続けると、正常なフィードの巡回枠が圧迫され続けてしまいます。
そこで、2つの防護策を入れました。
1つ目は、各フィードの処理を完全に独立したDBセッションで実行する「エラー隔離」です。バッチ全体で1つのDBセッションを使い回していると、あるフィードの取り込み中に予期しない例外が起きた際、セッションが壊れて後続の正常なフィードの保存まで失敗してしまいます。フィードごとにセッションを独立させ、失敗したフィードだけを個別にロールバックしてログに記録する構造に改めました。
2つ目は、連続失敗をカウントして巡回から除外する「サーキットブレーカー」です。
フィードの取得に失敗するたびに失敗回数をカウントし、連続5回失敗した時点でそのフィードを自動的に「停止状態」に切り替えます。
停止状態になったフィードは、定期巡回の対象クエリ（有効なものだけを取得する条件）から自動的に弾かれます。これにより、壊れた配信先に対して無駄なHTTP通信やタイムアウト待ちを繰り返すことが一切なくなりました。利用者の設定画面側にも「5回連続で取得に失敗したため、自動停止しています」と理由を表示し、URLの誤りなどに気づいてもらえるようにしています。
ここで大事だったのが、「失敗カウントのリセットタイミング」です。
巡回バッチ側で適当にリセットしてしまうと、壊れたフィードが定期的にゾンビのように復活して再び巡回網を詰まらせます。そのため、リセットは「利用者が設定画面でURLを確認し、明示的に再開操作を行ったときだけ」に限定しました。利用者の意思が介在しない限り、死んだエンドポイントへ余計な通信を飛ばさない設計です。
IT女子 アラ美ITアライグマまとめ
外部のRSSフィードを定期的に巡回してデータを取り込む仕組みは、一見すると単にループを回してHTTPリクエストを投げるだけのシンプルな処理に見えます。
しかし、利用者が自由にフィードを追加できるWebサービスでは、次のような「相手側の障害」や「自側の並行処理」を最初から織り込んでおく必要がありました。
- 複数プロセスによる二重取得：条件付きUPDATEと有効期限を組み合わせた「リース制御」により、ロック専用サーバーを用意することなく原子的（atomic）に処理権を排他する
- 処理の飢餓（Starvation）防止：フェッチ日時が古い順に上限件数ずつ巡回し、フェッチ開始時に日時を更新することで積み残されたフィードを順繰りに巡回させる
- 死んだエンドポイントへの無駄撃ち遮断：連続失敗をカウントして自動停止（サーキットブレーカー）させ、再開操作があるまで無駄なリクエストを完全に止める
- 個別のエラー隔離：フィードごとに独立したDBセッションを使い、一部の例外がバッチ全体に波及しないようにする
外部の不特定多数と通信する機能を作る際は、「相手が正常に応答してくれる前提」ではなく、「相手が壊れていても自分たちのシステムは平然と動き続けられる構造」を初期段階から仕込んでおくことが、安定運用の鍵だと実感しました。
IT女子 アラ美ITアライグマ
