メッセージキューを用いた非同期システムで、「run-id」を使って処理の疎通を確認する様子を示す図です。入力からプロデューサー、キュー、コンシューマー、データベースまで、一意のIDが追跡される流れをキャラクターが指し示しています。

はじめに

非同期処理を組み込んだシステムでは、HTTPレスポンスが正常に返ってきていても、その先のキュー処理が実は詰まっている、という状況が起こり得ます。特にメッセージキューを挟んだ構成では、「リクエストは受け付けられたが、後続の処理まで届いているかどうか」を継続的に確認する仕組みがないまま本番運用に入ってしまいがちです。

ある開発現場では、本番向けのスモークテストにHTTP経路の確認しか含まれておらず、「キューにメッセージは積まれているのに、consumer側で処理が止まっている」という障害を検知できない、という課題がありました。本稿では、この課題を「run-idトレーシング」という手法で解決したアプローチを、汎用的なパターンとして紹介します。

こんな人におすすめ

メッセージキューを使った非同期処理の疎通確認をどう自動化すればよいか悩んでいる方
本番スモークテストの対象をHTTP経路だけで止めてしまっている方
リトライやDLQ再送を前提にした冪等な書き込み処理を設計したい方
ESM（.mjs）スクリプトとTypeScriptのコードを安全に共有したい方

本論

run-idを目印にしてキューの通過を追跡する

非同期処理のE2E確認で難しいのは、「どのリクエストがどこまで届いたか」を紐づける手段がないことです。そこで、リクエストの中に一意のIDを埋め込み、そのIDがproducerからqueue、consumerまで通過したかをDBで確認する、という方法を取ります。

具体的な流れは次の通りです。

スモーク側が操作ログ作成リクエストのメモ欄に [smoke-test <run-id>] という形式でIDを埋め込む
producer側がメモからIDを正規表現で抽出し、キューメッセージに付与する
consumer側がメッセージを処理したタイミングで、専用テーブルにIDを冪等INSERTする
スモーク側がヘルスチェックAPIをポーリングし、IDに対応する記録が現れるのを待つ

このパターンは特定のキューサービスに依存しないため、SQSやPub/Subなど他のメッセージキューでも同じ考え方で応用できます。

// 共有モジュール: run-id抽出ユーティリティ
export const SMOKE_MEMO_RUN_ID_REGEX_SOURCE = '\\[smoke-test ([0-9a-f]{8,32})\\]'
const SMOKE_MEMO_RUN_ID_REGEX = new RegExp(SMOKE_MEMO_RUN_ID_REGEX_SOURCE)

export function extractSmokeRunId(memo: unknown): string | null {
  if (typeof memo !== 'string' || memo.length === 0) return null
  const match = SMOKE_MEMO_RUN_ID_REGEX.exec(memo)
  return match ? match[1] : null
}

型を unknown にしているのは、Workerランタイム側とNode.jsスクリプト側の両方から呼ばれる可能性があるためです。呼び出し元の型が違っても、この関数内でnullチェックまで完結させることで、利用側の実装を単純に保てます。

冪等INSERTでリトライやDLQ再送に耐える

キューはリトライやDead Letter Queueからの再送が起こる前提で設計する必要があります。同じメッセージが複数回consumerに渡ってきても記録が壊れないようにするには、ON CONFLICT DO NOTHING を使った冪等INSERTが有効です。

export async function recordQueueSmokePing(db: AppDatabase, runId: string): Promise<void> {
  try {
    await db
      .prepare(
        `INSERT INTO queue_smoke_pings (run_id) VALUES (?)
         ON CONFLICT (run_id) DO NOTHING`,
      )
      .bind(runId)
      .run()
  } catch (error) {
    const message = error instanceof Error ? error.message : String(error)
    console.warn(`[queue-smoke] failed to record ping runId=${runId}: ${message}`)
  }
}

ポイントは DO UPDATE ではなく DO NOTHING を選んでいることです。最初にINSERTされた時刻を「初回配送時刻」として保持したい場合、再送のたびに上書きされては困ります。加えて、この処理全体をtry-catchで囲み、失敗時は例外を投げずに警告ログへ留めています。マイグレーション未適用の環境でテーブルが存在しない場合でも、スモークスクリプト全体がクラッシュしないようにするための実戦的な配慮です。

ポーリングにリトライ回数と間隔を持たせる

疎通確認の最後は、記録が現れるまで待つポーリング処理です。固定回数・固定間隔で終わらせず、環境変数で調整できるようにしておくと、CIの実行時間制約や本番でのトリアージ作業に対応しやすくなります。

export async function runQueueHealthCheck({
  baseUrl, runId, authHeaders,
  pollMaxAttempts = 5,
  pollIntervalMs = 3000,
  sleep = delay,
}) {
  let lastResult = null
  for (let attempt = 0; attempt < pollMaxAttempts; attempt += 1) {
    if (attempt > 0) await sleep(pollIntervalMs)
    lastResult = await smokeFetch('queue.health', `${baseUrl}/api/queue/health?runId=${runId}`, {
      headers: authHeaders,
      validateBody: buildValidateQueueHealthDelivered(runId),
    })
    if (lastResult.ok) return lastResult
  }
  return lastResult
}

sleep を引数として差し込めるようにしているのは、テスト時に即時解決する関数へ差し替えるためです。本番相当のリトライロジックはそのままに、テスト実行時間だけを短縮できます。また、最大試行回数と間隔を環境変数で上書きできるようにしておくと、「これはflakeなのか、本当に壊れているのか」を切り分けたいときに、試行回数を増やして再実行する、という運用がしやすくなります。

ESMスクリプトとTypeScriptでロジックを共有する

Node.jsの .mjs スクリプトからは、TypeScriptのモジュールを直接importできません。同じ正規表現を両側で使いたい場合、正規表現の文字列そのものを定数としてexportし、利用側で new RegExp(...) に組み立て直す、という回避策が使えます。コンパイルを必要としない文字列だからこそ成立する方法です。

つまづきやすいポイント

キューサービスによっては、滞留メッセージ数などの内部状態をAPIで取得できない場合があります。「絶対量を監視する」のではなく「次の操作で生存確認できる」程度の割り切りが必要になることがあります
冪等INSERT用のテーブルは、放っておくとレコードが増え続けます。定期的なクリーンアップの仕組みは別途用意しておく必要があります
ポーリングの最大試行回数と間隔は、CIのタイムアウト設定と整合させておかないと、SLOの余裕枠を圧迫してしまいます
ヘルスチェック用エンドポイントを外部公開する場合、認証をどこまで厳格にするかは別途検討が必要です

まとめ

非同期処理を含むシステムのE2E確認は、HTTP経路の確認だけでは不十分になりがちです。リクエストに一意のIDを埋め込み、そのIDがproducerからconsumerまで通過したかをDBで確認する「run-idトレーシング」は、キューサービスの種類を問わず応用できる考え方です。冪等INSERTやリトライ設定の環境変数化とあわせて導入することで、本番運用に耐えるスモークテストを組み立てやすくなります。

カテゴリー: 生成AI

タグ: CI/CD E2Eテスト TypeScript メッセージキュー冪等性

メッセージキューの疎通確認を「run-idトレーシング」で実現する