Azure Functionsの設計：トリガー選定と失敗例（Queue / HTTP / Event Grid）

Last updated at 2026-07-07Posted at 2026-07-07

Azure Functionsの設計：トリガー選定と失敗例（Queue / HTTP / Event Grid）

この記事は 2026-06-19 時点の Microsoft 公式ドキュメントを前提にしています。
現場エピソードは、守秘義務に配慮して複数案件で起こりがちな事故を匿名化・再構成したものです。

はじめに

Azure Functions の設計で一番危ないのは、

「とりあえず HTTP Trigger で受ければいい」
「Blob が来たから Event Grid Trigger で直接処理すればいい」
「非同期にしたいから Queue Trigger に入れれば安全」

という雑なトリガー選定です。

Azure Functions のトリガーは、単なる起動方法ではありません。

リトライ、タイムアウト、重複、順序、監視、権限、障害時の責任範囲を決める設計上の契約です。

Azure Functions の関数は必ず 1 つのトリガーを持ちます。入力バインドや出力バインドは組み合わせられますが、起動契機としてのトリガーは 1 つです。

つまり、HTTP / Queue / Event Grid のどれを選ぶかは、後から軽く差し替えられる実装詳細ではなく、アーキテクチャ判断です。

この記事で扱うこと

この記事では、Azure Functions でよく使う以下の 3 種類のトリガーについて、設計観点と失敗例を整理します。

HTTP Trigger
Queue Trigger
Event Grid Trigger

特に、現場で本当に事故りやすい以下の論点を中心に書きます。

HTTP Trigger で長時間処理をやってはいけない理由
HTTP の Idempotency-Key を「受け取るだけ」では不十分な理由
Idempotency-Key は tenant / endpoint / user / request hash でスコープ化すべき理由
request hash を raw body の Trim() だけで作る危険性
RequestedBy をリクエスト本文から信用してはいけない理由
HTTP 受付と Queue 投入を 1 トランザクションにできない問題
Outbox パターンで Queue 送信の曖昧状態を吸収する設計
Queue SDK の送信呼び出し中に例外が発生した場合の DispatchUnknown
Queue Trigger にしただけでは安全にならない理由
Queue Trigger の冪等性を atomic に設計すべき理由
CurrentlyProcessing を正常終了扱いにすると Queue メッセージを消してしまう事故
ProcessAsync 成功後に MarkCompletedAsync が失敗した場合の再実行リスク
poison queue を「業務エラー」と決めつけてはいけない理由
AzureWebJobsStorage と業務 Queue 用 Storage を分けるべき理由
Event Grid Trigger で重い処理を直接やると燃える理由
Event Grid の HTTP 配信仕様と [EventGridTrigger] の実装モデルの違い
Event Grid の eventId だけでは業務的な重複排除として弱い場合
Blob URL をそのまま受け取る設計の危険性
Blob の ETag / VersionId 欠落時の扱い
Function App の分け方
監視設計

先に結論

現場では、だいたいこの考え方で分けると安定します。

トリガー	使う場面	向いていない場面	現場での一言
HTTP Trigger	API、Webhook、同期的な受付、ステータス照会	長時間処理、大量バッチ、リトライ前提の処理	「結果を返す」より「受け付ける」入口にする
Queue Trigger	非同期ジョブ、負荷平準化、再試行したい処理	厳密な順序保証、重複排除、巨大メッセージ	業務処理の本丸。冪等性がないと燃える
Event Grid Trigger	Azure リソースや外部イベントへの反応	重い処理、コマンド処理、結果応答が必要な処理	イベントを受けたら、基本はすぐ Queue に逃がす

一言でまとめるなら、こうです。

HTTP は受付、Queue は実務、Event Grid は通知。

多少単純化していますが、責務分離の合言葉としてはかなり使いやすいです。

判断フロー

トリガー選定時は、まず処理の起点と応答責任を確認します。

安定しやすい基本構成

私が現場でよく採用する構成は、かなりの確率でこの形です。

HTTP Trigger と Event Grid Trigger は、できるだけ軽くします。

重い処理、時間の読めない処理、失敗時に再試行したい処理は Queue Trigger に寄せます。

重要なのは、HTTP Trigger や Event Grid Trigger が直接 Queue に積むのではなく、Job と Outbox を同一 DB トランザクションで保存することです。

これにより、次の曖昧状態を管理できます。

Job は受付済み
でも Queue 送信がまだ成功したか分からない

この状態を設計していないと、サーバーレスというより「静かに詰まる分散システム」になります。

1. HTTP Trigger：一番便利で、一番事故る

HTTP Trigger が向いている処理

HTTP Trigger は、HTTP リクエストで関数を起動するトリガーです。API や Webhook の実装に向いています。

向いている処理は、たとえば以下です。

処理	向き不向き
入力チェック	向いている
認証・認可	向いている
ジョブ受付	向いている
ステータス照会	向いている
数秒で終わる軽い処理	条件付きで向いている
画像変換	向いていない
CSV 大量取込	向いていない
外部 API への大量連携	向いていない
数分かかる処理	同期 HTTP ではやらない

HTTP Trigger で重要なのは、HTTP 応答を返す責任があるということです。

Function App のタイムアウト設定を長くしても、HTTP Trigger が HTTP 応答を返せる時間には制限があります。

Azure Functions の HTTP Trigger は、230 秒以内に完了しないと Azure Load Balancer 側でタイムアウトし、HTTP 502 などが返ることがあります。しかもこの場合、関数の実行自体は裏で続く可能性があります。

ここが怖いところです。

呼び出し元から見ると失敗。
でもサーバー側の業務処理は途中まで進んでいる。

失敗例：HTTP Trigger で CSV 取込を同期処理した話

ある業務システムで、管理画面から CSV をアップロードして、数万件の取引データを取り込む機能がありました。

最初の設計はこうでした。

開発環境では 100 行程度の CSV しか使っていなかったので、問題なく見えました。

UAT で 3,000 行を流したときも、少し遅いけれど何とか動きました。

そして本番初日。

営業部門が 12 万行の CSV を投げました。

結果はこうです。

ブラウザは待ち続ける
230 秒付近で HTTP 502
でも Function の処理自体は裏で続いている
ユーザーは「失敗した」と思って再実行
同じ CSV が複数回投入される
DB に重複データ
外部 API にも重複連携
夜に経理から電話

この手の事故で怖いのは、HTTP 的には失敗しているが、業務処理は途中まで成功していることです。

オンプレ時代の感覚で、

「エラーが返ったなら、全部ロールバックされているはず」

と思っていると、まず見誤ります。

Azure Functions は HTTP リクエストの成否と、裏側の業務処理の成否がきれいに一致するとは限りません。

改善後の設計

この案件では、最終的にこう直しました。

HTTP Trigger は、次のことだけをします。

認証・認可
入力サイズと JSON の検証
Blob 参照の検証
必要なら Blob の存在、ETag、ContentLength、ContentType の確認
認証済み principal から tenantId / userId を取得
Idempotency-Key、API 名、tenant、user、request hash から受付キーを決める
Job と Outbox を同一 DB トランザクションで保存する
202 Accepted と jobId を返す

HTTP のレスポンス例です。

{
  "jobId": "imp_6d5f4c3a2b1e...",
  "statusUrl": "/api/imports/imp_6d5f4c3a2b1e..."
}

実処理は Queue Trigger に任せます。

HTTP の Idempotency-Key は「受け取るだけ」では不十分

HTTP API で Idempotency-Key を受け取るだけでは、まだ弱いです。

悪い例です。

1. HTTP Trigger が Idempotency-Key を jobId として使う
2. 毎回 Queue にメッセージを積む
3. Worker 側で重複を無視する

この設計でも Worker 側が冪等なら最終結果は守れます。

しかし、同じ Idempotency-Key でリクエストが再送されるたびに Queue に同じ jobId のメッセージが積まれます。

その結果、以下が起こります。

Queue のメッセージ数が無駄に増える
Worker が重複メッセージを何度も拾う
監視上、ノイズが増える
外部 API や DB の処理直前まで進むと危ない
障害時の原因切り分けが難しくなる

望ましいのは、HTTP 入口の時点で重複受付を抑えることです。

HTTP Trigger:
  tenantId + endpointName + authenticatedUserId + Idempotency-Key を受付スコープにする
  request body を正規化して hash を保存する
  同じスコープ・同じ key・同じ request hash なら既存 jobId を返す
  同じ key で request hash が違うなら 409 Conflict を返す
  新規受付の場合だけ Job + Outbox を atomic に作成する

つまり、HTTP Trigger 側でも atomic な重複受付抑止を入れます。

Idempotency-Key はスコープ付きで扱う

Idempotency-Key はグローバル一意とは限りません。

たとえば、別テナント、別ユーザー、別 API で偶然同じ Idempotency-Key が使われることはあり得ます。

そのため、実務では次のようにスコープを切ります。

idempotency scope =
  tenantId + endpointName + authenticatedUserId + idempotencyKey

さらに、同じ Idempotency-Key で異なるリクエスト本文が来た場合も検知します。

request fingerprint =
  canonical request body hash

受付レコードには、最低限以下を持たせます。

項目	用途
`jobId`	業務処理の識別子
`tenantId`	テナント境界
`endpointName`	API 境界
`requestedByUserId`	認証済みユーザー
`idempotencyKeyHash`	生の key を保存しないため
`requestBodyHash`	同じ key で異なる本文を検知するため
`status`	`Accepted` / `OutboxPending` / `Processing` / `Completed` / `Failed` など

同じ tenantId + endpointName + authenticatedUserId + idempotencyKeyHash で、requestBodyHash が同じなら既存 jobId を返します。

同じ key で本文が違うなら、既存 job を返すのではなく、409 Conflict などで不整合として扱います。

request hash は raw body の Trim だけでは弱い

説明用のサンプルでは、ついこう書きたくなります。

var requestBodyHash = Sha256Hex(rawBody.Trim());

ただし、これは実務では false conflict を起こしやすいです。

たとえば、以下の 2 つは意味として同じでも、raw 文字列は違います。

{"containerName":"imports","blobName":"tenant-a/file.csv"}

{
  "blobName": "tenant-a/file.csv",
  "containerName": "imports"
}

raw body をそのまま hash 化すると、プロパティ順序や空白差異で別リクエスト扱いになります。

実務では、次を決めたうえで hash 化します。

観点	方針例
JSON property order	DTO に deserialize して canonical JSON を生成する
余分な property	拒否する、または明示的に無視する
大文字小文字	`JsonSerializerDefaults.Web` などで方針を固定する
null と省略	同一扱いにするか、別扱いにするか決める
数値・日付	表現揺れを吸収するか決める

この記事のサンプルでは、DTO から必要な項目だけを安定した順序で再構成し、その canonical JSON を hash 化します。

RequestedBy をリクエスト本文から信用しない

これも現場で地味に危ないです。

たとえば、リクエスト本文にこういう項目を持たせる設計です。

{
  "requestedBy": "admin@example.com"
}

この値をそのまま監査ログや Queue メッセージに入れると、クライアントが好きなユーザーを名乗れてしまいます。

{
  "requestedBy": "ceo@example.com"
}

これで監査ログ上は CEO が実行したように見える、という事故が起きます。

RequestedBy は、クライアントが送る任意文字列を信用せず、Entra ID、APIM、JWT、Easy Auth などで検証済みの principal / claim から決定します。

リクエスト本文の requestedBy は、表示用の補助情報に留めるか、そもそも受け取らない方が安全です。

また、HTTP Trigger の AuthorizationLevel.Function は、Function Key による呼び出し制御です。

Function Key は、利用者本人のユーザー認証や業務認可の代替ではありません。

業務ユーザーの認証・認可は、Entra ID、APIM、Easy Auth、JWT 検証、アプリ側のロール判定などで別途設計します。

受付レコードと Queue 投入の注意点

ここで実務上の落とし穴があります。

受付レコードを DB に作る処理と、Storage Queue にメッセージを送る処理は、通常 1 つのトランザクションにはできません。

危ない実装例です。

try
{
    await _queue.SendMessageAsync(...);
    await _jobs.MarkEnqueuedAsync(jobId, cancellationToken);
}
catch (Exception ex)
{
    await _jobs.MarkEnqueueFailedAsync(jobId, ex.Message, cancellationToken);
    throw;
}

この実装は一見よさそうですが、次の曖昧状態を作ります。

1. DB に受付レコードを作成する
2. Queue 送信は成功する
3. MarkEnqueuedAsync が DB タイムアウトで失敗する
4. catch に入り MarkEnqueueFailedAsync を呼ぶ
5. HTTP は 500 を返す
6. しかし Queue にはメッセージが存在する
7. Worker が拾う
8. DB 状態は Enqueued ではなく EnqueueFailed / EnqueuePending のまま
9. 処理権取得が失敗する、または再 enqueue と重複する

Queue 送信と MarkEnqueuedAsync は atomic ではありません。

SendMessageAsync が成功した後に MarkEnqueuedAsync が失敗すると、**「Queue には存在するが DB は Enqueued ではない」**という曖昧状態になります。

この状態を EnqueueFailed と断定すると、実際には Queue にメッセージがあるのに「未送信」と誤認します。

そのため、実務では Outbox パターンを第一候補にします。

Outbox パターンで考える

Outbox パターンでは、HTTP Trigger は Queue に直接送信しません。

HTTP Trigger:
  DB トランザクション内で Job と Outbox を保存する
  202 Accepted を返す

Outbox Dispatcher:
  Outbox を読み取る
  Queue に送信する
  Outbox を Dispatched にする
  失敗時は retry_count / next_retry_at で再送制御する

この設計のポイントは、DB 側の状態を正とすることです。

Outbox Dispatcher でも、Queue 送信成功後に MarkDispatched が失敗する可能性はあります。

この場合は、DispatchFailed と断定せず、DispatchUnknown のような状態にします。

さらに重要なのは、Queue SDK の送信呼び出し中に例外が発生した場合も、送信されていないとは断定できないことです。

ネットワークタイムアウト、応答受信前の切断、クライアント側キャンセルなどでは、クライアントから見ると SendMessageAsync は失敗でも、サービス側には Queue メッセージが作成されている可能性があります。

そのため、原則はこうです。

Queue SDK 呼び出し前に失敗:
  DispatchFailed / retryable

Queue SDK 呼び出し中、または呼び出し後に失敗:
  DispatchUnknown
  再送されても Worker 側の jobId 冪等性で吸収する

Outbox パターンでは、Queue への二重送信は起こり得ます。

しかし、それは設計上許容します。

その代わり、Queue Worker 側は jobId で atomic に処理権を取り、重複メッセージを安全に無視できるようにします。

BlobUrl をそのまま受け取る設計は危険

サンプル記事や PoC では、ついこう書きたくなります。

{
  "blobUrl": "https://example.blob.core.windows.net/imports/file.csv"
}

しかし、本番で任意の BlobUrl をクライアントから受け取り、そのまま Worker に渡す設計は危険です。

理由は次です。

意図しない外部 URL を参照する可能性がある
許可していない Storage Account / Container を読みに行く可能性がある
SAS URL がログに出て長期露出する可能性がある
URL に権限情報が混ざると監査しにくい
権限境界が URL 文字列に依存してしまう
Worker の Managed Identity 設計が曖昧になる

本番では、次のように設計する方が安全です。

{
  "storageAccount": "stexampleprod",
  "containerName": "imports",
  "blobName": "tenant-a/2026/06/19/file.csv",
  "etag": "\"0x8DC....\"",
  "versionId": "2026-06-19T01:23:45.0000000Z"
}

ポイントは以下です。

任意の URL ではなく、許可された Storage Account / Container / Prefix のみ処理する
Queue には URL ではなく storageAccount / containerName / blobName / etag / versionId を入れる
Worker は Managed Identity で Blob を読む
SAS URL を Queue やログに残さない
etag や versionId を持たせ、処理対象のすり替わりを検知できるようにする

Blob 名は case-sensitive です。

そのため、テナント境界や業務境界を prefix で切るなら、StringComparison.OrdinalIgnoreCase ではなく StringComparison.Ordinal で判定します。

HTTP 入力サイズと Blob 存在確認

この記事の構成では、HTTP Function に巨大 CSV 本体を直接流し込みません。

HTTP では Blob 参照だけを受け取り、巨大ファイルは Blob Storage に事前アップロードする設計にします。

ただし、Blob 参照だけを受け取る場合でも、攻撃やクライアントバグで巨大な request body が送られる可能性があります。

そのため、HTTP Trigger では request body size の上限を設けます。

HTTP Trigger:
  request body は最大 32 KiB などに制限する
  巨大 CSV 本体は直接受け取らない
  Blob 参照だけ受け取る

Blob 参照の検証は、受付時と Worker 実行時の両方で行うのが安全です。

タイミング	目的
HTTP 受付時	早期 reject、ユーザーに即時エラーを返す
Worker 実行時	実際に読む直前のすり替わり防止、`ETag` / `VersionId` の再確認

受付時に Blob properties を確認しても、Worker 実行時までに Blob が差し替わる可能性があります。

そのため、ETag または VersionId を Queue メッセージに入れ、Worker は読み取り時にも一致確認します。

HTTP Trigger の設計ルール

HTTP Trigger では、以下をルールにしています。

ルール	理由
長時間処理を同期でやらない	230 秒制限に当たる
重い処理は Queue に逃がす	負荷平準化とリトライ制御のため
`202 Accepted` を返す	「処理完了」ではなく「受付完了」を明示する
request body size を制限する	巨大入力や攻撃を防ぐ
`Idempotency-Key` を必須にする	クライアント再送に備える
`Idempotency-Key` をスコープ付きで扱う	テナント・API・ユーザー間の衝突を避ける
request hash を canonical 化する	空白や property order 差異による誤判定を避ける
`jobId` に一意制約を置く	HTTP 入口で二重受付を抑止する
Job + Outbox を atomic に保存する	Queue 投入前後の曖昧状態を避ける
`RequestedBy` は principal から決める	なりすまし監査ログを防ぐ
任意の Blob URL を受け取らない	セキュリティ境界を明確にする
Blob properties を検証する	存在確認、サイズ制限、型確認、すり替わり防止
認証レベルを雑に `Anonymous` にしない	Web 公開 API 化しやすいため
Function Key を業務認可の代替にしない	Function Key はユーザー本人性を表さないため

2. Queue Trigger：非同期処理の王道。ただし冪等性がないと地獄

Queue Trigger が向いている処理

Queue Trigger は、Azure Queue Storage にメッセージが追加されたときに関数を起動します。

非同期処理、負荷平準化、バックグラウンドジョブに向いています。

たとえば以下です。

CSV 取込
メール送信
PDF 生成
画像変換
外部 API 連携
再試行したい業務処理
ユーザーを待たせたくない処理

Queue Trigger の強みは、処理が失敗したときの再試行が組み込まれていることです。

既定では、Queue Trigger の関数が失敗すると、同じメッセージに対して最大 5 回、初回実行を含めて処理が試行されます。

すべて失敗すると、以下の名前の poison queue にメッセージが移動されます。

<元のキュー名>-poison

たとえば、元のキューが import-jobs なら、失敗し続けたメッセージは以下に入ります。

import-jobs-poison

poison queue は「業務エラー」とは限らない

ここは強調しておきたいです。

poison queue に入ったメッセージを、いきなり「ユーザー影響のある業務エラー」と決めつけるのは危険です。

xxx-poison に入ったメッセージは、Functions が規定回数処理に失敗した未処理メッセージです。

原因は業務エラーとは限りません。

原因	例
業務エラー	CSV の必須項目不足、存在しない顧客コード
コード不具合	NullReference、想定外 JSON、バージョン不整合
構成ミス	接続名誤り、環境変数不足、Queue 名誤り
権限不足	Managed Identity の RBAC 不足、Key Vault 参照不可
下流障害	DB 障害、外部 API 障害、ネットワーク断
データ不整合	メッセージ形式不一致、古い schema のメッセージ
実装漏れ	シリアライズ失敗、日付形式の差異

そのため、poison queue は次のように扱います。

xxx-poison に入ったメッセージは、Functions が規定回数処理に失敗した未処理メッセージであり、運用上の調査・通知対象です。
原因は業務エラーとは限らず、コード不具合、構成ミス、一時障害、権限不足なども含まれます。

maxDequeueCount は、poison queue に移動する前の処理試行回数を制御します。既定値は 5 です。

Queue Trigger の重要な制約

Storage Queue は便利ですが、万能ではありません。

項目	Storage Queue の特徴
配信保証	At-Least-Once
順序保証	なし
重複配信	あり得る
1 メッセージサイズ	最大 64 KiB
Base64 使用時の実質ペイロード	より小さくなる
厳密 FIFO	不向き
重複検出	アプリ側で実装が必要

Storage Queue は At-Least-Once 配信です。

つまり、

少なくとも 1 回は処理される。
ただし、2 回以上処理される可能性もある。

ということです。

ここを見落とすと、かなりの確率で事故ります。

厳密な FIFO、重複検出、トランザクション、大きなメッセージが必要な場合は、Storage Queue ではなく Service Bus Queue を検討します。

ただし、ここでも注意があります。

Service Bus Queue を使えば常にグローバル FIFO になる、という意味ではありません。

Service Bus Queue で順序制御をしたい場合は、通常は SessionId によるセッション単位の順序処理として設計します。

また、Service Bus の重複検出も、送信側が MessageId を適切に設計し、重複検出を有効化していることが前提です。

失敗例：Queue にしたのに DB が燃えた話

ある案件で、HTTP Trigger のタイムアウト事故を避けるために Queue Trigger 化しました。

設計としては一見正しいです。

ところが本番で、夜間バッチの開始直後に Azure SQL が限界を迎えました。

原因は Queue Trigger の並列度です。

Queue Trigger の host.json では、以下のような設定があります。

batchSize
newBatchThreshold
visibilityTimeout
maxDequeueCount
messageEncoding

既定の batchSize は 16 です。

処理中メッセージ数が newBatchThreshold まで減ると、Functions runtime は次のバッチを取得します。

そのため、1 VM あたりの最大同時処理数は、概ね以下で見ます。

batchSize + newBatchThreshold

既定では、Consumption Plan の場合は概ね以下のようになります。

batchSize 16 + newBatchThreshold 8 = 1 VM あたり最大 24 件

Function App が 3 VM にスケールアウトすれば、同一 Queue Trigger で最大 72 件程度が同時処理され得ます。

24 件 / VM x 3 VM = 72 件

Premium / Dedicated / Flex では、newBatchThreshold の既定値に vCPU 数が影響するため、さらに注意が必要です。

現場で起きたことはこうでした。

ローカルでは 1 件ずつ処理していたので問題なし
本番では Queue に数万件投入
Functions がスケールアウト
DB に同時接続・同時更新が集中
一部がタイムアウト
Queue Trigger が例外で再試行
同じ DB 更新が再実行
重複更新を防ぐ制約がなく、データ不整合

非同期化しただけでは安全になりません。

Queue は「失敗したら再実行してくれる」ので、処理側が冪等でないと被害が増幅します。

`batchSize = 1` にしても完全直列とは限らない

ここも現場でよく誤解されます。

{
  "version": "2.0",
  "extensions": {
    "queues": {
      "batchSize": 1
    }
  }
}

これで 1 インスタンス内の並列度は抑えられます。

ただし、Function App が複数 VM にスケールアウトすれば、各 VM で 1 件ずつ処理されます。

つまり、

batchSize = 1 は「全体で完全に 1 件ずつ処理する」設定ではない

ということです。

完全な直列処理が必要な場合は、Storage Queue Trigger の設定だけで何とかしようとせず、以下を検討します。

Service Bus Sessions
DB 側の排他制御
ジョブ単位のロック
スケールアウト制御
Durable Functions
ワーカー分離
処理対象のパーティショニング

Queue Trigger の host.json 例

下流 DB や外部 API を守るために、並列度を抑える例です。

{
  "version": "2.0",
  "extensions": {
    "queues": {
      "visibilityTimeout": "00:05:00",
      "batchSize": 8,
      "newBatchThreshold": 4,
      "maxDequeueCount": 10,
      "messageEncoding": "base64"
    }
  }
}

各設定の考え方です。

設定	意味	現場での考え方
`visibilityTimeout`	失敗時に再処理されるまでの間隔	短すぎると下流をさらに苦しめる。長すぎると復旧が遅れる
`batchSize`	一度に取得して処理するメッセージ数	DB / API の耐性を見て決める
`newBatchThreshold`	次のバッチを取りに行くしきい値	`batchSize` と合わせて並列度を見る
`maxDequeueCount`	poison queue に移すまでの試行回数	少なすぎると一時的な競合で poison に行く。多すぎると障害を長引かせる
`messageEncoding`	メッセージエンコード	SDK 側と Functions 側で不一致にしない

Queue Trigger の並列度は、Functions の性能を上げる設定ではありません。

下流サービスを壊さないための安全弁です。

messageEncoding を使う場合は、Microsoft.Azure.Functions.Worker.Extensions.Storage.Queues 5.x 以上、または extension bundle 4.x 以上を前提にします。
送信側 SDK の QueueMessageEncoding と Functions の host.json の messageEncoding が不一致だと、デシリアライズ失敗や poison queue 送りの原因になります。

AzureWebJobsStorage を業務 Queue に使うのはサンプル限定にする

PoC や説明用サンプルでは、つい AzureWebJobsStorage をそのまま業務 Queue に使いがちです。

しかし、本番では Functions ホスト用 Storage と業務 Queue 用 Storage を分ける設計を推奨します。

AzureWebJobsStorage は、Functions runtime が動作するために重要な Storage です。トリガー管理、ホストの調整、キー、状態管理などに関係します。

本番では、たとえば次のように専用接続名を使います。

[QueueTrigger("import-jobs", Connection = "ImportQueueStorage")]

identity-based connection を使う場合の設定例です。

ImportQueueStorage__queueServiceUri=https://stexampleprod.queue.core.windows.net

可能であれば、connection string ではなく Managed Identity / RBAC を使います。

用途	推奨ロール例
Queue Trigger で読む	Storage Queue Data Message Processor
Queue に送信する	Storage Queue Data Message Sender / Storage Queue Data Contributor
Blob を読む	Storage Blob Data Reader
Blob に書く	Storage Blob Data Contributor

DefaultAzureCredential は説明用として便利ですが、本番では Function App に付与した managed identity と RBAC スコープを明示します。

ローカル開発者の権限で動く経路と、本番 managed identity で動く経路を混同しないようにします。

最小権限にするため、API 用 Function App、Event Bridge 用 Function App、Dispatcher 用 Function App、Worker 用 Function App で Managed Identity を分けることもよくあります。

Queue メッセージに業務データを詰め込まない

現場でよくある悪い例です。

{
  "customerId": "C001",
  "csvBody": "base64化した巨大CSV...",
  "requestedBy": "user@example.com"
}

これは避けます。

Storage Queue のメッセージサイズには制限があります。巨大な CSV、画像、PDF、JSON をそのまま Queue に詰める設計は危険です。

良い例です。

{
  "jobId": "imp_6d5f4c3a2b1e...",
  "tenantId": "tenant-a",
  "requestedByUserId": "user-123",
  "blob": {
    "storageAccount": "stexampleprod",
    "containerName": "imports",
    "blobName": "tenant-a/2026/06/19/file.csv",
    "etag": "\"0x8DC...\"",
    "versionId": null
  },
  "acceptedAt": "2026-06-19T10:15:00Z"
}

Queue メッセージに入れるべきなのは、処理対象そのものではなく、処理対象へのポインタと冪等性キーです。

Queue Trigger の冪等性は atomic に設計する

Queue Trigger では、最低限以下を入れます。

1. 冪等性キー

jobId、eventId、orderId などを使って、同じメッセージが複数回処理されても結果が壊れないようにします。

ただし、次のような実装はまだ弱いです。

if (IsCompleted(jobId))
{
    return;
}

MarkProcessing(jobId);

Process(jobId);

At-Least-Once 配信では、同じ jobId のメッセージがほぼ同時に 2 回処理される可能性があります。

その場合、2 つの Worker が同時に IsCompleted == false を見て、両方とも Process に進むリスクがあります。

そのため、処理権の取得は atomic な条件付き更新として実装します。

- jobId に一意制約を置く
- Enqueued / RetryableFailed の場合だけ Processing に遷移させる
- Processing でも lease が期限切れなら再取得できる
- 更新できなかった場合は、その理由を返す
- Processing の所有者を runId で持つ
- 完了更新も runId 条件付きで行う

SQL の疑似例です。

UPDATE ImportJobs
SET
    Status = 'Processing',
    ProcessingRunId = @RunId,
    ProcessingLeaseUntilUtc = @LeaseUntilUtc,
    UpdatedAtUtc = SYSUTCDATETIME()
WHERE
    JobId = @JobId
    AND (
        Status IN ('Enqueued', 'RetryableFailed')
        OR (
            Status = 'Processing'
            AND ProcessingLeaseUntilUtc < SYSUTCDATETIME()
        )
    );

この更新件数が 1 なら、その Worker が処理権を取得できたという意味です。

更新件数が 0 の場合は、理由を確認します。

ここで単純に false を返すだけでは不十分です。

`TryMarkProcessingAsync == false` で即 return は危険

ここはこの記事の中でもかなり重要です。

危険なサンプルです。

var acquired = await _jobs.TryMarkProcessingAsync(...);

if (!acquired)
{
    _logger.LogInformation(
        "Job is already processing or completed. JobId={JobId}",
        job.JobId);

    return;
}

これは危険です。

acquired == false の理由が、以下のどれなのか区別できていないためです。

既に完了済み
業務エラーとして終了済み
別 Worker が本当に処理中
前回 Worker がクラッシュし、DB 上だけ Processing が残っている
Job レコードが存在しない
想定外の状態

Azure Queue Trigger は、関数が正常完了するとメッセージを削除します。

関数が失敗すると、visibilityTimeout に基づいてメッセージが再表示されます。

Functions ホスト自体がクラッシュした場合は、ホストが visibilityTimeout を適用できず、Storage service 側の既定 10 分後にメッセージが再表示されます。

ここで、次の事故が起こり得ます。

1. Worker A が jobId=123 を Processing にする
2. Worker A がクラッシュする
3. Queue メッセージが 10 分後に再表示される
4. Worker B が同じメッセージを拾う
5. DB 上はまだ ProcessingLease が残っている
6. TryMarkProcessingAsync が false
7. Worker B が「誰か処理中か完了済み」と判断して return
8. Queue メッセージは正常完了扱いで削除される
9. DB には Processing の job だけが残る

これは最悪です。

「lease 期限切れ後に再取得可能」と書いていても、再取得のきっかけになる Queue メッセージを消してしまっています。

したがって、TryMarkProcessingAsync は true / false ではなく、理由を返す設計にします。

public enum ProcessingDecisionKind
{
    Acquired,
    AlreadyCompleted,
    AlreadyBusinessFailed,
    CurrentlyProcessing,
    LeaseNotExpired,
    NotFound,
    UnexpectedState
}

public sealed record ProcessingDecision(
    ProcessingDecisionKind Kind,
    string JobId,
    DateTimeOffset? ProcessingLeaseUntilUtc
);

Queue Trigger 側では、少なくとも次のように分けます。

switch (decision.Kind)
{
    case ProcessingDecisionKind.Acquired:
        break;

    case ProcessingDecisionKind.AlreadyCompleted:
    case ProcessingDecisionKind.AlreadyBusinessFailed:
        return;

    case ProcessingDecisionKind.CurrentlyProcessing:
    case ProcessingDecisionKind.LeaseNotExpired:
        throw new RetryLaterException(
            $"Job is still marked as Processing. JobId={job.JobId}");

    default:
        throw new InvalidOperationException(
            $"Unexpected job state. JobId={job.JobId}");
}

CurrentlyProcessing を常に正常終了扱いにすると、Worker クラッシュ後に再表示された Queue メッセージを削除してしまい、DB 上の Processing だけが残る可能性があります。

そのため、次のどれかを設計します。

方針	内容
retry-later	`CurrentlyProcessing` の場合は例外にして Queue の再表示に任せる
watchdog / re-enqueue	`Processing` の期限切れを監視し、必要なら再 enqueue する
delayed re-enqueue	現メッセージを消す前に、遅延付きで再投入する。ただし再投入失敗時の扱いも設計する
Durable Functions	状態管理が重くなったら Durable Functions に寄せる

CurrentlyProcessing で例外を投げる場合、visibilityTimeout、ProcessingLeaseUntilUtc、maxDequeueCount の整合が重要です。
visibilityTimeout が短すぎると、lease 期限切れ前に何度も再試行して poison queue に入る可能性があります。
この場合は、watchdog / re-enqueue や Durable Functions も検討します。

2. Processing の期限切れを設計する

Worker が Processing にした直後にクラッシュすることがあります。

その場合、DB 上のステータスが永遠に Processing のままだと、以後のリトライが進みません。

そのため、ProcessingLeaseUntilUtc のような期限を持たせます。

Processing:
  runId = abc123
  processingLeaseUntilUtc = 2026-06-19T10:30:00Z

期限切れの Processing は、次の Worker が再取得できるようにします。

ただし、期限を短くしすぎると、処理中に別 Worker が奪ってしまいます。

目安としては、以下を考慮します。

Queue の visibilityTimeout
Functions ホストクラッシュ時の 10 分再表示
処理の最大想定時間
外部 API のタイムアウト
DB 更新の最大時間
Function App のスケールイン
処理がハングした場合の回復時間
maxDequeueCount 到達までの時間

3. ProcessAsync 成功後に MarkCompletedAsync が失敗することもある

Queue Worker の設計で見落としがちなのが、次のパターンです。

1. ProcessAsync が外部 API 連携や DB 更新を完了する
2. MarkCompletedAsync が DB タイムアウトで失敗する
3. 関数が throw する
4. Queue Trigger が再試行する
5. 外部 API / DB 更新がもう一度実行される可能性がある

これは Queue Trigger の設計上、完全には避けられません。

Queue Trigger は、関数が失敗するとメッセージを再表示し、正常完了すると削除します。

一方、業務処理の副作用と状態更新は、常に 1 つのトランザクションにできるわけではありません。

そのため、ProcessAsync 内の副作用は必ず冪等にします。

副作用	守り方
DB 登録	`jobId` / business key に一意制約を置く
外部 API 連携	相手に `Idempotency-Key` 相当を渡す
外部 API が冪等 key 非対応	自システム側で送信履歴テーブルを持つ
ファイル生成	出力 blob 名を `jobId` ベースにする
メール送信	`jobId + notificationType` で送信履歴を持つ

特に外部 API 連携では、相手側に Idempotency-Key 相当を渡せるなら渡します。

渡せない場合は、自システム側で送信履歴テーブルを持ち、同じ jobId + externalOperation を二重送信しない設計が必要です。

4. 恒久エラーと一時エラーを分ける

たとえば以下は、何回リトライしても成功しない可能性が高いです。

CSV の列数が違う
必須項目がない
顧客コードが存在しない
ファイル形式が違う
業務ルール上、処理できない

これらは恒久エラーです。

例外を投げ続けて poison queue に送るより、業務エラーとして Failed にして、ユーザーに修正を促す方がよいです。

逆に以下は一時エラーです。

SQL の一時的なタイムアウト
外部 API の 429
外部 API の 503
ネットワーク断
一時的な DNS 解決失敗
DB のデッドロック

これらはリトライ対象です。

poison queue の監視

poison queue は、必ず監視対象にします。

ただし、繰り返しますが、poison queue は「業務エラー専用置き場」ではありません。

規定回数処理に失敗した未処理メッセージの隔離先です。

また、本番では poison message の raw body をそのままログに出すのは避けます。

Queue メッセージに個人情報、業務データ、SAS URL、アクセストークンが混ざる可能性があるためです。

ログには以下を出し、raw message はアクセス制御された隔離 Blob などに保存します。

jobId
message hash
schema version
queue message id
dequeue count
Function version
exception summary

3. Event Grid Trigger：軽く見えるが、実は重い設計判断

Event Grid Trigger が向いている処理

Event Grid は、イベント発行元で何かが起きたことを通知するサービスです。

Azure Functions では Event Grid Trigger を使って、Event Grid イベントを受け取れます。

たとえば以下のような用途です。

Blob が作成された
Blob が削除された
リソースが変更された
カスタムイベントが発行された
SaaS / 外部システムからイベントが来た

Event Grid Trigger が向いているのは、イベントに反応する処理です。

ただし、ここで大事なのは Event Grid が「通知」であるということです。

Event Grid は「この業務処理を必ず最後まで完了させてくれるジョブキュー」ではありません。

Event Grid の重要な性質

Event Grid には、設計上必ず押さえるべき性質があります。

項目	Event Grid の性質
配信保証	少なくとも 1 回
順序保証	なし
応答待ち	30 秒
成功扱い	HTTP 200〜204
失敗時	リトライ、dead-letter、破棄のいずれか
既定の最大試行回数	30 回
既定 TTL	1440 分
dead-letter	既定では有効ではない

Event Grid も重複配信があり得ます。

また、Event Grid はイベント配信後、エンドポイントから 30 秒以内に応答がない場合、配信失敗としてリトライキューに入れます。

Event Grid Trigger 内で重い処理をして 30 秒を超えると、Event Grid から見ると「失敗」に見えます。

その結果、同じイベントが再配信される可能性があります。

Event Grid の HTTP 配信仕様と EventGridTrigger の実装モデルを分けて考える

Event Grid の HTTP 配信仕様としては、HTTP 200〜204 のみが成功応答です。

それ以外のステータスコードは失敗扱いです。

ただし、失敗扱いになった後の挙動はステータスコードによって異なります。

ステータスコード	扱いの例
200〜204	成功
400	再試行されない
401	Azure Resources endpoint では 5 分以上後に再試行される場合がある
403	再試行されない
404	Azure Resources endpoint では 5 分以上後に再試行される場合がある
408	2 分以上後に再試行される場合がある
413	再試行されない
503	30 秒以上後に再試行される場合がある
その他	標準の指数バックオフに従う

つまり、

Event Grid では 200〜204 のみが成功応答であり、その他は失敗扱いとなる。
ただし 400 / 403 / 413 などは再試行されないため、意図せず 4xx 相当の失敗にすると dead-letter または破棄につながる。

ということです。

一方、Azure Functions の [EventGridTrigger] を使う場合、HTTP Trigger のようにアプリコードが HttpResponseData を返して 400 / 403 を明示する設計ではありません。

通常は、関数が正常終了するか、例外を投げるか、バインディング処理が成功するかによって、Event Grid への応答が決まります。

そのため、無効イベントをどう扱うかは設計で決めます。

方針	内容
正常終了してログだけ残す	再試行不要な無効イベント向き
invalid-event queue に退避	後で調査したい場合
例外を投げてリトライ	一時障害の可能性がある場合
dead-letter に任せる	Event Grid 側の dead-letter 設定と運用がある場合

「入力がおかしいから 400 を返す」という HTTP Webhook 的な説明と、[EventGridTrigger] の実装モデルは分けて考えた方が安全です。

失敗例：Blob Created を Event Grid で受けて直接 OCR した話

ある画像処理システムで、Blob に PDF が置かれたら Event Grid Trigger で OCR を実行する構成がありました。

設計時の会話はこうです。

「Blob が置かれたらすぐ処理したいので Event Grid で」
「Functions で OCR 呼べばサーバーレスですね」
「件数もそんなに多くないです」

本番後に起きたことはこうでした。

PDF サイズが大きいと OCR が 30 秒を超える
Event Grid から見ると Functions が応答しない
Event Grid が同じイベントをリトライ
同じ PDF に対して OCR が複数回走る
OCR API の課金が跳ねる
DB に重複結果
たまにイベント順序が前後し、後続処理が先に走る
dead-letter 未設定で、一部イベントは追跡不能

Event Grid Trigger は「イベントを受ける」のには向いています。

しかし、イベントを受けた先で重い処理を完了させる設計には向いていません。

改善後の設計

こう直しました。

Event Grid Trigger は、次のことだけをします。

イベント種別を確認する
Blob URL / ETag / eventId を取り出す
許可された Storage Account / Container / Prefix か確認する
ETag または VersionId が取れているか確認する
業務上の jobId を生成する
Job + Outbox を保存する
すぐ正常終了する

OCR のような重い処理は Queue Trigger 側で実行します。

この構成にすると、以下のメリットがあります。

Event Grid には素早く応答できる
OCR の失敗は Queue Trigger 側で再試行できる
Queue の poison queue で失敗ジョブを追跡できる
Queue の並列度で OCR API への負荷を制御できる
eventId / blobName / etag で重複処理を防げる
Blob 参照を構造化でき、ログに SAS URL を残さずに済む
Event Grid 受信と Queue 送信の曖昧状態を Outbox で扱える

Event Grid の eventId だけでは業務的な重複排除として弱い場合がある

Event Grid の再配信重複だけを抑えるなら、eventGridEvent.Id を使うのは妥当です。

しかし、OCR や CSV 取込で「同じ Blob 実体を 1 回だけ処理したい」という業務要件では、eventId だけでは弱い場合があります。

同じ Blob に対して複数のイベントが来ることがあります。

また、Event Grid は順序保証をしません。

そのため、業務上の jobId は次のような情報から生成する方が堅いです。

jobId =
  hash(
    storageAccount,
    containerName,
    blobName,
    eTag or versionId,
    eventType
  )

これにより、Event Grid のイベント ID ではなく、「処理対象の Blob バージョン」をキーにできます。

ETag / VersionId が取れないイベントをどう扱うか

同じ Blob バージョンを 1 回だけ処理したい設計では、ETag または VersionId の取得を原則必須にします。

ETag / VersionId が取れない状態で、以下だけから jobId を作るのは危険です。

storageAccount + containerName + blobName + eventType

この場合、同じ Blob 名の別バージョンまで同じ job と誤判定する可能性があります。

そのため、実務では次のどちらかを選びます。

方針	内容
厳格に扱う	`ETag` / `VersionId` が取れないイベントは invalid-event queue に退避する
イベント単位に寄せる	`eventId` を job key に含め、誤った重複排除を避ける

この記事のサンプルでは、同じ Blob バージョンを 1 回だけ処理したい前提なので、ETag / VersionId が取れないイベントは無効イベントとして扱います。

Event Grid Trigger の注意点

Event Grid Trigger を使う場合、最低限以下を確認します。

観点	確認内容
処理時間	30 秒以内に応答できる軽い処理か
成功応答	Event Grid の HTTP 配信仕様では 200〜204 が成功であることを理解しているか
EventGridTrigger	コードでは HTTP status を直接返すのではなく、正常終了 / 例外で扱うことを理解しているか
4xx 相当	再試行されない失敗を意図せず発生させていないか
重複	`eventId` だけでなく、Blob 実体単位の job key が必要か
ETag / VersionId	欠落時に invalid queue 退避するか、eventId を key に含めるか
順序	イベント順序を前提にしていないか
dead-letter	未配信イベントを保存する設定をしているか
フィルタ	不要イベントまで拾っていないか
後続処理	重い処理を Queue に逃がしているか

特に dead-letter は重要です。

Event Grid の dead-letter は既定では有効ではありません。

配信不能イベントを追跡したい場合は、イベントサブスクリプション作成時に dead-letter 用の Storage Account / Container を設定しておきます。

実装例：HTTP / Event Grid で受けて Outbox 経由で Queue に逃がす

以下は .NET isolated worker のイメージです。

C# の in-process model は 2026-11-10 にサポート終了予定です。
新規で Azure Functions を作る場合は、基本的に isolated worker model を検討するのが無難です。

以下のコードは設計意図を説明するためのサンプルです。
実案件では、DB 実装、Blob 検証、認証認可、例外分類、Outbox、監視、リトライポリシーをプロジェクト標準に合わせて実装してください。

app settings 例

本番では、Functions ホスト用 Storage と業務 Queue 用 Storage を分けます。

identity-based connection の例です。

AzureWebJobsStorage__accountName=stfuncprod

ImportQueueStorage__queueServiceUri=https://stimportprod.queue.core.windows.net
OcrQueueStorage__queueServiceUri=https://stocrprod.queue.core.windows.net

user-assigned managed identity を使う場合は、接続ごとに client id を指定する方針もあります。

ImportQueueStorage__queueServiceUri=https://stimportprod.queue.core.windows.net
ImportQueueStorage__clientId=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

ローカル開発や簡易検証では connection string を使うこともありますが、本番では Managed Identity / RBAC を優先します。

Program.cs

2026 年時点のテンプレート寄りに、FunctionsApplication.CreateBuilder(args) を使う例にします。

using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Builder;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

var builder = FunctionsApplication.CreateBuilder(args);

builder.ConfigureFunctionsWebApplication();

builder.Services
    .AddApplicationInsightsTelemetryWorkerService()
    .ConfigureFunctionsApplicationInsights();

builder.Services.AddSingleton<IQueueClientFactory, QueueClientFactory>();
builder.Services.AddSingleton<IBlobReferencePolicy, BlobReferencePolicy>();
builder.Services.AddSingleton<IBlobVerifier, BlobVerifier>();
builder.Services.AddSingleton<IPrincipalProvider, PrincipalProvider>();

// 以下は説明用。実案件では Azure SQL / Cosmos DB / Table Storage などに合わせて実装する。
builder.Services.AddSingleton<IJobRepository, JobRepository>();
builder.Services.AddSingleton<IOutboxRepository, OutboxRepository>();
builder.Services.AddSingleton<IImportProcessor, ImportProcessor>();

builder.Build().Run();

共通モデル

Queue メッセージでは、Blob URL ではなく構造化された Blob 参照を持つようにします。

また、RequestedBy はリクエスト本文ではなく、認証済み principal から取得した user id を使います。

public sealed record BlobReference(
    string StorageAccount,
    string ContainerName,
    string BlobName,
    string? ETag,
    string? VersionId
);

public sealed record SubmitImportRequest(
    string ContainerName,
    string BlobName,
    string? ETag,
    string? VersionId
);

public sealed record ImportJobMessage(
    string JobId,
    string TenantId,
    string RequestedByUserId,
    BlobReference Blob,
    DateTimeOffset AcceptedAt
);

public sealed record BlobEventJobMessage(
    string JobId,
    string EventId,
    string EventType,
    string Subject,
    BlobReference Blob,
    DateTimeOffset EnqueuedAt
);

public sealed record OutboxWorkItem(
    string OutboxId,
    string JobId,
    string QueueConnectionName,
    string QueueName,
    string PayloadJson,
    int AttemptCount
);

JSON サポート

HTTP 入力では Web API 的な JSON を扱うため、JsonSerializerDefaults.Web を明示します。

これにより、camelCase の property 名や case-insensitive matching など、Web 向けの既定値になります。

using System.Text.Json;

public static class JsonSupport
{
    public static readonly JsonSerializerOptions Options = new(JsonSerializerDefaults.Web);

    public static string ToCanonicalSubmitImportJson(SubmitImportRequest request)
    {
        // 余分な property は無視する方針の例。
        // 拒否したい場合は JSON schema validation などを追加する。
        var canonical = new
        {
            containerName = request.ContainerName,
            blobName = request.BlobName,
            eTag = request.ETag,
            versionId = request.VersionId
        };

        return JsonSerializer.Serialize(canonical, Options);
    }
}

request hash を raw body の Trim() だけで作ると、空白や property order の差異で別リクエスト扱いになります。
実務では、受け入れる JSON schema、余分な property の扱い、null と省略の扱いを決め、正規化済み DTO から canonical JSON を生成して hash 化します。

PrincipalProvider

実務では、Entra ID、APIM、JWT、Easy Auth などで検証済みの claim から取得します。

ここでは説明用に簡略化しています。

using Microsoft.Azure.Functions.Worker.Http;

public sealed record AuthenticatedPrincipal(
    string TenantId,
    string UserId
);

public interface IPrincipalProvider
{
    AuthenticatedPrincipal GetPrincipal(HttpRequestData req);
}

public sealed class PrincipalProvider : IPrincipalProvider
{
    public AuthenticatedPrincipal GetPrincipal(HttpRequestData req)
    {
        // 説明用の簡略実装。
        // 実案件では Entra ID / APIM / Easy Auth / JWT 検証済み claim から取得する。
        // Function Key は業務ユーザー認証ではない点に注意。

        var tenantId = req.Headers.TryGetValues("X-Tenant-Id", out var tenantValues)
            ? tenantValues.FirstOrDefault()
            : null;

        var userId = req.Headers.TryGetValues("X-Authenticated-UserId", out var userValues)
            ? userValues.FirstOrDefault()
            : null;

        if (string.IsNullOrWhiteSpace(tenantId) || string.IsNullOrWhiteSpace(userId))
        {
            throw new UnauthorizedAccessException("Authenticated principal is missing.");
        }

        return new AuthenticatedPrincipal(
            TenantId: tenantId,
            UserId: userId);
    }
}

この PrincipalProvider は説明用です。
本番で X-Authenticated-UserId のようなヘッダーをクライアントから直接信用してはいけません。APIM や Easy Auth など、信頼できる境界で付与・検証された情報だけを使います。

Blob 参照の許可ポリシー

本番では、任意の Blob を処理しないようにします。

Blob 名は case-sensitive なので、prefix 判定は StringComparison.Ordinal にします。

public interface IBlobReferencePolicy
{
    string AllowedStorageAccount { get; }

    bool IsAllowed(BlobReference blob);
}

public sealed class BlobReferencePolicy : IBlobReferencePolicy
{
    public string AllowedStorageAccount => "stexampleprod";

    public bool IsAllowed(BlobReference blob)
    {
        if (string.IsNullOrWhiteSpace(blob.StorageAccount) ||
            string.IsNullOrWhiteSpace(blob.ContainerName) ||
            string.IsNullOrWhiteSpace(blob.BlobName))
        {
            return false;
        }

        if (!string.Equals(
                blob.StorageAccount,
                AllowedStorageAccount,
                StringComparison.OrdinalIgnoreCase))
        {
            return false;
        }

        if (!string.Equals(
                blob.ContainerName,
                "imports",
                StringComparison.Ordinal))
        {
            return false;
        }

        // Blob 名は case-sensitive。
        // テナント境界や業務境界として prefix を使う場合は OrdinalIgnoreCase を使わない。
        if (!blob.BlobName.StartsWith("tenant-a/", StringComparison.Ordinal))
        {
            return false;
        }

        // 先頭 / やパストラバーサル風の表現は業務上不要なら拒否する。
        if (blob.BlobName.StartsWith("/", StringComparison.Ordinal) ||
            blob.BlobName.Contains("..", StringComparison.Ordinal) ||
            blob.BlobName.Contains("\\", StringComparison.Ordinal))
        {
            return false;
        }

        if (blob.BlobName.Any(char.IsControl))
        {
            return false;
        }

        return true;
    }
}

このポリシーは簡略化しています。

実務では以下も検討します。

テナントごとの prefix 制御
ファイル拡張子制限
Content-Type 検証
Blob サイズ制限
ETag / VersionId の必須化
ウイルススキャン済みフラグ
アップロード者と処理権限の整合性
実際に Blob properties を取得して ETag / ContentLength / ContentType を検証する

BlobVerifier

HTTP 受付時点で Blob の存在と属性を検証するためのインターフェースです。

Worker 実行時にも同じ確認を行い、すり替わりを防ぎます。

public sealed record BlobVerificationResult(
    bool IsValid,
    string? ErrorMessage
);

public interface IBlobVerifier
{
    Task<BlobVerificationResult> VerifyForImportAsync(
        BlobReference blob,
        CancellationToken cancellationToken);
}

public sealed class BlobVerifier : IBlobVerifier
{
    public Task<BlobVerificationResult> VerifyForImportAsync(
        BlobReference blob,
        CancellationToken cancellationToken)
    {
        // 説明用のスタブ。
        // 実案件では BlobClient を Managed Identity で作り、以下を確認する。
        //
        // - Blob が存在する
        // - ETag / VersionId が一致する
        // - ContentLength が上限内
        // - ContentType / 拡張子が許可内
        // - ウイルススキャン済み metadata がある
        // - テナント prefix と認可が一致している

        return Task.FromResult(
            new BlobVerificationResult(
                IsValid: true,
                ErrorMessage: null));
    }
}

QueueClientFactory

本番では専用接続名を使います。

ここでは identity-based connection の queueServiceUri を優先し、ローカル開発向けに connection string も扱える形にしています。

using Azure.Identity;
using Azure.Storage.Queues;
using Azure.Storage.Queues.Models;
using Microsoft.Extensions.Configuration;

public interface IQueueClientFactory
{
    QueueClient CreateClient(string connectionName, string queueName);
}

public sealed class QueueClientFactory : IQueueClientFactory
{
    private readonly IConfiguration _configuration;

    public QueueClientFactory(IConfiguration configuration)
    {
        _configuration = configuration;
    }

    public QueueClient CreateClient(string connectionName, string queueName)
    {
        var queueServiceUri =
            _configuration[$"{connectionName}__queueServiceUri"];

        var options = new QueueClientOptions
        {
            // host.json の messageEncoding と合わせる。
            MessageEncoding = QueueMessageEncoding.Base64
        };

        if (!string.IsNullOrWhiteSpace(queueServiceUri))
        {
            var serviceUri = new Uri(queueServiceUri.TrimEnd('/') + "/" + queueName);

            var credentialOptions = new DefaultAzureCredentialOptions();

            var managedIdentityClientId =
                _configuration[$"{connectionName}__clientId"];

            if (!string.IsNullOrWhiteSpace(managedIdentityClientId))
            {
                credentialOptions.ManagedIdentityClientId = managedIdentityClientId;
            }

            return new QueueClient(
                serviceUri,
                new DefaultAzureCredential(credentialOptions),
                options);
        }

        var connectionString = _configuration[connectionName];

        if (string.IsNullOrWhiteSpace(connectionString))
        {
            throw new InvalidOperationException(
                $"Queue connection '{connectionName}' is not configured.");
        }

        return new QueueClient(
            connectionString,
            queueName,
            options);
    }
}

DefaultAzureCredential は説明用として便利ですが、本番では Function App の system-assigned / user-assigned managed identity と RBAC スコープを明確に設計します。
ローカル開発者の権限で動く経路と、本番 managed identity で動く経路を混同しないようにしてください。

IJobRepository / IOutboxRepository の考え方

ここが一番大事です。

HTTP Trigger は Queue に直接送信せず、Job と Outbox を同一 DB トランザクションで保存します。

public enum AcceptDecisionKind
{
    Created,
    AlreadyAccepted,
    Conflict
}

public sealed record AcceptDecision(
    AcceptDecisionKind Kind,
    string JobId
);

public enum ProcessingDecisionKind
{
    Acquired,
    AlreadyCompleted,
    AlreadyBusinessFailed,
    CurrentlyProcessing,
    LeaseNotExpired,
    NotFound,
    UnexpectedState
}

public sealed record ProcessingDecision(
    ProcessingDecisionKind Kind,
    string JobId,
    DateTimeOffset? ProcessingLeaseUntilUtc
);

public interface IJobRepository
{
    /// <summary>
    /// tenantId + endpointName + userId + idempotencyKeyHash を一意キーとして受付する。
    /// 同じ key で requestBodyHash が同じなら既存 jobId を返す。
    /// 同じ key で requestBodyHash が違うなら Conflict を返す。
    ///
    /// 新規の場合は、Job と Outbox を同一 DB トランザクションで保存する。
    /// ここでは Queue に直接送信しない。
    /// </summary>
    Task<AcceptDecision> TryAcceptAndCreateOutboxAsync(
        string tenantId,
        string endpointName,
        string requestedByUserId,
        string idempotencyKeyHash,
        string requestBodyHash,
        string jobId,
        BlobReference blob,
        string queueConnectionName,
        string queueName,
        string payloadJson,
        CancellationToken cancellationToken);

    /// <summary>
    /// Event Grid 由来の job を作成し、Outbox も同じトランザクションで作る。
    /// 同じ Blob identity の job が既にある場合は AlreadyAccepted を返す。
    /// </summary>
    Task<AcceptDecision> TryCreateEventJobAndOutboxAsync(
        string jobId,
        string eventId,
        string eventType,
        BlobReference blob,
        string queueConnectionName,
        string queueName,
        string payloadJson,
        CancellationToken cancellationToken);

    /// <summary>
    /// Enqueued / RetryableFailed / 期限切れ Processing の場合だけ Processing に遷移させる。
    /// true/false ではなく理由を返す。
    /// </summary>
    Task<ProcessingDecision> TryAcquireProcessingAsync(
        string jobId,
        string runId,
        DateTimeOffset leaseUntilUtc,
        CancellationToken cancellationToken);

    Task MarkCompletedAsync(
        string jobId,
        string runId,
        CancellationToken cancellationToken);

    Task MarkBusinessFailedAsync(
        string jobId,
        string runId,
        string reason,
        CancellationToken cancellationToken);

    Task MarkRetryableFailedAsync(
        string jobId,
        string runId,
        string reason,
        CancellationToken cancellationToken);
}

public interface IOutboxRepository
{
    /// <summary>
    /// 送信期限が来た Outbox を lease 付きで取得する。
    /// 複数 Dispatcher がいても同じ outbox を同時に送らないよう、atomic に lease を取る。
    /// </summary>
    Task<IReadOnlyList<OutboxWorkItem>> LeaseDueAsync(
        int maxItems,
        string dispatcherId,
        DateTimeOffset leaseUntilUtc,
        CancellationToken cancellationToken);

    Task MarkDispatchedAsync(
        string outboxId,
        string dispatcherId,
        string queueMessageId,
        CancellationToken cancellationToken);

    Task MarkDispatchFailedAsync(
        string outboxId,
        string dispatcherId,
        string reason,
        DateTimeOffset nextRetryAtUtc,
        CancellationToken cancellationToken);

    /// <summary>
    /// Queue SDK の送信呼び出し中または呼び出し後に失敗した可能性がある状態。
    /// クライアントからは失敗に見えても、サービス側にはメッセージ作成済みの可能性がある。
    /// ここで Failed と断定しない。
    /// </summary>
    Task MarkDispatchUnknownAsync(
        string outboxId,
        string dispatcherId,
        string reason,
        DateTimeOffset nextCheckAtUtc,
        CancellationToken cancellationToken);
}

重要なのは、TryAcquireProcessingAsync を単純な bool にしないことです。

AlreadyCompleted と CurrentlyProcessing は意味がまったく違います。

AlreadyCompleted なら Queue メッセージを消してよいです。

しかし、CurrentlyProcessing を正常終了扱いにすると、Worker クラッシュ後に再表示された Queue メッセージを消してしまう可能性があります。

HashHelper

Idempotency-Key や request body hash、Blob identity 由来の job id を作るためのヘルパーです。

ToBlobJobId は ETag または VersionId を必須にしています。

using System.Security.Cryptography;
using System.Text;

public static class HashHelper
{
    public static string Sha256Hex(string value)
    {
        using var sha256 = SHA256.Create();

        return Convert
            .ToHexString(sha256.ComputeHash(Encoding.UTF8.GetBytes(value)))
            .ToLowerInvariant();
    }

    public static string ToImportJobId(
        string tenantId,
        string endpointName,
        string userId,
        string idempotencyKey)
    {
        var material = string.Join(
            "|",
            tenantId,
            endpointName,
            userId,
            idempotencyKey);

        return "imp_" + Sha256Hex(material)[..32];
    }

    public static string ToBlobJobId(
        BlobReference blob,
        string eventType)
    {
        var blobVersionKey =
            !string.IsNullOrWhiteSpace(blob.VersionId)
                ? blob.VersionId
                : blob.ETag;

        if (string.IsNullOrWhiteSpace(blobVersionKey))
        {
            throw new InvalidOperationException(
                "ETag or VersionId is required to build a blob job id.");
        }

        var material = string.Join(
            "|",
            blob.StorageAccount,
            blob.ContainerName,
            blob.BlobName,
            blobVersionKey,
            eventType);

        return "blob_" + Sha256Hex(material)[..32];
    }
}

HTTP Trigger：Job + Outbox を作って 202 を返す

以下のサンプルでは、Idempotency-Key を必須にしています。

また、WriteAsJsonAsync ではステータスコードを明示する overload を使っています。

isolated worker 1.x では WriteAsJsonAsync() がステータスコードを 200 OK に上書きする挙動がありました。
isolated worker 2.x ではその挙動は変更されていますが、バージョン差を避けるため、サンプルではステータスコードを明示する overload を使っています。

using System.Net;
using System.Text;
using System.Text.Json;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.Extensions.Logging;

public sealed class RequestBodyTooLargeException : Exception
{
    public RequestBodyTooLargeException(string message) : base(message)
    {
    }
}

public sealed class SubmitImportFunction
{
    private const string EndpointName = "POST /imports";
    private const string QueueConnectionName = "ImportQueueStorage";
    private const string QueueName = "import-jobs";
    private const long MaxRequestBodyBytes = 32 * 1024;

    private readonly IJobRepository _jobs;
    private readonly IBlobReferencePolicy _blobPolicy;
    private readonly IBlobVerifier _blobVerifier;
    private readonly IPrincipalProvider _principalProvider;
    private readonly ILogger<SubmitImportFunction> _logger;

    public SubmitImportFunction(
        IJobRepository jobs,
        IBlobReferencePolicy blobPolicy,
        IBlobVerifier blobVerifier,
        IPrincipalProvider principalProvider,
        ILogger<SubmitImportFunction> logger)
    {
        _jobs = jobs;
        _blobPolicy = blobPolicy;
        _blobVerifier = blobVerifier;
        _principalProvider = principalProvider;
        _logger = logger;
    }

    [Function(nameof(SubmitImportFunction))]
    public async Task<HttpResponseData> Run(
        [HttpTrigger(AuthorizationLevel.Function, "post", Route = "imports")]
        HttpRequestData req,
        CancellationToken cancellationToken)
    {
        AuthenticatedPrincipal principal;

        try
        {
            principal = _principalProvider.GetPrincipal(req);
        }
        catch (UnauthorizedAccessException)
        {
            var unauthorized = req.CreateResponse();
            await unauthorized.WriteAsJsonAsync(
                new { error = "Unauthorized." },
                HttpStatusCode.Unauthorized,
                cancellationToken);

            return unauthorized;
        }

        if (TryGetContentLength(req, out var contentLength) &&
            contentLength > MaxRequestBodyBytes)
        {
            var tooLarge = req.CreateResponse();
            await tooLarge.WriteAsJsonAsync(
                new { error = "Request body is too large." },
                HttpStatusCode.RequestEntityTooLarge,
                cancellationToken);

            return tooLarge;
        }

        if (!req.Headers.TryGetValues("Idempotency-Key", out var keyValues))
        {
            var badRequest = req.CreateResponse();
            await badRequest.WriteAsJsonAsync(
                new { error = "Idempotency-Key header is required." },
                HttpStatusCode.BadRequest,
                cancellationToken);

            return badRequest;
        }

        var idempotencyKey = keyValues.FirstOrDefault();

        if (string.IsNullOrWhiteSpace(idempotencyKey))
        {
            var badRequest = req.CreateResponse();
            await badRequest.WriteAsJsonAsync(
                new { error = "Idempotency-Key header is empty." },
                HttpStatusCode.BadRequest,
                cancellationToken);

            return badRequest;
        }

        string rawBody;

        try
        {
            rawBody = await ReadLimitedUtf8BodyAsync(
                req.Body,
                MaxRequestBodyBytes,
                cancellationToken);
        }
        catch (RequestBodyTooLargeException)
        {
            var tooLarge = req.CreateResponse();
            await tooLarge.WriteAsJsonAsync(
                new { error = "Request body is too large." },
                HttpStatusCode.RequestEntityTooLarge,
                cancellationToken);

            return tooLarge;
        }

        SubmitImportRequest? body;

        try
        {
            body = JsonSerializer.Deserialize<SubmitImportRequest>(
                rawBody,
                JsonSupport.Options);
        }
        catch (JsonException)
        {
            var badRequest = req.CreateResponse();
            await badRequest.WriteAsJsonAsync(
                new { error = "Request body is invalid JSON." },
                HttpStatusCode.BadRequest,
                cancellationToken);

            return badRequest;
        }

        if (body is null)
        {
            var badRequest = req.CreateResponse();
            await badRequest.WriteAsJsonAsync(
                new { error = "Request body is required." },
                HttpStatusCode.BadRequest,
                cancellationToken);

            return badRequest;
        }

        var blob = new BlobReference(
            StorageAccount: _blobPolicy.AllowedStorageAccount,
            ContainerName: body.ContainerName,
            BlobName: body.BlobName,
            ETag: body.ETag,
            VersionId: body.VersionId);

        if (!_blobPolicy.IsAllowed(blob))
        {
            var forbidden = req.CreateResponse();
            await forbidden.WriteAsJsonAsync(
                new { error = "The specified blob is not allowed." },
                HttpStatusCode.Forbidden,
                cancellationToken);

            return forbidden;
        }

        var verification = await _blobVerifier.VerifyForImportAsync(
            blob,
            cancellationToken);

        if (!verification.IsValid)
        {
            var badRequest = req.CreateResponse();
            await badRequest.WriteAsJsonAsync(
                new { error = verification.ErrorMessage ?? "The specified blob is invalid." },
                HttpStatusCode.BadRequest,
                cancellationToken);

            return badRequest;
        }

        var idempotencyKeyHash = HashHelper.Sha256Hex(idempotencyKey);

        var canonicalRequestJson =
            JsonSupport.ToCanonicalSubmitImportJson(body);

        var requestBodyHash =
            HashHelper.Sha256Hex(canonicalRequestJson);

        var jobId = HashHelper.ToImportJobId(
            principal.TenantId,
            EndpointName,
            principal.UserId,
            idempotencyKey);

        var message = new ImportJobMessage(
            JobId: jobId,
            TenantId: principal.TenantId,
            RequestedByUserId: principal.UserId,
            Blob: blob,
            AcceptedAt: DateTimeOffset.UtcNow);

        var payloadJson = JsonSerializer.Serialize(
            message,
            JsonSupport.Options);

        var decision = await _jobs.TryAcceptAndCreateOutboxAsync(
            tenantId: principal.TenantId,
            endpointName: EndpointName,
            requestedByUserId: principal.UserId,
            idempotencyKeyHash: idempotencyKeyHash,
            requestBodyHash: requestBodyHash,
            jobId: jobId,
            blob: blob,
            queueConnectionName: QueueConnectionName,
            queueName: QueueName,
            payloadJson: payloadJson,
            cancellationToken: cancellationToken);

        if (decision.Kind == AcceptDecisionKind.Conflict)
        {
            var conflict = req.CreateResponse();
            await conflict.WriteAsJsonAsync(
                new
                {
                    error = "The same Idempotency-Key was used with a different request body."
                },
                HttpStatusCode.Conflict,
                cancellationToken);

            return conflict;
        }

        if (decision.Kind == AcceptDecisionKind.AlreadyAccepted)
        {
            _logger.LogInformation(
                "Duplicate HTTP request accepted without creating new outbox. JobId={JobId}, TenantId={TenantId}, UserId={UserId}",
                decision.JobId,
                principal.TenantId,
                principal.UserId);

            var duplicateResponse = req.CreateResponse();
            await duplicateResponse.WriteAsJsonAsync(
                new
                {
                    jobId = decision.JobId,
                    statusUrl = $"/api/imports/{decision.JobId}"
                },
                HttpStatusCode.Accepted,
                cancellationToken);

            return duplicateResponse;
        }

        _logger.LogInformation(
            "Import job accepted. Job and outbox were created. JobId={JobId}, TenantId={TenantId}, UserId={UserId}, Container={Container}, BlobName={BlobName}",
            decision.JobId,
            principal.TenantId,
            principal.UserId,
            blob.ContainerName,
            blob.BlobName);

        var response = req.CreateResponse();
        await response.WriteAsJsonAsync(
            new
            {
                jobId = decision.JobId,
                statusUrl = $"/api/imports/{decision.JobId}"
            },
            HttpStatusCode.Accepted,
            cancellationToken);

        return response;
    }

    private static bool TryGetContentLength(
        HttpRequestData req,
        out long contentLength)
    {
        contentLength = 0;

        if (!req.Headers.TryGetValues("Content-Length", out var values))
        {
            return false;
        }

        return long.TryParse(
            values.FirstOrDefault(),
            out contentLength);
    }

    private static async Task<string> ReadLimitedUtf8BodyAsync(
        Stream stream,
        long maxBytes,
        CancellationToken cancellationToken)
    {
        using var memory = new MemoryStream();
        var buffer = new byte[8192];
        long total = 0;

        while (true)
        {
            var read = await stream.ReadAsync(
                buffer.AsMemory(0, buffer.Length),
                cancellationToken);

            if (read == 0)
            {
                break;
            }

            total += read;

            if (total > maxBytes)
            {
                throw new RequestBodyTooLargeException(
                    $"Request body exceeded {maxBytes} bytes.");
            }

            memory.Write(buffer, 0, read);
        }

        return Encoding.UTF8.GetString(memory.ToArray());
    }
}

このサンプルの重要点は以下です。

request body size を制限する
壊れた JSON を 400 Bad Request にする
JsonSerializerDefaults.Web を明示する
request hash は canonical JSON から作る
Idempotency-Key を必須にする
tenant / endpoint / user / key でスコープ化する
request body hash を保存し、同じ key で異なる本文なら 409 にする
RequestedBy は本文ではなく principal から決める
Blob 参照を policy と verifier で検証する
HTTP Trigger は Queue に直接送信しない
Job + Outbox を同一 DB トランザクションで保存する
WriteAsJsonAsync でステータスコードを明示する
任意の Blob URL ではなく構造化された Blob 参照を使う

Outbox Dispatcher：送信試行後の例外は DispatchUnknown に寄せる

Outbox Dispatcher は、Outbox を読んで Queue に送信します。

ここでも、Queue 送信と DB 更新は atomic ではありません。

特に重要なのは、SendMessageAsync が例外を返しても、サービス側ではメッセージ作成済みの可能性があることです。

そのため、Queue SDK 呼び出し中または呼び出し後に発生した例外は、原則として DispatchUnknown に寄せます。

また、Outbox Dispatcher では「1 件失敗したらバッチ全体を止めるか」も設計します。

方式	内容
方式 A	1 件失敗したら throw して Dispatcher 全体を失敗として監視する。シンプルだが、同じバッチ内の後続 item は次回まで送られない
方式 B	item ごとに try/catch し、失敗 item だけ状態更新して次 item を処理する。大量 Outbox では滞留しにくい

大量 Outbox を扱うなら、方式 B の方が滞留しにくいです。

以下のサンプルは方式 B です。

using Azure.Storage.Queues;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

public sealed class OutboxDispatcherFunction
{
    private static readonly TimeSpan DispatchLease = TimeSpan.FromMinutes(2);

    private readonly IOutboxRepository _outbox;
    private readonly IQueueClientFactory _queueClientFactory;
    private readonly ILogger<OutboxDispatcherFunction> _logger;

    public OutboxDispatcherFunction(
        IOutboxRepository outbox,
        IQueueClientFactory queueClientFactory,
        ILogger<OutboxDispatcherFunction> logger)
    {
        _outbox = outbox;
        _queueClientFactory = queueClientFactory;
        _logger = logger;
    }

    [Function(nameof(OutboxDispatcherFunction))]
    public async Task Run(
        [TimerTrigger("*/10 * * * * *")]
        TimerInfo timer,
        CancellationToken cancellationToken)
    {
        var dispatcherId = Environment.GetEnvironmentVariable("WEBSITE_INSTANCE_ID")
            ?? Environment.MachineName
            ?? Guid.NewGuid().ToString("N");

        var items = await _outbox.LeaseDueAsync(
            maxItems: 50,
            dispatcherId: dispatcherId,
            leaseUntilUtc: DateTimeOffset.UtcNow.Add(DispatchLease),
            cancellationToken: cancellationToken);

        var successCount = 0;
        var failureCount = 0;

        foreach (var item in items)
        {
            try
            {
                await DispatchOneAsync(
                    item,
                    dispatcherId,
                    cancellationToken);

                successCount++;
            }
            catch (Exception ex)
            {
                failureCount++;

                _logger.LogError(
                    ex,
                    "Outbox item dispatch failed. OutboxId={OutboxId}, JobId={JobId}",
                    item.OutboxId,
                    item.JobId);

                // 方式 B:
                // 1 件失敗しても次の Outbox item を処理する。
                // 関数全体を failed にしたい場合は、ループ後に failureCount を見て throw する方針もある。
            }
        }

        _logger.LogInformation(
            "Outbox dispatch batch finished. SuccessCount={SuccessCount}, FailureCount={FailureCount}",
            successCount,
            failureCount);

        // 運用方針次第:
        // failureCount > 0 のときに throw して Timer 実行を failed として見せることもできる。
        // ただし、失敗 item は各 item の状態として記録済みにしておく。
    }

    private async Task DispatchOneAsync(
        OutboxWorkItem item,
        string dispatcherId,
        CancellationToken cancellationToken)
    {
        bool sendAttemptStarted = false;
        string? queueMessageId = null;

        try
        {
            var queue = _queueClientFactory.CreateClient(
                item.QueueConnectionName,
                item.QueueName);

            sendAttemptStarted = true;

            var sendResult = await queue.SendMessageAsync(
                item.PayloadJson,
                cancellationToken: cancellationToken);

            queueMessageId = sendResult.Value.MessageId;

            await _outbox.MarkDispatchedAsync(
                item.OutboxId,
                dispatcherId,
                queueMessageId,
                cancellationToken);

            _logger.LogInformation(
                "Outbox dispatched. OutboxId={OutboxId}, JobId={JobId}, QueueName={QueueName}, QueueMessageId={QueueMessageId}",
                item.OutboxId,
                item.JobId,
                item.QueueName,
                queueMessageId);
        }
        catch (Exception ex) when (!sendAttemptStarted)
        {
            // Queue SDK 呼び出し前の検証・設定エラーなど。
            // 本当に送信試行前と判断できるものだけ Failed。
            var nextRetryAt = DateTimeOffset.UtcNow.Add(
                Backoff(item.AttemptCount));

            await _outbox.MarkDispatchFailedAsync(
                item.OutboxId,
                dispatcherId,
                ex.Message,
                nextRetryAt,
                cancellationToken);

            _logger.LogWarning(
                ex,
                "Outbox dispatch failed before queue send attempt. OutboxId={OutboxId}, JobId={JobId}, NextRetryAt={NextRetryAt}",
                item.OutboxId,
                item.JobId,
                nextRetryAt);

            throw;
        }
        catch (Exception ex)
        {
            // SendMessageAsync を試行した後の例外は、
            // 実際に Queue に入ったかどうか断定できない。
            // ネットワークタイムアウト、応答受信前の切断、クライアント側キャンセルなどでは、
            // サービス側にはメッセージが作成済みの可能性がある。
            var nextCheckAt = DateTimeOffset.UtcNow.AddMinutes(5);

            await _outbox.MarkDispatchUnknownAsync(
                item.OutboxId,
                dispatcherId,
                ex.Message,
                nextCheckAt,
                cancellationToken);

            _logger.LogError(
                ex,
                "Outbox dispatch state is unknown after queue send attempt. OutboxId={OutboxId}, JobId={JobId}, QueueMessageId={QueueMessageId}",
                item.OutboxId,
                item.JobId,
                queueMessageId);

            throw;
        }
    }

    private static TimeSpan Backoff(int attemptCount)
    {
        var seconds = Math.Min(300, Math.Pow(2, Math.Min(attemptCount, 8)));
        return TimeSpan.FromSeconds(seconds);
    }
}

この設計で大事なのは、以下です。

HTTP Trigger では Queue に直接送信しない
DB の Job + Outbox 作成を atomic にする
Dispatcher は Queue SDK 呼び出し後の例外を DispatchUnknown として扱う
SendMessageAsync の例外を「未送信」と断定しない
再送による Queue 重複は Worker 側の jobId 冪等性で吸収する
Outbox の滞留、DispatchFailed、DispatchUnknown を監視する
1 件失敗でバッチ全体を止めるか、item ごとに処理継続するかを運用方針として決める

Queue Trigger：理由付きで処理権を判断する

Queue Trigger では、QueueMessage を受け取ると message id や dequeue count もログに出しやすくなります。

QueueMessage などの SDK 型にバインドする場合は、Microsoft.Azure.Functions.Worker.Extensions.Storage.Queues 5.2.0 以上を前提にします。
環境や拡張機能バージョンによっては string バインドを使う方がシンプルです。

using System.Text.Json;
using Azure.Storage.Queues.Models;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

public sealed class RetryLaterException : Exception
{
    public RetryLaterException(string message) : base(message)
    {
    }
}

public sealed class BusinessValidationException : Exception
{
    public BusinessValidationException(string message) : base(message)
    {
    }
}

public interface IImportProcessor
{
    Task ProcessAsync(
        ImportJobMessage job,
        CancellationToken cancellationToken);
}

public sealed class ProcessImportFunction
{
    private static readonly TimeSpan ProcessingLease = TimeSpan.FromMinutes(15);

    private readonly IImportProcessor _processor;
    private readonly IJobRepository _jobs;
    private readonly IBlobVerifier _blobVerifier;
    private readonly ILogger<ProcessImportFunction> _logger;

    public ProcessImportFunction(
        IImportProcessor processor,
        IJobRepository jobs,
        IBlobVerifier blobVerifier,
        ILogger<ProcessImportFunction> logger)
    {
        _processor = processor;
        _jobs = jobs;
        _blobVerifier = blobVerifier;
        _logger = logger;
    }

    [Function(nameof(ProcessImportFunction))]
    public async Task Run(
        [QueueTrigger("import-jobs", Connection = "ImportQueueStorage")]
        QueueMessage queueMessage,
        CancellationToken cancellationToken)
    {
        var rawMessage = queueMessage.Body.ToString();

        ImportJobMessage job;

        try
        {
            job = JsonSerializer.Deserialize<ImportJobMessage>(
                    rawMessage,
                    JsonSupport.Options)
                ?? throw new InvalidOperationException("Invalid queue message.");
        }
        catch (JsonException ex)
        {
            _logger.LogError(
                ex,
                "Queue message JSON is invalid. QueueMessageId={QueueMessageId}, DequeueCount={DequeueCount}, MessageHash={MessageHash}",
                queueMessage.MessageId,
                queueMessage.DequeueCount,
                HashHelper.Sha256Hex(rawMessage));

            throw;
        }

        var runId = Guid.NewGuid().ToString("N");
        var leaseUntilUtc = DateTimeOffset.UtcNow.Add(ProcessingLease);

        using var scope = _logger.BeginScope(new Dictionary<string, object>
        {
            ["JobId"] = job.JobId,
            ["RunId"] = runId,
            ["QueueMessageId"] = queueMessage.MessageId,
            ["DequeueCount"] = queueMessage.DequeueCount
        });

        var decision = await _jobs.TryAcquireProcessingAsync(
            job.JobId,
            runId,
            leaseUntilUtc,
            cancellationToken);

        switch (decision.Kind)
        {
            case ProcessingDecisionKind.Acquired:
                break;

            case ProcessingDecisionKind.AlreadyCompleted:
            case ProcessingDecisionKind.AlreadyBusinessFailed:
                _logger.LogInformation(
                    "Job already finished. Message will be deleted. JobId={JobId}, Decision={Decision}",
                    job.JobId,
                    decision.Kind);

                return;

            case ProcessingDecisionKind.CurrentlyProcessing:
            case ProcessingDecisionKind.LeaseNotExpired:
                _logger.LogWarning(
                    "Job is still marked as Processing. Throwing to avoid deleting queue message. JobId={JobId}, LeaseUntil={LeaseUntil}",
                    job.JobId,
                    decision.ProcessingLeaseUntilUtc);

                throw new RetryLaterException(
                    $"Job is still marked as Processing. JobId={job.JobId}");

            case ProcessingDecisionKind.NotFound:
            case ProcessingDecisionKind.UnexpectedState:
            default:
                _logger.LogError(
                    "Unexpected job state. JobId={JobId}, Decision={Decision}",
                    job.JobId,
                    decision.Kind);

                throw new InvalidOperationException(
                    $"Unexpected job state. JobId={job.JobId}, Decision={decision.Kind}");
        }

        var verification = await _blobVerifier.VerifyForImportAsync(
            job.Blob,
            cancellationToken);

        if (!verification.IsValid)
        {
            await _jobs.MarkBusinessFailedAsync(
                job.JobId,
                runId,
                verification.ErrorMessage ?? "Blob verification failed.",
                cancellationToken);

            _logger.LogWarning(
                "Blob verification failed. JobId={JobId}, Reason={Reason}",
                job.JobId,
                verification.ErrorMessage);

            return;
        }

        try
        {
            await _processor.ProcessAsync(job, cancellationToken);
        }
        catch (BusinessValidationException ex)
        {
            await _jobs.MarkBusinessFailedAsync(
                job.JobId,
                runId,
                ex.Message,
                cancellationToken);

            _logger.LogWarning(
                ex,
                "Import job failed by business validation. JobId={JobId}",
                job.JobId);

            // リトライしても成功しない業務エラーなので throw しない
            return;
        }
        catch (Exception ex)
        {
            await _jobs.MarkRetryableFailedAsync(
                job.JobId,
                runId,
                ex.Message,
                cancellationToken);

            _logger.LogError(
                ex,
                "Import job failed by transient or unknown error. JobId={JobId}",
                job.JobId);

            // 一時障害の可能性があるため Queue Trigger の再試行に任せる
            throw;
        }

        try
        {
            await _jobs.MarkCompletedAsync(
                job.JobId,
                runId,
                cancellationToken);

            _logger.LogInformation(
                "Import job completed. JobId={JobId}",
                job.JobId);
        }
        catch (Exception ex)
        {
            // ProcessAsync の副作用は既に完了している可能性がある。
            // ここで再実行されても結果が壊れないよう、ProcessAsync 側の副作用は必ず冪等にする。
            _logger.LogError(
                ex,
                "Import job side effects may have completed, but MarkCompleted failed. JobId={JobId}, RunId={RunId}",
                job.JobId,
                runId);

            throw;
        }
    }
}

設計の肝は、ProcessAsync の前に atomic に処理権を取ることです。

さらに、ProcessAsync 成功後に MarkCompletedAsync が失敗しても、再実行で結果が壊れないようにします。

種別	例	扱い
一時障害	SQL タイムアウト、外部 API 503	`RetryableFailed` にして `throw`
恒久的な業務エラー	CSV フォーマット不正、必須項目不足	`BusinessFailed` にして終了
重複メッセージ	同じ `jobId` が再実行	完了済みなら成功扱い
本当に処理中か不明	`CurrentlyProcessing`	無条件 return せず retry-later / watchdog
Worker クラッシュ	`Processing` のまま停止	lease 期限切れ後に再取得、または watchdog で再 enqueue
副作用後の完了更新失敗	外部 API 送信後に `MarkCompleted` 失敗	副作用側を `jobId` で冪等化

poison queue handler：raw message をそのままログに出さない

using Azure.Storage.Queues.Models;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

public sealed class ImportPoisonHandler
{
    private readonly ILogger<ImportPoisonHandler> _logger;

    public ImportPoisonHandler(ILogger<ImportPoisonHandler> logger)
    {
        _logger = logger;
    }

    [Function(nameof(ImportPoisonHandler))]
    public Task Run(
        [QueueTrigger("import-jobs-poison", Connection = "ImportQueueStorage")]
        QueueMessage queueMessage,
        CancellationToken cancellationToken)
    {
        var rawMessage = queueMessage.Body.ToString();
        var messageHash = HashHelper.Sha256Hex(rawMessage);

        _logger.LogError(
            "Poison message detected. QueueMessageId={QueueMessageId}, DequeueCount={DequeueCount}, MessageHash={MessageHash}",
            queueMessage.MessageId,
            queueMessage.DequeueCount,
            messageHash);

        // 実案件ではここで以下を行う:
        // - 運用通知
        // - 障害管理チケット作成
        // - 対象 jobId の調査
        // - メッセージ schema の確認
        // - コード不具合 / 構成ミス / 権限不足 / 下流障害の切り分け
        // - raw message はアクセス制御された隔離 Blob に保存
        // - 再実行可能な形式で保管

        return Task.CompletedTask;
    }
}

poison queue に入ったら、最低限以下を確認します。

jobId
message hash
Queue message id
dequeue count
Function の例外ログ
対象 Function のバージョン
App Settings / Key Vault 参照
Managed Identity の権限
下流 DB / 外部 API の状態
メッセージ schema のバージョン
同時刻のデプロイ有無

Event Grid Trigger：イベントを Job + Outbox に変換するだけ

Event Grid Trigger 側では、重い処理をしません。

以下は、Blob Created イベントを受けて Job + Outbox を作るだけの例です。

JobId は eventGridEvent.Id ではなく、Blob 実体を表す情報から作ります。

using System.Text.Json;
using Azure.Messaging.EventGrid;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

public sealed class BlobCreatedBridgeFunction
{
    private const string QueueConnectionName = "OcrQueueStorage";
    private const string QueueName = "ocr-jobs";

    private readonly IJobRepository _jobs;
    private readonly IBlobReferencePolicy _blobPolicy;
    private readonly ILogger<BlobCreatedBridgeFunction> _logger;

    public BlobCreatedBridgeFunction(
        IJobRepository jobs,
        IBlobReferencePolicy blobPolicy,
        ILogger<BlobCreatedBridgeFunction> logger)
    {
        _jobs = jobs;
        _blobPolicy = blobPolicy;
        _logger = logger;
    }

    [Function(nameof(BlobCreatedBridgeFunction))]
    public async Task Run(
        [EventGridTrigger] EventGridEvent eventGridEvent,
        CancellationToken cancellationToken)
    {
        if (eventGridEvent.EventType != "Microsoft.Storage.BlobCreated")
        {
            _logger.LogInformation(
                "Ignored event. EventType={EventType}",
                eventGridEvent.EventType);

            return;
        }

        var blob = TryParseBlobReference(eventGridEvent);

        if (blob is null)
        {
            _logger.LogWarning(
                "Blob reference could not be parsed. EventId={EventId}",
                eventGridEvent.Id);

            // EventGridTrigger では HTTP status を直接返すのではなく、
            // 正常終了 / 例外のどちらにするかで再試行方針を決める。
            // この例では、再試行不要な無効イベントとして正常終了する。
            // 調査したい場合は invalid-event queue への退避も検討する。
            return;
        }

        if (string.IsNullOrWhiteSpace(blob.ETag) &&
            string.IsNullOrWhiteSpace(blob.VersionId))
        {
            _logger.LogWarning(
                "Blob ETag and VersionId are missing. Event will not be converted to job. EventId={EventId}, StorageAccount={StorageAccount}, Container={Container}, BlobName={BlobName}",
                eventGridEvent.Id,
                blob.StorageAccount,
                blob.ContainerName,
                blob.BlobName);

            // 同じ Blob バージョンを 1 回だけ処理したい設計では、
            // ETag / VersionId なしに jobId を作ると誤った重複排除になる可能性がある。
            // 実務では invalid-event queue へ退避するか、eventId を key に含める設計へ切り替える。
            return;
        }

        if (!_blobPolicy.IsAllowed(blob))
        {
            _logger.LogWarning(
                "Blob reference is not allowed. EventId={EventId}, StorageAccount={StorageAccount}, Container={Container}, BlobName={BlobName}",
                eventGridEvent.Id,
                blob.StorageAccount,
                blob.ContainerName,
                blob.BlobName);

            // 許可されないイベントをどう扱うかは運用設計次第。
            // ここでは Event Grid の再試行や意図しない dead-letter を避けるため正常終了している。
            return;
        }

        var jobId = HashHelper.ToBlobJobId(
            blob,
            eventGridEvent.EventType);

        var message = new BlobEventJobMessage(
            JobId: jobId,
            EventId: eventGridEvent.Id,
            EventType: eventGridEvent.EventType,
            Subject: eventGridEvent.Subject,
            Blob: blob,
            EnqueuedAt: DateTimeOffset.UtcNow);

        var payloadJson = JsonSerializer.Serialize(
            message,
            JsonSupport.Options);

        var decision = await _jobs.TryCreateEventJobAndOutboxAsync(
            jobId: jobId,
            eventId: eventGridEvent.Id,
            eventType: eventGridEvent.EventType,
            blob: blob,
            queueConnectionName: QueueConnectionName,
            queueName: QueueName,
            payloadJson: payloadJson,
            cancellationToken: cancellationToken);

        if (decision.Kind == AcceptDecisionKind.AlreadyAccepted)
        {
            _logger.LogInformation(
                "Duplicate blob event ignored. JobId={JobId}, EventId={EventId}",
                decision.JobId,
                eventGridEvent.Id);

            return;
        }

        _logger.LogInformation(
            "Event Grid event converted to job and outbox. JobId={JobId}, EventId={EventId}, Subject={Subject}, BlobName={BlobName}",
            decision.JobId,
            eventGridEvent.Id,
            eventGridEvent.Subject,
            blob.BlobName);
    }

    private static BlobReference? TryParseBlobReference(EventGridEvent eventGridEvent)
    {
        try
        {
            using var doc = JsonDocument.Parse(eventGridEvent.Data.ToString());

            if (!doc.RootElement.TryGetProperty("url", out var urlProperty))
            {
                return null;
            }

            var url = urlProperty.GetString();

            if (string.IsNullOrWhiteSpace(url))
            {
                return null;
            }

            if (!Uri.TryCreate(url, UriKind.Absolute, out var uri))
            {
                return null;
            }

            // 説明用の簡略実装。
            // sovereign cloud / private endpoint / custom domain / dfs endpoint を使う場合は、
            // host の解析を環境に合わせて実装する。
            const string expectedSuffix = ".blob.core.windows.net";

            if (!uri.Host.EndsWith(expectedSuffix, StringComparison.OrdinalIgnoreCase))
            {
                return null;
            }

            var storageAccount = uri.Host[..^expectedSuffix.Length];

            if (string.IsNullOrWhiteSpace(storageAccount))
            {
                return null;
            }

            var path = uri.AbsolutePath.TrimStart('/');
            var parts = path.Split('/', 2);

            if (parts.Length != 2)
            {
                return null;
            }

            var containerName = Uri.UnescapeDataString(parts[0]);
            var blobName = Uri.UnescapeDataString(parts[1]);

            if (string.IsNullOrWhiteSpace(containerName) ||
                string.IsNullOrWhiteSpace(blobName))
            {
                return null;
            }

            string? etag = TryGetStringProperty(doc.RootElement, "eTag")
                ?? TryGetStringProperty(doc.RootElement, "etag");

            string? versionId = TryGetStringProperty(doc.RootElement, "versionId");

            return new BlobReference(
                StorageAccount: storageAccount,
                ContainerName: containerName,
                BlobName: blobName,
                ETag: etag,
                VersionId: versionId);
        }
        catch (JsonException)
        {
            return null;
        }
        catch (UriFormatException)
        {
            return null;
        }
        catch (ArgumentException)
        {
            return null;
        }
    }

    private static string? TryGetStringProperty(
        JsonElement element,
        string propertyName)
    {
        if (!element.TryGetProperty(propertyName, out var property))
        {
            return null;
        }

        return property.ValueKind == JsonValueKind.String
            ? property.GetString()
            : null;
    }
}

Event Grid Trigger 側で OCR、画像変換、DB 更新、外部 API 呼び出しを全部やらない。

これがかなり大事です。

また、Event Grid の再配信重複だけを抑えるなら eventGridEvent.Id でよいですが、同じ Blob バージョンを 1 回だけ処理したい場合は、storageAccount + containerName + blobName + eTag/versionId + eventType から業務上の jobId を生成します。

トリガー別の失敗パターンまとめ

HTTP Trigger の失敗パターン

失敗	症状	対策
長時間処理を同期実行	230 秒付近で 502 / タイムアウト	受付だけして Queue / Durable Functions
request body size 未制限	巨大入力でメモリ圧迫	サイズ上限を設ける
JSON 不正を catch しない	400 ではなく 500 になる	`JsonException` を捕捉する
`Idempotency-Key` を受け取るだけ	同じ Queue メッセージが何度も積まれる	受付レコードを atomic に作成する
key をグローバル扱い	別 tenant / user / API と衝突する	tenant / endpoint / user でスコープ化
raw body hash	空白や property order で false conflict	canonical JSON から hash 化
同じ key で異なる本文を許す	クライアントバグや攻撃を見逃す	request hash を保存し、違えば 409
`RequestedBy` を本文から信用	監査ログ上のなりすまし	principal / claim から決定する
Function Key を業務認可扱い	利用者本人性を誤認する	Entra ID / APIM / JWT などで認証認可
Queue へ直接送信	Queue 送信と DB 更新の曖昧状態	Outbox パターンを使う
Queue 送信成功後に DB 更新失敗	Queue にはあるが DB は未送信扱い	`DispatchUnknown` / 再送 / 冪等 Worker
例外時の状態管理がない	「失敗したが途中まで成功」になる	job status を持つ
`WriteAsJsonAsync` の挙動差を知らない	想定外の 200 OK になる	ステータスコードを明示する overload を使う
任意の Blob URL を処理	外部 URL 参照、SAS 露出	構造化 Blob 参照と許可ポリシーを使う
Blob 存在確認なし	受付後に Worker で失敗	受付時と Worker 実行時に検証

HTTP Trigger は「処理する場所」ではなく、処理を受け付ける窓口として設計すると安定します。

Outbox の失敗パターン

失敗	症状	対策
Job と Outbox が別トランザクション	Job だけ作られて送信されない	同一 DB トランザクションで保存
Queue SDK 例外を未送信扱い	実は Queue に入っているのに再送制御を誤る	送信試行後は `DispatchUnknown`
Queue 送信後の DB 更新失敗	Queue にはあるが Outbox は未完了	`DispatchUnknown` として再照合
1 件失敗で全件停止	Outbox が滞留する	item ごとに try/catch する
再送重複を考えていない	Worker が二重処理	Worker 側を `jobId` で冪等にする
Outbox 滞留を監視しない	受付済みなのに処理されない	`Pending` / `Unknown` 滞留時間にアラート

Queue Trigger の失敗パターン

失敗	症状	対策
冪等性なし	再試行で重複更新	`jobId` / `eventId` で排他
事前確認だけの冪等性	並行重複で両方処理される	atomic な条件付き更新にする
`false` で即 return	Worker クラッシュ後に Queue メッセージを削除	`Acquired` / `Completed` / `CurrentlyProcessing` を分ける
Processing の期限切れなし	クラッシュ後に永遠に止まる	lease を持たせる
lease と visibilityTimeout 不整合	再試行しすぎ、または復旧が遅い	retry-later / watchdog / re-enqueue を設計
`ProcessAsync` 後の完了更新失敗を考えない	副作用が二重実行される	外部 API / DB / ファイル生成を冪等化
既定の並列度を知らない	DB / API が一気に詰まる	`batchSize` / `newBatchThreshold` 調整
poison queue を業務エラー扱い	原因切り分けを誤る	未処理・要調査メッセージとして扱う
poison queue 未監視	障害が静かに溜まる	poison queue にアラート
poison raw body をログ出力	個人情報や SAS が漏れる	hash と隔離 Blob を使う
巨大 JSON を投入	メッセージサイズ制限で失敗	Blob に置いて参照だけ送る
`batchSize = 1` で完全直列だと思う	スケールアウト時に複数実行	完全直列が必要なら別設計
順序保証を期待する	処理順が前後する	Service Bus Sessions などを検討
`AzureWebJobsStorage` を業務 Queue に流用	ホスト用 Storage と業務データ境界が曖昧	専用 Storage / 専用接続名を使う

Queue Trigger は便利ですが、At-Least-Once 前提です。

レビューでは必ず、

同じメッセージが 2 回、しかもほぼ同時に来ても壊れないか？

を確認します。

Event Grid Trigger の失敗パターン

失敗	症状	対策
重い処理を直接実行	30 秒超過でリトライ、重複処理	Queue に逃がす
順序を期待する	後続イベントが先に処理される	順序前提の設計にしない
dead-letter 未設定	配信不能イベントが消える	dead-letter Blob を設定
HTTP 4xx の扱いを知らない	再試行されず dead-letter / 破棄	ステータスコード別挙動を確認する
EventGridTrigger と Webhook を混同	コードで 400 を返すような説明になる	正常終了 / 例外 / invalid queue を設計する
`eventId` だけで業務重複排除	同じ Blob 実体を複数回処理	Blob identity から jobId を作る
ETag / VersionId 欠落を許す	別 Blob バージョンを同一 job 扱いする	invalid queue 退避、または eventId を key に含める
`eventId` を保存しない	再配信重複を検知できない	`eventId` も監査情報として保存
フィルタなし	不要イベントまで処理	Event Grid subscription filter を使う
Event Grid をジョブキュー扱いする	処理責任が曖昧になる	Queue / Service Bus に変換する
Blob URL をそのまま渡す	SAS 露出、権限境界が曖昧	構造化 Blob 参照に変換する

Event Grid は「イベント通知」です。

「この処理を必ず完了してほしい」というコマンドキューではありません。

監視設計：Functions は作ってからではなく、作る前に見る

Azure Functions の設計では、実装より先に監視を考えます。

最低限、以下を監視します。

対象	監視するもの
HTTP Trigger	5xx、429、平均応答時間、P95 / P99、リクエスト数
HTTP 受付	`Accepted` / `Conflict` / `Failed` の件数
Idempotency	`409 Conflict` 件数、同一 key 再送数
入力検証	JSON 不正、request body too large、Blob 不正
Outbox	`Pending` / `DispatchFailed` / `DispatchUnknown` / 滞留時間
Queue	キュー長、最古メッセージの滞留時間、poison queue 件数
Queue Trigger	失敗回数、リトライ増加、処理時間、下流 API 失敗
Queue Trigger	`CurrentlyProcessing` による retry-later 件数
Completion	`ProcessAsync` 成功後の `MarkCompleted` 失敗
Job 状態	`Processing` のまま lease 期限切れした job
Event Grid	配信失敗、dead-letter 件数、イベント数急増
DB / 外部 API	スロットリング、タイムアウト、接続数
セキュリティ	許可外 Blob 参照、403、Managed Identity 権限エラー

Application Insights のログ設計

構造化ログでは、最低限以下を入れます。

JobId
RunId
TenantId
RequestedByUserId
EventId
CorrelationId
StorageAccount
ContainerName
BlobName
QueueMessageId
DequeueCount
MessageHash
OutboxId
DispatcherId

悪い例です。

_logger.LogInformation("Job started: " + job.JobId);

良い例です。

_logger.LogInformation(
    "Job started. JobId={JobId}, RunId={RunId}, CorrelationId={CorrelationId}",
    job.JobId,
    runId,
    correlationId);

雑に文字列連結すると、あとで KQL が書きにくくなります。

また、SAS URL やアクセストークンをログに出さないようにします。

// 悪い例
_logger.LogInformation("BlobUrl={BlobUrl}", sasUrl);

// 良い例
_logger.LogInformation(
    "Blob target. StorageAccount={StorageAccount}, Container={Container}, BlobName={BlobName}",
    blob.StorageAccount,
    blob.ContainerName,
    blob.BlobName);

KQL 例：失敗したジョブを見る

traces
| where timestamp > ago(24h)
| where message has "Job failed"
| project timestamp, message, customDimensions
| order by timestamp desc

KQL 例：Outbox の曖昧状態を見る

traces
| where timestamp > ago(24h)
| where message has "Outbox dispatch state is unknown"
| project timestamp, message, customDimensions
| order by timestamp desc

KQL 例：Outbox 送信失敗を見る

traces
| where timestamp > ago(24h)
| where message has "Outbox dispatch failed"
| project timestamp, message, customDimensions
| order by timestamp desc

KQL 例：Queue Trigger の例外傾向を見る

exceptions
| where timestamp > ago(24h)
| where operation_Name has "ProcessImportFunction"
| summarize count() by type, bin(timestamp, 1h)
| order by timestamp desc

KQL 例：特定 jobId を追跡する

union traces, exceptions
| where timestamp > ago(24h)
| where tostring(customDimensions.JobId) == "imp_6d5f4c3a2b1e..."
| order by timestamp asc

KQL 例：CurrentlyProcessing による retry-later を見る

traces
| where timestamp > ago(24h)
| where message has "Job is still marked as Processing"
| summarize count() by tostring(customDimensions.JobId), bin(timestamp, 10m)
| order by timestamp desc

KQL 例：ProcessAsync 後の MarkCompleted 失敗を見る

traces
| where timestamp > ago(24h)
| where message has "side effects may have completed, but MarkCompleted failed"
| project timestamp, message, customDimensions
| order by timestamp desc

KQL 例：許可外 Blob 参照を検知する

traces
| where timestamp > ago(24h)
| where message has "Blob reference is not allowed"
| project timestamp, message, customDimensions
| order by timestamp desc

KQL 例：Poison message を見る

traces
| where timestamp > ago(24h)
| where message has "Poison message detected"
| project timestamp, message, customDimensions
| order by timestamp desc

Function App の分け方も設計に入れる

1 つの Function App に、HTTP Trigger、Queue Trigger、Event Grid Trigger を全部詰め込む構成はよくあります。

小規模なら問題ありません。

ただし、本番で負荷や障害特性が違う場合は分けます。

分ける理由は次です。

HTTP と Queue でスケール特性が違う
Queue の host.json 設定は Function App 内の Queue Trigger に影響する
Worker だけ Premium / Flex Consumption にしたい場合がある
HTTP のデプロイと Worker のデプロイを分けたい
障害時に API 受付だけは生かしたい
Queue Worker の暴走で HTTP API まで巻き込みたくない
CPU / メモリを使う処理を API と同居させたくない
ネットワーク制御や Managed Identity の権限を分けたい
Functions ホスト用 Storage と業務 Queue 用 Storage を分けたい
Outbox Dispatcher の再送ロジックを API / Worker と分離したい

Flex Consumption の per-function scaling だけに頼りすぎない

2026 年時点の Azure Functions では、Flex Consumption Plan において per-function scaling の考え方があります。

Flex Consumption では、従来より Function App 内のスケール分離が効きやすくなっています。

ただし、注意があります。

HTTP triggers は HTTP グループとして同じインスタンスを共有する
Blob storage Event Grid triggers は Blob グループとして同じインスタンスを共有する
Durable Functions triggers も Durable グループとして同じインスタンスを共有する
host.json、デプロイ、Managed Identity、ネットワーク境界、障害切り分けは Function App 単位で考える必要がある

したがって、Function App を分けるかどうかは、スケールだけではなく以下で判断します。

権限境界
運用境界
デプロイ境界
障害境界
監視境界
コスト管理
ネットワーク制御

現場では、最初から全部を分ける必要はありません。

ただし、以下の兆候が出たら分離を検討します。

Queue の滞留で HTTP API まで遅くなる
Worker のデプロイが API に影響する
Worker だけメモリ不足になる
DB への負荷を Worker 側で強く制御したい
認可やネットワーク境界を分けたい
障害時の切り分けが難しくなってきた
Managed Identity の権限が広がりすぎている
Outbox Dispatcher の再送が API の応答性に影響している

Durable Functions との使い分け

HTTP で開始し、長時間の状態管理が必要な場合は Durable Functions も候補です。

ざっくり使い分けるとこうです。

要件	推奨
単純な非同期ジョブ	HTTP + Outbox + Queue
状態遷移が少ないバッチ	Queue Trigger
複数ステップのワークフロー	Durable Functions
人手承認、長期待機	Durable Functions
fan-out / fan-in	Durable Functions
イベント通知を受けて単発処理	Event Grid + Outbox + Queue
厳密な順序処理	Service Bus Sessions など
lease / watchdog / re-enqueue が複雑化	Durable Functions を検討

Queue で頑張りすぎて、自前でワークフローエンジンを作り始めたら Durable Functions を検討するサインです。

たとえば、以下のような処理です。

1. CSV を受け付ける
2. ファイルを検証する
3. 顧客ごとに分割する
4. 並列で外部 API に連携する
5. 全件完了を待つ
6. 集計する
7. メール通知する
8. 失敗分だけ再実行できるようにする
9. Processing の lease を監視する
10. stuck job を再 enqueue する

これを Queue と DB の状態管理だけで作り込むと、だんだん独自ワークフローエンジンになります。

その場合は Durable Functions を検討した方が、実装も運用も見通しがよくなることがあります。

実務チェックリスト

HTTP Trigger チェック

230 秒以内に必ず応答できるか
request body size の上限を設けているか
JSON 不正時に 400 を返しているか
JsonSerializerOptions を明示しているか
request hash を canonical JSON から作っているか
失敗時にクライアントが再送しても安全か
Idempotency-Key または jobId があるか
Idempotency-Key / jobId に一意制約があるか
Idempotency-Key を tenant / endpoint / user でスコープ化しているか
request body hash を保存しているか
同じ key で本文が違う場合に 409 Conflict などを返すか
同じ Idempotency-Key では Job / Outbox を重複作成しないか
HTTP Trigger が Queue に直接送信していないか
Job + Outbox を同一 DB トランザクションで保存しているか
Queue 送信試行後の例外を DispatchUnknown として扱えるか
長時間処理を HTTP 内でやっていないか
受付後に 202 Accepted とステータス照会 URL を返しているか
WriteAsJsonAsync でステータスコードを明示しているか
認証レベルを Anonymous のまま放置していないか
Function Key を業務ユーザー認証の代替にしていないか
RequestedBy をリクエスト本文から信用していないか
principal / claim から user id を決定しているか
外部 API 呼び出しにタイムアウトを設定しているか
レスポンスのステータスコードと業務状態が一致しているか
二重クリック対策をしているか
任意の Blob URL を受け取っていないか
Blob 参照は許可された Storage Account / Container / Prefix のみに制限しているか
HTTP 受付時と Worker 実行時の両方で Blob を検証しているか
SAS URL をログに出していないか

Outbox チェック

Job と Outbox を同一 DB トランザクションで保存しているか
Outbox Dispatcher が lease 付きでレコードを取得しているか
複数 Dispatcher がいても同じ Outbox を同時送信しないか
Queue SDK 呼び出し前の失敗と呼び出し後の失敗を区別しているか
SendMessageAsync の例外を未送信と断定していないか
Queue 送信試行後の DB 更新失敗を Failed と断定していないか
DispatchUnknown / Dispatching 期限切れを監視しているか
再送による Queue 重複を Worker 側で吸収できるか
Outbox 滞留時間にアラートがあるか
retry_count / next_retry_at を持っているか
1 件失敗でバッチ全体を止めるか、item ごとに継続するか決めているか
Dispatcher の権限を Queue 送信に必要な範囲へ絞っているか

Queue Trigger チェック

メッセージサイズは 64 KiB 未満か
メッセージに巨大データを入れず Blob 参照にしているか
batchSize と newBatchThreshold を下流性能に合わせたか
1 VM あたりの最大同時処理数を batchSize + newBatchThreshold で見ているか
スケールアウト後の総並列数を見積もっているか
messageEncoding の拡張機能バージョン前提を確認しているか
送信側 SDK と host.json の encoding が一致しているか
同じメッセージが複数回実行されても壊れないか
同じ jobId が同時実行されても atomic に 1 つだけ処理されるか
TryAcquireProcessing が Acquired / Completed / CurrentlyProcessing / LeaseNotExpired を区別しているか
CurrentlyProcessing を無条件に正常終了扱いしていないか
Processing の lease 期限切れを設計しているか
visibilityTimeout、lease、maxDequeueCount の整合を取っているか
stuck job の watchdog / re-enqueue を検討しているか
完了更新が runId 条件付きになっているか
ProcessAsync の副作用が冪等か
外部 API に Idempotency-Key 相当を渡しているか
外部 API が冪等 key 非対応の場合、送信履歴テーブルを持っているか
poison queue を監視しているか
poison queue を業務エラーと決めつけていないか
poison raw body をログにそのまま出していないか
恒久エラーと一時エラーを分けているか
順序保証が必要なのに Storage Queue を使っていないか
重複排除が必要なのに Storage Queue だけで済ませていないか
DB 側に一意制約や排他制御があるか
Processing のまま止まるジョブを検知できるか
AzureWebJobsStorage を業務 Queue として流用していないか
専用 Storage / 専用接続名 / Managed Identity / RBAC を使っているか

Event Grid Trigger チェック

30 秒以内に応答できる軽い処理か
重い処理は Queue に逃がしているか
Event Grid の dead-letter を設定しているか
eventId を保存して再配信重複を検知できるか
同じ Blob バージョンを 1 回だけ処理したい場合、Blob identity から jobId を作っているか
ETag / VersionId が欠落した場合の扱いを決めているか
欠落時は invalid-event queue 退避、または eventId を key に含めるか
Event Grid 受信後に直接 Queue へ送らず、Outbox で曖昧状態を扱う設計にしているか
順序保証を前提にしていないか
イベントフィルタを設定しているか
200〜204 以外は失敗扱いになることを理解しているか
400 / 403 / 413 などが再試行されない可能性を理解しているか
[EventGridTrigger] では HTTP status を直接返す実装モデルではないことを理解しているか
無効イベントをログ、invalid queue、dead-letter のどこで扱うか決めているか
Blob の etag や versionId を保存しているか
後続処理の失敗を追跡できるか
Blob URL をそのまま Queue に渡していないか
許可された Storage Account / Container / Prefix のみ処理しているか
Blob prefix 判定を case-sensitive にしているか
URL parse を本番環境の cloud / endpoint 構成に合わせて実装しているか

まとめ

Azure Functions のトリガー選定は、次の一言に尽きます。

HTTP は受付、Queue は実務、Event Grid は通知。

現場で燃える設計は、だいたいこの境界が曖昧です。

HTTP Trigger で長時間処理をやると、230 秒制限とクライアント再送で燃えます。

HTTP の Idempotency-Key は、受け取るだけでは不十分です。tenant / endpoint / user / request hash でスコープ化し、同じ key では Job / Outbox を重複作成しない設計が必要です。

request hash は raw body の Trim() ではなく、DTO から canonical JSON を作って hash 化します。

RequestedBy を本文から信用するのも危険です。認証済み principal から user id を決定します。Function Key は業務ユーザー認証の代替ではありません。

HTTP Trigger で Queue に直接送信すると、Queue 送信成功後に DB 更新が失敗した場合に曖昧状態が発生します。実務では Job + Outbox を atomic に保存し、Dispatcher が Queue に送る構成を第一候補にします。

さらに、SendMessageAsync の例外を「未送信」と断定してはいけません。Queue SDK の送信呼び出し中に例外が発生した場合、クライアントは失敗として観測していても、サービス側ではメッセージ作成済みの可能性があります。送信試行後の例外は DispatchUnknown として扱い、再送による重複は Worker 側の jobId 冪等性で吸収します。

Queue Trigger は非同期化の王道ですが、冪等性と poison queue 監視がないと燃えます。

Queue の冪等性は、IsCompleted を見てから処理するだけでは足りません。同じ jobId が同時に来る前提で、atomic な条件付き更新、lease、runId を設計します。

特に、TryMarkProcessingAsync == false で常に return するのは危険です。AlreadyCompleted と CurrentlyProcessing を分けないと、Worker クラッシュ後に再表示された Queue メッセージを削除してしまい、DB 上の Processing だけが残る可能性があります。

また、ProcessAsync が外部 API 呼び出し、DB 更新、ファイル生成などの副作用を完了した後に MarkCompletedAsync が失敗することもあります。そのため、ProcessAsync 内の副作用は jobId / runId / 外部 API の idempotency key / DB 一意制約で守り、完了状態の更新失敗による再実行でも結果が壊れないようにします。

poison queue は「業務エラー」ではなく、規定回数失敗した未処理メッセージの隔離先です。原因はコード不具合、構成ミス、権限不足、一時障害、業務エラーなどさまざまです。

Event Grid Trigger はイベント反応には便利ですが、重い処理を直接やると 30 秒応答・再試行・重複で燃えます。

Event Grid の eventId は再配信重複の検知には使えますが、同じ Blob バージョンを 1 回だけ処理したいなら、Blob identity から業務上の jobId を生成します。ETag / VersionId が取れない場合は、invalid-event queue へ退避するか、eventId を key に含めるなど、誤った重複排除を避ける設計が必要です。

Azure Functions はサーバーレスなので、サーバー管理は楽になります。

でも、リトライ、重複、順序、タイムアウト、監視、権限境界の設計は楽になりません。

むしろ、そこを曖昧にしたまま本番投入すると、オンプレより静かに、速く、広範囲に燃えます。

参考リンク

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

Azure Functionsの設計：トリガー選定と失敗例（Queue / HTTP / Event Grid）

Azure Functionsの設計：トリガー選定と失敗例（Queue / HTTP / Event Grid）

はじめに

この記事で扱うこと

先に結論

判断フロー

安定しやすい基本構成

1. HTTP Trigger：一番便利で、一番事故る

HTTP Trigger が向いている処理

失敗例：HTTP Trigger で CSV 取込を同期処理した話

改善後の設計

HTTP の Idempotency-Key は「受け取るだけ」では不十分

Idempotency-Key はスコープ付きで扱う

request hash は raw body の Trim だけでは弱い

RequestedBy をリクエスト本文から信用しない

受付レコードと Queue 投入の注意点

Outbox パターンで考える

BlobUrl をそのまま受け取る設計は危険

HTTP 入力サイズと Blob 存在確認

HTTP Trigger の設計ルール

2. Queue Trigger：非同期処理の王道。ただし冪等性がないと地獄

Queue Trigger が向いている処理

poison queue は「業務エラー」とは限らない

Queue Trigger の重要な制約

失敗例：Queue にしたのに DB が燃えた話

batchSize = 1 にしても完全直列とは限らない

Queue Trigger の host.json 例

AzureWebJobsStorage を業務 Queue に使うのはサンプル限定にする

Queue メッセージに業務データを詰め込まない

Queue Trigger の冪等性は atomic に設計する

1. 冪等性キー

TryMarkProcessingAsync == false で即 return は危険

2. Processing の期限切れを設計する

3. ProcessAsync 成功後に MarkCompletedAsync が失敗することもある

4. 恒久エラーと一時エラーを分ける

poison queue の監視

3. Event Grid Trigger：軽く見えるが、実は重い設計判断

Event Grid Trigger が向いている処理

Event Grid の重要な性質

Event Grid の HTTP 配信仕様と EventGridTrigger の実装モデルを分けて考える

失敗例：Blob Created を Event Grid で受けて直接 OCR した話

改善後の設計

Event Grid の eventId だけでは業務的な重複排除として弱い場合がある

ETag / VersionId が取れないイベントをどう扱うか

Event Grid Trigger の注意点

実装例：HTTP / Event Grid で受けて Outbox 経由で Queue に逃がす

app settings 例

Program.cs

共通モデル

JSON サポート

PrincipalProvider

Blob 参照の許可ポリシー

BlobVerifier

QueueClientFactory

IJobRepository / IOutboxRepository の考え方

HashHelper

HTTP Trigger：Job + Outbox を作って 202 を返す

Outbox Dispatcher：送信試行後の例外は DispatchUnknown に寄せる

Queue Trigger：理由付きで処理権を判断する

poison queue handler：raw message をそのままログに出さない

Event Grid Trigger：イベントを Job + Outbox に変換するだけ

トリガー別の失敗パターンまとめ

HTTP Trigger の失敗パターン

Outbox の失敗パターン

Queue Trigger の失敗パターン

Event Grid Trigger の失敗パターン

監視設計：Functions は作ってからではなく、作る前に見る

Application Insights のログ設計

KQL 例：失敗したジョブを見る

KQL 例：Outbox の曖昧状態を見る

KQL 例：Outbox 送信失敗を見る

KQL 例：Queue Trigger の例外傾向を見る

KQL 例：特定 jobId を追跡する

KQL 例：CurrentlyProcessing による retry-later を見る

KQL 例：ProcessAsync 後の MarkCompleted 失敗を見る

KQL 例：許可外 Blob 参照を検知する

KQL 例：Poison message を見る

Function App の分け方も設計に入れる

Flex Consumption の per-function scaling だけに頼りすぎない

Durable Functions との使い分け

`batchSize = 1` にしても完全直列とは限らない

`TryMarkProcessingAsync == false` で即 return は危険