2026.09.03

サーバーレスで順序保証つきの直列処理を作る — SQS・Lambda・Step Functions を組み合わせた制御

numa

Contents

はじめに

はじめに

こんにちは、ぬまです。

サーバーレスというと「スケールする」「並列に処理できる」という文脈で語られることが多いですが、実際のシステムでは別の要件に出会うことがあります。

「処理を並列にしてはいけない。順番どおりに、1 本ずつ処理してほしい」 という要件です。

最近、同様の要件を持つファイル連携基盤の設計・構築に携わる機会がありました。外部システムから届くファイルを、順序を守って処理します。

送信元（テナント）ごとにファイルが届く
同一テナント内では、ファイルは届いた順に処理しなければならない
同じファイルを二重に処理してはならない
一方で、テナント間は互いにブロックせず並列に処理したい

この記事では、SQS・Lambda・Step Functions を組み合わせて、この「テナント内は直列、テナント間は並列」をどう実現したかを紹介します。

課題設定

前提となる要件を整理します。

外部システムは、連携ファイルを S3 に配置します。こちらはそのファイルを検知して、ファイル内のレコードを 1 件ずつ基幹システムに投入します。

このとき守らなければならない制約が 3 つありました。

順序保証: 同一テナント内では、ファイルが届いた順、かつファイル内のレコード順に処理する
二重処理の禁止: 同じファイルに対する処理が同時に 2 本走ると、競合・重複登録・データ不整合の原因になり得る
欠落の禁止: 処理中に届いたファイルも、捨てずに前の処理が終わってから必ず処理する

そして、テナントは複数あり、テナント間には依存関係がないため、あるテナントの処理が別のテナントの処理を待たせるのは避けたい、という条件も加わります。

つまり「全体としては並列、テナント内では完全に直列」という、粒度によって並列度が変わる制御が必要でした。

全体像

構成は次のとおりです。

外部システムが S3 にファイルを配置すると、EventBridge がそのイベントを受け取り、テナント別の FIFO キューへ送ります。ポイントは、FIFO キュー・逐次制御 Lambda・Step Functions のセットをテナントごとに 1 組ずつ持つことです。IaC（Terraform）の for_each でテナント設定の map からまとめて生成しているので、テナントが増えても設定を 1 エントリ追加するだけで対応できます。

テナント間の並列性は、セットのリソースをテナントごとに分けていることで自然に担保されるため、以降はテナント内の直列化だけを考えればよくなります。

なお、この記事の主題は順序保証です。S3、EventBridge、SQS、Lambda、Step Functions を組み合わせる構成では、実運用上、各サービス間の権限を最小権限で設計すること、ファイル内容や実行ログに機密情報が出力されないようにすることも重要です。

直列化を支える複数の仕組み

「直列にする」と一言で言っても、実際には粒度の異なる複数の「1 本化」が必要でした。この構成では、次の 4 つの仕組みで直列化しています。

仕組み	守っている粒度	実現手段
キュー	テナント内のイベント順序	SQS FIFO + MessageGroupId
消費	メッセージ処理の同時実行	reserved concurrency = 1 + batch_size = 1
ワークフロー起動	ワークフローの同時実行	実行中チェック + 意図的な失敗
ワークフロー内	分割ファイル・レコードの順序	Map の MaxConcurrency = 1

順番に説明していきます。

1: SQS FIFO + MessageGroupId

まず、イベントの到着順を守るために FIFO キューを使います。

resource "aws_sqs_queue" "tenant_event" {
  name                        = "${var.tenant}-event-queue.fifo"
  fifo_queue                  = true
  content_based_deduplication = true
  visibility_timeout_seconds  = 30
  message_retention_seconds   = 1209600 # 14 日
  receive_wait_time_seconds   = 20
}

FIFO キューは MessageGroupId 単位で順序を保証します。この構成では MessageGroupId をテナント固定にしているため、テナント内の全メッセージが 1 つのグループとして到着順に配信されます。content_based_deduplication = true にしているのは、同じ内容のメッセージが短時間に重複して入った場合に、キュー側で除外するためです。

なお、FIFO キューが保証するのは、キューに入ったあとの配信順です。Amazon S3 のイベント通知は、オブジェクトの発生順どおりに届くことは保証されていません。EventBridge 経由でも、発火の順序は担保されません。今回の要件では、同一テナントで同時間帯にイベントが集中することはないため、キュー到達前の順序は考慮対象から外しています。

message_retention_seconds を最大の 14 日にしている理由は、後述する「DLQ をあえて設けない」判断とセットです。

2: reserved concurrency = 1 , batch_size = 1

FIFO キューだけでは不十分です。FIFO はグループ内の「配信順」を守りますが、消費側の Lambda が複数同時に走れば処理の完了順は保証されません。

そこで、キューを消費する Lambda の同時実行数を 1 に固定します。

resource "aws_lambda_function" "sequential_controller" {
  # ...
  reserved_concurrent_executions = 1
}

resource "aws_lambda_event_source_mapping" "sqs_trigger" {
  event_source_arn = aws_sqs_queue.tenant_event.arn
  function_name    = aws_lambda_function.sequential_controller.arn
  batch_size       = 1
}

reserved_concurrent_executions = 1: この Lambda の同時実行数を 1 に制限する
batch_size = 1: 1 回の起動で 1 メッセージだけ処理する

これで「メッセージを 1 件ずつ、順番に処理する」状態になります。

3: 実行中チェックと「失敗＝待機」

ここまでで Lambda の直列化はできました。ただし、この Lambda の役割は Step Functions の起動（StartExecution）です。

StartExecution は非同期です。そのため、Lambda はすぐ終わるが、起動されたワークフローはまだ実行中という時間帯が生まれます。その間に次のメッセージが来ると、Lambda 自体は空いているので、ワークフローが 2 本並走してしまいます。

そこで、起動前にワークフローの実行状況を確認します。

def lambda_handler(event, context):
    running = sfn_client.list_executions(
        stateMachineArn=STATE_MACHINE_ARN,
        statusFilter="RUNNING",
        maxResults=1,
    )
    if running["executions"]:
        # 実行中なら「意図的に」例外を投げて Lambda を失敗させる
        raise WorkflowAlreadyRunningError()

    sfn_client.start_execution(
        stateMachineArn=STATE_MACHINE_ARN,
        name=execution_name,  # 後述: イベント ID で冪等化
        input=json.dumps(sfn_input),
    )

ここで重要なのは、実行中だった場合に例外を投げて意図的に Lambda を失敗させていることです。

SQS トリガーの Lambda が失敗すると、メッセージは削除されずキューに残り、visibility timeout（30 秒）が経過すると再配信されます。つまり、

visibility timeout が「空き確認のポーリング間隔」
SQS の再配信が「リトライ機構」

として機能します。「失敗」を「30 秒後にもう一度試すための待機」として使っています。

メッセージを削除せずキューに残すため、処理中に届いたファイルも捨てられません。前のワークフローが終わると、残っていたメッセージが順に処理されます。課題設定の「欠落の禁止」は、この動きで満たしています。

この再配信は、ワークフローが実行中のあいだ次のメッセージを待たせるためのものです。一方、StartExecution が成功した時点でメッセージは削除されるため、その後にワークフローが失敗しても、次のファイルは起動できます。ワークフローが失敗した時に後続を止める設計も考えられますが、今回は連携されるファイルの内容が保証されている前提だったため、この考慮は対象外にしています。ワークフローの失敗は、後述する ExecutionsFailed の監視で検知します。

list_executions と start_execution の間に別の起動が入る余地はあります。この競合は、後述する実行名による冪等化で吸収します。

この方式にする前に、代替案も検討しました。

Lambda 内で sleep して完了をポーリングする: 待っている間ずっと Lambda の実行時間に課金され、タイムアウト管理も複雑になるため不採用
DynamoDB で分散ロックを取る: ロックの解放漏れ（ワークフロー異常終了時など）への対処が必要になり、ロジックの複雑性が増すこと、管理するリソースとして DynamoDB が 1 つ増えるため不採用

「例外で終了してメッセージを残す」方式は、追加のリソースが不要で、状態管理をすべて SQS に寄せられるのが利点です。ただし後述するとおり、監視設計とのセットで考える必要があります。

なお、Lambda のタイムアウトは 20 秒に設定し、visibility timeout の 30 秒より必ず短くなるようにしています。逆転すると、処理中のメッセージが再配信されて二重処理の原因になります。

4: Map の MaxConcurrency = 1

ワークフローの中でも直列化は続きます。

連携ファイルは 1 ファイルに大量のレコードが含まれることがあります。そこで、ワークフロー内の分割 Lambda で、最大 100 レコードごとの小さなファイルに分割します。目的は、後続処理を Lambda のタイムアウトに収まる粒度にすることです。一方で、分割すると「分割ファイル間の順序」を守る必要が出てきます。

そこで、分割ファイルを処理する Map ステートに MaxConcurrency: 1 を指定します。

{
  "ProcessSplitFiles": {
    "Type": "Map",
    "ItemsPath": "$.splitResult.Payload.splitFiles",
    "MaxConcurrency": 1,
    "Iterator": {
      "StartAt": "InvokeRegisterApi",
      "States": {
        "InvokeRegisterApi": {
          "Type": "Task",
          "Resource": "arn:aws:states:::lambda:invoke",
          "Retry": [
            {
              "ErrorEquals": [
                "Lambda.ServiceException",
                "Lambda.TooManyRequestsException"
              ],
              "IntervalSeconds": 1,
              "MaxAttempts": 3,
              "BackoffRate": 2
            }
          ],
          "End": true
        }
      }
    }
  }
}

Map ステートはデフォルトでは要素を並列処理しますが、MaxConcurrency: 1 にすると配列の先頭から 1 つずつ順に処理します。さらに途中の分割ファイルで失敗した場合、後続の分割ファイルは実行されません。「3番目が失敗したのに4番目が登録されてしまう」という順序崩れや歯抜けを防げます。

最後に、分割ファイル内のレコードも順に処理します。

こうして、テナント → メッセージ → ワークフロー → 分割ファイル → レコードと、すべての粒度で「1 本ずつ・順番どおり」が貫かれます。

トレードオフ

この構成には、意図的に選んだ判断が 2 つ含まれています。どちらも理由とセットで紹介します。

DLQ をあえて設けない

SQS を使うなら DLQ（デッドレターキュー）を設定するのがよくある構成です。しかしこの構成では、順序保証を優先して、DLQ をあえて設けない選択をしました。

理由は単純で、先頭のメッセージだけが DLQ に退避されると、後続のメッセージが先に処理されて順序が崩れるからです。

例えば、ファイル A → B → C の順に届いたとして、A の処理がエラーになり maxReceiveCount 超過で DLQ に移動すると、B と C は A を追い越して処理されてしまいます。

到着順:     A ──> B ──> C

DLQ あり:   A ──(エラー)──> [DLQ]
                 B ──> 処理 ✓
                 C ──> 処理 ✓

結果の処理順:  B -> C   （A は DLQ に残ったまま）

少なくとも本構成では、エラーをDLQへ退避する設計は順序保証と相性が悪くなります。

そこで、

DLQ は設定しない。エラーになったメッセージも先頭に残り続ける
メッセージ保持期間を最大の 14 日にして、その間は再配信され続ける
恒久的なエラー（ファイル形式不正など）は、アラートを受けた運用者が原因を取り除く

という設計にしました。「後続を止めてでも順序を守る」ことを明示的に選んだ形です。その代わり、エラーを放置するとそのテナントの処理が最大 14 日間止まり続けるという副作用があるので、次章の監視とセットで成立する判断です。

Step Functions の StartExecution をイベント ID で冪等化する

「失敗＝待機」方式には、再配信が絡むエッジケースがあります。

StartExecution は成功したが、その直後に逐次制御 Lambda が異常終了した（タイムアウトなど）場合、メッセージは削除されず再配信されます。何も対策しないと、同じファイルに対してワークフローがもう 1 本起動してしまいます。

これは Step Functions の実行名の一意性で解決できます。Step Functions Standard Workflowでは、実行名を指定できますので、同じ実行名で再度StartExecutionを呼び出した場合、実行状態やinputの内容によって、既存実行として扱われる、またはExecutionAlreadyExistsが返ることになります。

本構成では、EventBridgeのイベントIDを実行名に使うことで、同一イベントに対する二重起動を抑制しています。

この一意制約は、実行中チェックと起動の間に別の起動が入った場合の競合も吸収します。

try:
    sfn_client.start_execution(
        stateMachineArn=STATE_MACHINE_ARN,
        name=event_id,  # EventBridge のイベント ID
        input=json.dumps(sfn_input),
    )
except sfn_client.exceptions.ExecutionAlreadyExists:
    # すでに起動済み = このメッセージのworkflowは完了している
    # 正常終了してメッセージを削除させる
    pass

重要なのは、ExecutionAlreadyExists をエラーではなく成功として扱うことです。「すでに起動済み」ということは、このメッセージがやるべきことは終わっています。逐次制御 Lambda が正常終了してメッセージを削除させれば、二重起動もメッセージの滞留も起きません。

ここで抑えている範囲を明確にしておきます。イベント ID で防げるのは、「同じ EventBridge イベントが、再配信などをきっかけにワークフローを 2 本起動してしまう」ことです。課題設定の「同じファイルを二重に処理してはならない」のうち、この仕組みが担っているのはイベント単位の冪等性です。

EventBridge のイベント ID はイベントごとに発行されます。そのため、次のようなケースは別イベントになり、実行名も別になります。

外部システムが同じファイルを再アップロードした（同一キーへの上書きを含む）
同じ内容のファイルが、別のオブジェクトとして届いた

これらは業務上「同じファイル」に見えても、イベント ID では同一と判定できません。再アップロードを「訂正版として再処理する」とみなすか、「重複として抑止する」とみなすかは、業務要件によって分かれます。ファイル単位の一意性（オブジェクトキー、ETag、内容のハッシュなど）まで担保したい場合は、実行名や別の冪等キーにその識別子を使う必要があります。今回の構成では、逐次制御 Lambda の再配信に起因する二重起動を対象として、イベント ID を選んでいます。

監視設計: 「意図的なエラー」とどう付き合うか

「失敗＝待機」方式の最大の副作用は、正常な待機でも Lambda の Errors メトリクスが増えることです。

前のワークフローの実行中にファイルが届けば、逐次制御 Lambda は何度か失敗してから成功します。これは設計どおりの動きですが、Lambda の Errors にアラームを張っていると、正常運転でアラートが発生し続けます。

そこで、この逐次制御 Lambda に限ってはこのように考えました。

Errors メトリクスのアラームは張らない（偽陽性を発生させてしまうため）
待機による意図的な失敗は INFO レベルでログ出力し、本当の異常だけを ERROR レベルで出す
ERROR ログをメトリクスフィルターで拾ってアラートする
キューの滞留は ApproximateAgeOfOldestMessage（最古メッセージの経過時間）で監視し、閾値超過で「処理が詰まっている」ことを検知する
ワークフロー自体の失敗は Step Functions の ExecutionsFailed で検知する

「エラーを制御フローとして使う」設計を選んだ時点で、メトリクスの意味が変わります。アーキテクチャと監視はセットで決める必要があります。