AWS Lambda durable functions で非同期 API のポーリングを試す
本記事はサーバーワークス Advent Calendar 2025 (シリーズ 1)の16日目の記事です。
はじめに
サーバーワークスの宮本です。本記事では re:Invent 2025 で発表された AWS Lambda durable functions のユースケースの 1 つである、非同期 API のポーリング実装について試してみた結果を共有します。
AWS Lambda durable functions とは
AWS Lambda durable functions （以下、durable functions）の概要については以下の記事をご覧ください。
durable functions 登場前
durable functions 登場前は、非同期 API のポーリングは Step Functions で以下のようなフローを作成することで実現していました。
Start job で非同期 API の呼び出し、Get Job Status で実行ステータスの確認、Job Complete? でジョブが終了していればJob Succeeded、実行中であれば再びWait X Seconds に戻るような流れです。API を呼び出して結果を確認するだけにも関わらず、このようなポーリングフローの実装が必要でした。
※ 例外的にStateMachine から.sync で同期呼び出し可能な API もありますが、対応サービスが限定的であるため、ポーリングフローの実装が必要なケースが多いです
durable functions でポーリングフローを実装してみる
Step Functions でのポーリングフローの代わりに、durable functions で同様のポーリングを実現することができます。本記事では例として EC2 インスタンスの起動し、ステータスがrunning 状態になるまで待機するフローを durable functions で実装してみます。（実際にこのようなフローの出番が必要になるかはさておき...）
なお、実現したいことを Step Functions で実装すると以下のようなイメージになります。
実装してみた結果
以下のような実装になりました。Python では durable functions の機能を利用するためのSDKaws/aws-durable-execution-sdk-python が用意されているため、これを使用しました。
ポイントとしては、36 行目のwait_for_condition を使用することです。
result = context.wait_for_condition( check=check_status, config=WaitForConditionConfig( initial_state={"instance_id": instance_id}, wait_strategy=wait_strategy, ), )引数check には非同期 API のステータスをチェックする関数を指定します。今回は以下のように実装しました。
defcheck_status(state: State, context: WaitForConditionCheckContext) -> State:print(f"ステータスチェック開始: {state}") res = ec2.describe_instances(InstanceIds=[state["instance_id"]]) instance_state = res["Reservations"][0]["Instances"][0]["State"]["Name"] state = {"instance_id": state["instance_id"],"status": instance_state}print(f"ステータスチェック結果: {state}")return state
第一引数と戻り値には、ステータス情報を管理するための任意の型を指定します。今回は TypedDict でState を定義しました（20 行目）。この中に実際のステータスと、ステータスを問い合わせるためのキー情報を含めておくと良いでしょう。注意点として、この型はシリアライズ・デシリアライズ可能なものでなければなりません。durable functions では各ステップの結果をチェックポイントとして保存し、再実行する際に復元するような動きをするため、このような制約があります。
引数config にはWaitForConditionConfig のインスタンスを指定します。
引数initial_state にはcheck_status を初回実行する際の第一引数を指定します。初回はstatus の状態がないので、instance_id のみとしています。
引数wait_strategy は、Callable[[T, int], WaitForConditionDecision] を満たす関数を定義する必要があります。他にも方法があるようですが、今回はシンプルに以下の関数を定義しました。
defwait_strategy(state: State, attempt:int) -> WaitForConditionDecision:if state.get("status") =="running":print("インスタンスが running 状態になりました。ポーリングを終了します")return WaitForConditionDecision.stop_polling()else:print("インスタンスが running 状態ではありません。ポーリングを継続します")return WaitForConditionDecision.continue_waiting(Duration.from_seconds(10))
ステータスがrunning になった場合はWaitForConditionDecision.stop_polling() でポーリングを停止します。それ以外の場合はWaitForConditionDecision.continue_waiting(Duration.from_seconds(10)) としてポーリングを継続します。その際Duration.from_seconds(10) で次のステータスチェックまで 10 秒間待つ設定としています。
なお、WaitForConditionDecision にはオプショナルでserdes を指定することができます。標準でシリアライズ・デシリアライズ出来ない型を使用する場合は、この引数に自前のシリアライズ・デシリアライズ処理を実装することになります。
実行してみた結果
以下が作成したコードをデプロイし実行した結果です。
- ①
start_instance実行 - ②
check_statusを実行し、ステータスがrunningではなかったので 10 秒待機してポーリング継続 - ③
check_statusを実行し、ステータスがrunningではなかったので 10 秒待機してポーリング継続 - ④
check_statusを実行し、ステータスがrunningではなかったので 10 秒待機してポーリング継続 - ⑤
check_statusを実行し、ステータスがrunningだったのでポーリング停止
以下が実行ログです。
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| timestamp | message ||---------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|| 1765804990294 | {"time":"2025-12-15T13:23:10.294Z","type":"platform.initStart","record":{"initializationType":"on-demand","phase":"init","runtimeVersion":"python:3.14.DurableFunction.v6","runtimeVersionArn":"arn:aws:lambda:us-east-1::runtime:0cc0256109c9e7966dd2d347d81048fd9cc0de176d12c2d2150d4847a4874e4e","functionName":"myDurableFunction","functionVersion":"$LATEST","instanceId":"2025/12/15/[$LATEST]886e855a4ca640178f0239e6a2ac27a4","instanceMaxMemory":134217728}} || 1765804990812 | {"timestamp": "2025-12-15T13:23:10Z", "level": "INFO", "message": "Found credentials in environment variables.", "logger": "botocore.credentials", "requestId": ""} || 1765804991172 | {"time":"2025-12-15T13:23:11.172Z","type":"platform.start","record":{"requestId":"2a490011-c440-4823-a325-81d3c0eb31e2","functionArn":"arn:aws:lambda:us-east-1:123456789012:function:myDurableFunction:$LATEST","version":"$LATEST"}} || 1765804991409 | Durable function の実行を開始します |①start_instance②| 1765804993229 | ステータスチェック開始: {'instance_id': 'i-04098d724fb976e7b'} || 1765804993491 | ステータスチェック結果: {'instance_id': 'i-04098d724fb976e7b', 'status': 'pending'} || 1765804993491 | インスタンスが running 状態ではありません。ポーリングを継続します || 1765804993805 | {"time":"2025-12-15T13:23:13.805Z","type":"platform.report","record":{"requestId":"2a490011-c440-4823-a325-81d3c0eb31e2","metrics":{"durationMs":2631.396,"billedDurationMs":3508,"memorySizeMB":128,"maxMemoryUsedMB":102,"initDurationMs":875.685},"status":"success"}} || 1765805003941 | {"time":"2025-12-15T13:23:23.941Z","type":"platform.start","record":{"requestId":"b6b8ca28-cd0a-4cc2-ad5e-741b850394ce","functionArn":"arn:aws:lambda:us-east-1:123456789012:function:myDurableFunction:$LATEST","version":"$LATEST"}} |③| 1765805004009 | Durable function の実行を開始します || 1765805004009 | ステータスチェック開始: {'instance_id': 'i-04098d724fb976e7b', 'status': 'pending'} || 1765805004240 | ステータスチェック結果: {'instance_id': 'i-04098d724fb976e7b', 'status': 'pending'} || 1765805004240 | インスタンスが running 状態ではありません。ポーリングを継続します || 1765805004786 | {"time":"2025-12-15T13:23:24.786Z","type":"platform.report","record":{"requestId":"b6b8ca28-cd0a-4cc2-ad5e-741b850394ce","metrics":{"durationMs":844.266,"billedDurationMs":845,"memorySizeMB":128,"maxMemoryUsedMB":102},"status":"success"}} || 1765805014916 | {"time":"2025-12-15T13:23:34.916Z","type":"platform.start","record":{"requestId":"8c47318d-f716-4d77-98b4-88ba093df843","functionArn":"arn:aws:lambda:us-east-1:123456789012:function:myDurableFunction:$LATEST","version":"$LATEST"}} |④| 1765805015010 | Durable function の実行を開始します || 1765805015010 | ステータスチェック開始: {'instance_id': 'i-04098d724fb976e7b', 'status': 'pending'} || 1765805015311 | ステータスチェック結果: {'instance_id': 'i-04098d724fb976e7b', 'status': 'pending'} || 1765805015311 | インスタンスが running 状態ではありません。ポーリングを継続します || 1765805015833 | {"time":"2025-12-15T13:23:35.833Z","type":"platform.report","record":{"requestId":"8c47318d-f716-4d77-98b4-88ba093df843","metrics":{"durationMs":916.524,"billedDurationMs":917,"memorySizeMB":128,"maxMemoryUsedMB":102},"status":"success"}} || 1765805025938 | {"time":"2025-12-15T13:23:45.938Z","type":"platform.start","record":{"requestId":"ed50c52c-a077-4be5-9512-6a72c77810c7","functionArn":"arn:aws:lambda:us-east-1:123456789012:function:myDurableFunction:$LATEST","version":"$LATEST"}} |⑤| 1765805026029 | Durable function の実行を開始します || 1765805026049 | ステータスチェック開始: {'instance_id': 'i-04098d724fb976e7b', 'status': 'pending'} || 1765805026329 | ステータスチェック結果: {'instance_id': 'i-04098d724fb976e7b', 'status': 'running'} || 1765805026349 | インスタンスが running 状態になりました。ポーリングを終了します || 1765805026710 | Durable function の実行を終了します || 1765805026830 | {"time":"2025-12-15T13:23:46.830Z","type":"platform.report","record":{"requestId":"ed50c52c-a077-4be5-9512-6a72c77810c7","metrics":{"durationMs":891.772,"billedDurationMs":892,"memorySizeMB":128,"maxMemoryUsedMB":102},"status":"success"}} |------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------初回のステータスチェックのログはステータスチェック開始: {'instance_id': 'i-04098d724fb976e7b'} となっており、initial_state に指定した値がstatus_check 関数に渡されていることがわかります。
また、注目すべきはDurable function の実行を開始します のログが複数回出力されていることです。durable functions の以下動作原理の通り、Wait 後は関数の再実行が行われているため、ハンドラー冒頭のログが都度出力されています。
おわりに
durable functions で非同期 API のポーリングを試してみました。SDKaws/aws-durable-execution-sdk-python はまだドキュメントが充実していないため、現状はソースコードを確認しながら実装する必要があると感じました。（DeepWiki が大いに役立ちました）
少し慣れが必要ですが、Lambda 1 つでポーリングを実現できるため、今後のアーキテクチャ設計で選択肢の 1 つになりそうです。
