非同期バルク処理とキュー設計

非同期バルク処理で、数件から数十万件まで同じキュー・同じ worker プールへ流すと、少数の長時間ジョブが worker 枠を占有し、数件の短時間ジョブまで開始待ちになる。FIFO キューのメッセージグループやテナント単位ロックは「同一主体内の直列化」には有効だが、worker 枠そのものを長時間ジョブが占有する問題は解決しない。

判断基準

• 小規模ジョブに対して「いつでも同程度の開始体験」を求めるなら、小規模ジョブ用の queue / topic / priority lane と、そこ専用の予約済み worker 枠を持つ。
• 大規模ジョブは別レーンに流し、低めの並列度・長めの visibility / lease・chunk 処理でスループットを最適化する。
• 同一主体で同時実行したくない業務制約がある場合は、キュー順序だけに寄せず、DB の active lock やユニーク制約でジョブ作成時点に排他を確定する。これにより UI は「待たせる」ではなく即座に競合を返せる。
• cold start が大きい実行基盤を小規模ジョブに使うと開始体験が悪化する。小規模は温かい worker、巨大処理は専用 task / bulk lane という分離が扱いやすい。

落とし穴

• 単一 FIFO キューで MessageGroupId を主体 ID にしても、worker 数が少なく長時間ジョブで埋まれば他主体の短時間ジョブは待つ。
• 優先度付きの自前 scheduler を作ると柔軟だが、lease、再実行、監視、DLQ 相当を自分で持つことになる。まずは size class による少数レーン分離で足りるかを見る。
• DB と外部キューへの二重書き込みは不整合になり得る。重要なジョブでは transactional outbox や未送信イベントのリカバリを用意する。

検証

長時間ジョブを予約済み bulk 枠いっぱいに投入した状態で、小規模ジョブを投入し、開始までの時間が小規模レーンの SLO 内に収まることを確認する。同一主体からの二重投入はジョブ作成時点で競合になり、キュー内で待ち続けないことも確認する。

Amazon SQS の standard queue では、MessageGroupId を付けると FIFO の順序制御ではなく fair queues のテナント識別子として扱われる。高ボリュームの tenant / client / request type が同じキューに大量メッセージを積んでも、SQS が受信順を調整して他グループの message dwell time を低く保とうとする。

適用条件

• 1 つの standard queue を複数テナント・顧客・リクエスト種別で共有している。
• 一部のグループだけ大量投入または重い処理を行い、他グループの短時間処理が待たされる noisy neighbor 問題を避けたい。
• 厳密な順序保証は不要で、高スループットと低待機時間を優先する。

使い分け

• standard queue + MessageGroupId: テナント間の公平性を改善する。consumer 側の変更は不要で、同じ MessageGroupId のメッセージも並列処理され得る。
• FIFO queue + MessageGroupId: 同一グループ内を直列化し順序を守る。head-of-line blocking とスループット制限を受けやすい。
• fair queues は per-tenant の実行数上限ではない。重いジョブが worker 枠を占有する問題には、処理サイズ別レーン、予約済み worker 枠、chunk 化、アプリ側の同時実行制御を併用する。

検証

負荷テストでは、1 グループに大量メッセージを投入した状態で別グループの小さいメッセージを投入し、CloudWatch の fair queues 関連メトリクスや、送信から受信までの dwell time が小さいまま保たれるかを見る。

バックグラウンドのバルク処理ワーカーで「全レコードの更新を1つの DB トランザクションで囲んでアトミックにする」と整合性が上がるように見えるが、そのトランザクションの内側で実行される非トランザクショナルな副作用は DB のロールバック対象外であり、途中失敗時に DB だけ巻き戻って外部状態が残る不整合を生む。

非トランザクショナルな副作用の例

• 検索インデックス（OpenSearch/Elasticsearch 等）への upsert/delete
• 外部 API 連携（カレンダー、会議、決済、通知など）
• 別コネクション/別サービスへの書き込み

なぜ危険が増幅されるか（単一更新との差）

• 単一レコード更新でも同じ構造はあるが、巻き込まれるのは1件。バルクでは1件の失敗で数千〜数万件分の DB がロールバックされる一方、それまでに反映済みの外部副作用は全部残る。
• さらにキュー消費ワーカーが「失敗時もメッセージを無条件削除」していると、リトライで再実行されて外部状態が正しい値に上書き収束する余地も消え、恒久不整合になる。

判断基準・対処

• トランザクション内では DB 書込のみに限定し、検索インデックス更新・外部 API 連携はコミット後の別フェーズに出す（commit→外部反映の順）。
• 外部反映は冪等化し、失敗は記録して後追い再実行できるようにする（outbox / 失敗レコード化）。
• トランザクション境界を全件で1つにせず chunk 単位で切ると、ロールバック範囲と保持時間を限定でき、中断後再開もしやすい。
• 「失敗=即終了・リトライなし」を選ぶなら、その前提と外部副作用の不整合可能性を設計として明示する。一時障害（インデックスの 429、デッドロック、lock wait timeout）でリトライしたいなら、失敗時はメッセージを削除せず再配信/DLQ 経路を残す。

検証

処理の途中（外部副作用を出した後）で意図的に例外を投げ、(1) DB がロールバックされること、(2) 外部システム（検索インデックス・外部 API）に部分反映が残ること、(3) リトライ経路があるなら再実行で収束し、無いなら不整合が残ることを確認する。

大量の ID 配列を JSON request body で受ける API では、OpenAPI やバリデーションの maxItems だけを上げても不十分。実際に通るかは、アプリの body parser limit、proxy / gateway / WAF の body 処理、保存先 payload のサイズ上限、非同期キューへ載せるメッセージサイズのすべてで決まる。

判断基準

• まず代表的な ID 長と最大件数で JSON.stringify 後の byte size を見積もる。数 MB に収まるなら通常の request body として扱いやすいが、既定の 1MB 制限では落ちることが多い。
• body parser limit は、想定 payload サイズに更新値やメタデータ分の余裕を足して設定する。上限だけを大きくしすぎると全 API のメモリ使用量と DoS 面が広がるため、可能なら対象 endpoint の性質に合わせて限定する。
• WAF が body を検査する構成では、巨大 JSON や自由入力値が body ルールに触れる可能性を確認する。ファイルや HTML でなくても、長文・URL・記号を含む更新値を受ける endpoint は誤検知緩和対象になり得る。
• 非同期処理に渡す場合、キューには巨大な ID 配列を直接載せず、task ID や payload 参照だけを送る。キューの message size 上限と再試行コストを避けられる。

検証

最大件数の payload を実際に生成して byte size を測り、body parser の limit 未満であることを確認する。境界テストでは最大件数ちょうどが通り、最大件数 + 1 がバリデーションで落ちることを確認する。WAF 配下では、自由入力値を含む最大級 payload がアプリまで届くかも疎通確認する。

排他ロックや処理中状態を持つ長時間キューワーカーでは、アプリ側の heartbeat だけでは正常実行中のジョブを守り切れない。メッセージブローカー側の visibility timeout が切れると、元 worker が生きていても同じ message が再配信され、receive count が進む。redrive 上限が小さいと、正常実行中の message が DLQ に落ち、元 worker がその後落ちたときに復旧用の message が失われる。

判断基準

• 1 回の処理が visibility timeout を超え得る worker は、処理中に現在の receipt handle / delivery token へ visibility を定期延長する。
• DB lock や task heartbeat は「元 worker が生きているか」を判断する材料であり、message の redrive count を止める材料ではない。両方が必要。
• 処理中状態の message が再配信され、既存 heartbeat が新しい場合は、active job と判断して失敗扱いにしない。ただし message を何もせず放置すると receive count が進むため、その再配信分の visibility も延長する。
• active job の再配信 message を削除すると、元 worker が後で落ちたときに再開・失敗確定の入口を失う。削除ではなく visibility 延長を選ぶ。
• visibility timeout や redrive 上限を大きくするだけでは、処理時間上限を仕様化しない限り根本対策になりにくく、障害検知も遅れる。

検証

visibility timeout より長い正常処理を模擬し、処理中に visibility 延長 API が定期実行されることを確認する。さらに、処理中状態の message を再配信として受けたケースで、job を失敗扱いせず、message も削除せず、visibility だけ延長することを確認する。

キューワーカーで処理対象の状態が COMPLETED / FAILED などの終端状態になっていても、関連する排他ロック、lease、外部リソース解放などの cleanup が別操作なら、message を helper / parser 側で即削除してはいけない。task は終端でも cleanup は未完了の可能性があるため、terminal message は cleanup recovery の入口として worker に渡す。

判断基準

• 状態更新と cleanup が別操作で、cleanup 失敗時に message を残す設計なら、terminal status も worker に渡す。
• worker は terminal message を受けたら、通常処理や user / role の再構築より前に cleanup を実行し、cleanup 成功後にだけ message を削除する。
• cleanup 失敗時は message を削除しない。再配信で同じ cleanup を再試行できるようにする。
• helper / parser 側で削除してよいのは、対象が存在しない、payload が壊れているなど、再配信しても cleanup できない message に限定する。

失敗症状

処理本体が完了して task が terminal になった後、lock 解放だけが一時障害で失敗する。その後、再配信された terminal message を parser が「処理不要」と判断して削除すると、cleanup の再実行機会が消え、lock や lease が残り続ける。

検証

terminal status の message を受信したケースで、worker が cleanup を実行してから message を削除することを確認する。cleanup を意図的に失敗させたケースでは、message delete が呼ばれず、再配信で cleanup retry できることを確認する。

結果集合を1ページずつ取得しながら同じレコード群を更新していく一括処理（検索条件で全件選択して更新する等）では、ページング方式の選択が取りこぼし・二重処理を左右する。一覧の純粋な読み取りでは表面化しないが、「読みながら同じ集合を変える」走査では致命的になる。

なぜ offset / ページ番号方式は破綻するか

• 件数スキップ方式は「現在の結果集合の N 件目以降」を毎ページ取り直す。更新によって対象が検索条件から外れて結果集合が縮むと、後続レコードの位置が前方へ詰まり、スキップ境界がずれて未処理レコードを飛ばす（逆に対象が増える方向なら二重処理）。
• つまり走査と変更が同じ集合に対して同時に起きると、絶対位置ベースのページングは前提が崩れる。

keyset / カーソル方式が安全な理由と成立条件

• 直前ページ最終要素のソート値をカーソルにして「そのソート値より後」を取る方式（searchAfter / keyset）は、更新で対象が結果集合から外れても残りレコードのカーソル相対位置が動かないため取りこぼさない。更新済みで条件から外れたレコードはカーソル以前に位置するので再出現もしない。
• 成立条件は「ソートキーが安定（一意で順序が決定的）かつ、その走査で更新する対象外のフィールドであること」。ソートキー自体や検索条件に使うフィールドを更新値に含めると保証が壊れ、更新でレコードがカーソルを飛び越えてスキップ／重複が起きうる。
• 一意性を担保するため、ソートは業務キーにレコード ID 等の tie-breaker を足した複合キーにする。

スナップショット方式との判断

• 「作成時点で対象 ID を全件展開してスナップショット化し、固定リストを chunk 処理する」方式は走査中の変動を完全に排除できるが、ID 保存コスト・作成時の展開負荷・スナップショットの陳腐化を抱える。
• 「処理時にライブ再検索＋keyset」方式はスナップショット不要だが、上記のソートキー不変条件と、検索エンジンの refresh 遅延（更新直後は反映前の値が見えうる）を前提に設計する。保存コスト／対象変動の許容度／件数規模で選ぶ。

検証

ソート境界をまたぐ件数（chunk サイズの数倍）で走査しながら、全件が漏れなく1回ずつ更新されること、さらにソートキーやフィルタ対象フィールドを更新値に含めた場合に取りこぼし／重複が再現することを実データ・実検索エンジンで確認する。固定ページを返す mock ではこの不変条件を検出できない。

heartbeat/TTL で「期限切れロックを別アクターが奪取する」復旧機構を入れたシステムで、別の経路が「ロックを保持したまま非同期に待機する」（典型: ワーカーが transient 失敗時に処理を諦めず、メッセージの可視性タイムアウト経過後の再配信を待つ間ロックを握り続ける）と、両者が競合して同一資源の二重実行が起きる。これは『リトライ枯渇時にロックを能動解放する』とは別の、復旧機構導入によって新たに生まれる競合である。

なぜ起きるか

• 保持中はワーカーが能動的に何もしないため heartbeat 更新を止めがち。すると保持しているロックが自分の stale 閾値を超え、自分が入れた奪取復旧機構によって他アクターに奪われる。
• 奪われた後に当初の待機がタイムアウトして再開すると、状態が「未処理（PENDING 相当）」に戻っているため再取得チェックを通過し、奪取した側と並走する。排他不変条件（同一テナント/資源で 1 ジョブ）が破れ、二重更新やインデックス不整合に直結する。
• 競合窓は「最後の heartbeat から stale 閾値経過」〜「待機タイムアウトでの再開」まで。stale 閾値 < 待機窓 だと必ず空く。

判断基準（いずれか）

• リースの寿命（stale 閾値）を、意図的な保持/再配信窓（可視性タイムアウト等）より十分長く順序づける。各定数を別々に決めず「stale 閾値 > 再配信間隔」という関係自体を不変条件として持つ。
• 保持中もリースを延命し続ける（待機の裏で heartbeat/期限を更新）。
• 再開・再取得の直後に「ロックがまだ自分の所有か」を条件付き更新の影響行数等で再確認し、奪われていれば中断する（防御的で最も堅牢）。
• そもそも保持せず解放し、再開側が取り直す設計にできるなら競合面を消せる。

検証

transient 失敗で待機保持に入った状態で stale 閾値ぶんだけ時間を進め、同一資源への別 acquire が成功してしまわないこと（あるいは再開側が所有権喪失を検知して中断すること）を確認する。逐次の単体検証では窓が再現しにくいため、閾値と待機タイムアウトの大小関係そのものをテストで固定する。

共通の排他ロックで複数種類の worker やジョブを直列化する設計では、ジョブ作成時の lock 取得だけでは不十分。stale lock の奪取、キューメッセージの再配信、worker の異常終了があると、古い message が後から処理を再開し、新しい所有者と並行実行する可能性がある。

判断基準

• 共有 lock を排他の正本にするなら、すべての処理主体が同じ lock を見る。片方の worker だけ task status を見る、別の worker だけ lock を見る、という非対称な正本を作らない。
• worker は処理開始時に、lock が自分の job ID / owner ID を指していることを条件付き更新や heartbeat 更新の結果で確認する。
• 長時間処理では、処理中も heartbeat を更新し、chunk 境界など副作用のまとまりごとに所有権を再確認する。所有権を失っていたら後続処理を止める。
• stale lock を別ジョブが奪取できる設計では、古い message の再配信を正常系として扱う。再配信側が lock 所有権を持っていなければ、処理本体へ進ませない。

落とし穴

ジョブ作成時の排他チェックだけを見て「同時実行しない」と判断すると、再配信や stale 復旧の経路で不変条件が破れる。とくに複数種類の worker が同じ資源を扱う場合、片方だけに所有権チェックや heartbeat を実装すると、古い経路が新しい lock 所有者を無視して動けてしまう。

検証

古い worker の message を再配信させる前に、同じ資源の lock を別 job に奪取させる。再配信された worker が処理本体に入らず、所有権喪失として停止することを確認する。処理中に lock 所有者を差し替えた場合も、次の chunk 境界で停止することを確認する。