SQSをローカルでテストするための互換シミュレーター、 lqs を公開した。名前は local queue service の略。

AWS SDKから接続できるHTTP APIを用意していて、StandardキューとFIFOキューの送受信や、可視性タイムアウト、DLQへの移動などを手元で試せる。

一旦、用意していたissueが全部片付いたので公開することにした。

SQSをローカルで試したかった

SQSを使うアプリケーションを書いていると、キューへ送ったメッセージをワーカーが受け取るところまで、ローカルで動かしたくなる。

単に送信と受信が成功するだけでなく、処理中にワーカーが落ちたらどうなるか、同じメッセージを再送したらどうなるかも確認したい。そのたびにAWS上のキューを用意するのも少し面倒だった。

ローカルでSQSを使うためのものは既にあって、 LocalStack や ElasticMQ がある。

LocalStackはSQSを含む複数のAWSサービスをローカルで扱う環境で、ElasticMQはSQS互換インターフェースを持つメッセージキュー。SQSのエミュレーターを探すと、このあたりが出てくる。

自分でも、SQSを使う処理の開発・テストに使えるものを作ってみたくなった。そこから作り始めたのがlqsになる。

AWS SDKから接続できるHTTP API

lqsはローカルでHTTPサーバーとして起動する。アプリケーション側では、SQSクライアントの接続先をlqsへ向けて使う。

既定の接続先はhttp://127.0.0.1:9324。

キューの作成、メッセージの送信・受信・削除といった操作をHTTP APIから実行できる。JSON形式とSQS Query形式の両方に対応している。

実際にAWS SDK for Go v2からサーバーへ接続する結合テストも用意した。送受信だけでなく、FIFOやDLQ、エラー応答なども確認している。

署名の検証は行わないので、SDKに設定する認証情報はローカル用のダミー値でよい。テストではアクセスキーとシークレットにそれぞれtest、リージョンにus-east-1を指定している。

今できること

主な対応機能はこんな感じ。

機能試せること
Standard / FIFOメッセージの送受信、FIFOのグループ内の順序制御
可視性タイムアウト受信後の一時的な非表示、期限切れ後の再受信、期限の変更
FIFOの重複排除IDや本文を使った送信の重複排除
ロングポーリングメッセージの到着を待って受信
DLQ受信回数の上限に達したメッセージの移動
遅延・保持期限すぐに配信しないメッセージや、期限切れの扱い
バッチ操作複数メッセージの送信・削除・可視性変更
属性・キュー管理メッセージ属性、キュー設定、一覧、タグ、パージなど

キューの設定やメッセージはローカルに保存できるので、再起動して続きを確認することもできる。

処理に失敗したときの再配信とDLQ

キューから受信しただけでは、メッセージは削除されない。処理が終わってから削除する。

途中でワーカーが落ちて削除できなかった場合は、可視性タイムアウトが切れると再び受信できる。この動きをローカルでも試せるようにした。

何度やっても処理できないメッセージは、設定した受信回数の上限に達したあと、再び受信対象になるタイミングでDLQへ移す。

正常なメッセージだけを流していると気づかないこともあるので、こういう失敗側の動きを試せるのは欲しかったところ。

FIFOの順序と重複排除

FIFOでは、同じグループのメッセージの順序を維持しながら、別のグループの処理を進められる。

例えば注文ごとにグループを分ければ、同じ注文のイベントを順に処理しつつ、別の注文も扱える。

送信の重複排除もあり、5分間の重複排除期間内に同じIDで再送しても、新しいメッセージを追加しない。本文から重複を判定する設定も使える。

再送してしまった場合や、先行する処理が止まった場合にアプリケーションがどう動くかを確認するために使う想定。

ローカルで起動する

Rust / Cargoのedition 2024に対応したビルド環境があれば、リポジトリを取得して起動できる。

git clone https://github.com/ieee0824/lqs.git
cd lqs
cargo run --locked

別のターミナルから、起動したことを確認する。

curl --fail -i http://127.0.0.1:9324/health

HTTP 200が返れば起動している。あとは利用するAWS SDKのエンドポイント設定をhttp://127.0.0.1:9324へ変更して接続する。

送受信に使うQueue URLは、CreateQueueまたはGetQueueUrlの戻り値を使う。SDKの接続先だけでなく、操作対象のキューURLもローカルのものにする。

待ち受けアドレスや保存先は環境変数で変更できる。

環境変数用途既定値
LQS_BIND_ADDR待ち受けアドレス127.0.0.1:9324
LQS_BASE_URL返されるキューURLのベース待ち受けアドレスから生成
LQS_DATABASE_PATH状態を保存するファイルlqs.sqlite

SDK側の設定例は Go SDKの結合テスト にも置いている。

Rustのテストから直接使うこともできる

HTTPサーバーを起動せず、Rustライブラリとして使う方法もある。テストごとにインメモリのDBを用意して、送信・受信・削除を試せる。

時刻を指定して操作できるので、可視性タイムアウトや重複排除の期限を、実際に何分も待たずに確認できる。ただし、ロングポーリングで待つ動きはHTTP側の機能で、ライブラリの同期受信は待機しない。

直接使う場合の例は READMEのRustライブラリの項目 に載せている。

本物のSQSとの違いもある

用意したissueは片付いたけれど、SQSのすべてを再現したという意味ではない。

AWSのリージョンごとのスループットや分散環境の挙動までは再現しない。キューの設定変更や件数の反映など、本物のSQSとタイミングが違う部分もある。

DLQへの移動には対応しているが、HTTP APIで非同期に再投入するmove-task系の操作は未実装。

また、ポリシーの一部やSSE/KMSの設定を扱う機能はあるものの、AWSのIAM全体や実際の暗号化を再現しているわけではない。保存されるメッセージは平文で、認証も本番サービス相当ではないので、ローカルの開発・テスト用として使う。

ポリシー未設定のキューはアクセスを制限しない。インターネットや、信頼できないクライアントへサーバーを公開する用途には使わない。

対応範囲と細かな違いは SQS互換仕様・制約 にまとめている。

まずは自分の手元で、送って、受け取って、失敗したらもう一度処理するところを試せるようになった。今後も実際に使いながら、足りないところを直していくつもり。