From 63b0a2e800279fcd7af80cf8a4879a36134e7ac4 Mon Sep 17 00:00:00 2001 From: Ferhat Elmas Date: Mon, 9 Nov 2026 01:02:13 +0100 Subject: [PATCH 1/2] feat: add rust workspace with queue crate Signed-off-by: Ferhat Elmas --- .docker/docker-compose-infra.yml | 7 + .dockerignore | 3 +- .github/workflows/rust.yml | 56 + rust/.gitignore | 1 + rust/Cargo.lock | 1424 ++++++++++++++++++++++ rust/Cargo.toml | 8 + rust/README.md | 16 + rust/crates/queue/Cargo.toml | 12 + rust/crates/queue/src/lib.rs | 129 ++ rust/crates/queue/src/sqs.rs | 322 +++++ rust/crates/queue/tests/contract.rs | 334 +++++ rust/crates/queue/tests/failure_paths.rs | 317 +++++ 12 files changed, 2628 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/rust.yml create mode 100644 rust/.gitignore create mode 100644 rust/Cargo.lock create mode 100644 rust/Cargo.toml create mode 100644 rust/README.md create mode 100644 rust/crates/queue/Cargo.toml create mode 100644 rust/crates/queue/src/lib.rs create mode 100644 rust/crates/queue/src/sqs.rs create mode 100644 rust/crates/queue/tests/contract.rs create mode 100644 rust/crates/queue/tests/failure_paths.rs diff --git a/.docker/docker-compose-infra.yml b/.docker/docker-compose-infra.yml index aadabaeb9..056a84951 100644 --- a/.docker/docker-compose-infra.yml +++ b/.docker/docker-compose-infra.yml @@ -164,6 +164,13 @@ services: done " + elasticmq: + profiles: + - sqs + image: softwaremill/elasticmq-native:1.7.1 + ports: + - "9324:9324" + imgproxy: image: darthsim/imgproxy:v3.26.0 ports: diff --git a/.dockerignore b/.dockerignore index a0d218e33..bacd547c1 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,3 +1,4 @@ node_modules dist -.env \ No newline at end of file +.env +rust diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml new file mode 100644 index 000000000..e3fa1b84a --- /dev/null +++ b/.github/workflows/rust.yml @@ -0,0 +1,56 @@ +name: Rust + +on: + pull_request: + paths: + - "rust/**" + - ".github/workflows/rust.yml" + - ".docker/docker-compose-infra.yml" + push: + branches: + - master + paths: + - "rust/**" + - ".github/workflows/rust.yml" + - ".docker/docker-compose-infra.yml" + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + test: + name: Test / Rust + runs-on: blacksmith-4vcpu-ubuntu-2404 + timeout-minutes: 15 + defaults: + run: + working-directory: rust + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + - name: Start ElasticMQ + working-directory: . + run: docker compose --project-directory . -f ./.docker/docker-compose-infra.yml up -d elasticmq + - name: Install Rust toolchain + run: | + rustup toolchain install stable --profile minimal --component clippy,rustfmt + rustup default stable + - uses: Swatinem/rust-cache@d1527c8f67ef9af357290475d4af4b0fab8f12e0 # v2.9.3 + with: + workspaces: rust + - name: Format + run: cargo fmt --check + - name: Clippy + run: cargo clippy --locked --all-targets -- -D warnings + - name: Wait for ElasticMQ + run: timeout 30 sh -c 'until curl -s -o /dev/null http://127.0.0.1:9324; do sleep 1; done' + - name: Test + run: cargo test --locked + - name: Test SQS contract + run: cargo test --locked --test contract -- --ignored + env: + STORAGE_QUEUE_TEST_SQS_ENDPOINT: http://127.0.0.1:9324 diff --git a/rust/.gitignore b/rust/.gitignore new file mode 100644 index 000000000..eb5a316cb --- /dev/null +++ b/rust/.gitignore @@ -0,0 +1 @@ +target diff --git a/rust/Cargo.lock b/rust/Cargo.lock new file mode 100644 index 000000000..7a95615ab --- /dev/null +++ b/rust/Cargo.lock @@ -0,0 +1,1424 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "arc-swap" +version = "1.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db4c94f57a570fce5d381be3289b3f79aef6332d72948fe82ab5772e423ccf92" +dependencies = [ + "rustversion", +] + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "aws-credential-types" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "576e8da35751dc85409ec1cd15999f692c277a1fbe9336780fe6f15e49dd637d" +dependencies = [ + "aws-smithy-async", + "aws-smithy-runtime-api", + "aws-smithy-types", + "zeroize", +] + +[[package]] +name = "aws-lc-rs" +version = "1.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.47.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8bfbe32ac36b3c428696ce315464181bc4029d689fd58e3f2a4e58c252242f52" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", +] + +[[package]] +name = "aws-runtime" +version = "1.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a0cc2144962cc5e358467b16e3d82981b1d424a7c3d34602be0036103814a1a" +dependencies = [ + "aws-credential-types", + "aws-sigv4", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-types", + "bytes", + "bytes-utils", + "fastrand", + "http", + "http-body", + "percent-encoding", + "pin-project-lite", + "tracing", + "uuid", +] + +[[package]] +name = "aws-sdk-sqs" +version = "1.119.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "86e08dcb1abb0a0d3771fa7eac3af7550fd1708628e79f89db9bf41da8d2c030" +dependencies = [ + "arc-swap", + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-observability", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http", + "regex-lite", + "tracing", +] + +[[package]] +name = "aws-sigv4" +version = "1.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f385da9b7757ed4c8214797ad663ff534fb622006143530aa80e21f8e06ff85c" +dependencies = [ + "aws-credential-types", + "aws-smithy-http", + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes", + "form_urlencoded", + "hex", + "hmac", + "http", + "percent-encoding", + "sha2", + "time", + "tracing", +] + +[[package]] +name = "aws-smithy-async" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de681a9df7ff79823b759ee17061076ec049cd61a8fa28ece4293d89c9f6964e" +dependencies = [ + "futures-util", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "aws-smithy-http" +version = "0.64.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19af609dd5edf8b056cd0bb001c706c8b3b1458055328b334ec13d7f46f6aaa8" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes", + "bytes-utils", + "futures-core", + "futures-util", + "http", + "http-body", + "http-body-util", + "percent-encoding", + "pin-project-lite", + "tracing", +] + +[[package]] +name = "aws-smithy-http-client" +version = "1.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "859a81364f8185349c584fb41cd8765a400cf982f447adada590cf008a9f3b85" +dependencies = [ + "aws-smithy-async", + "aws-smithy-runtime-api", + "aws-smithy-types", + "h2", + "http", + "hyper", + "hyper-rustls", + "hyper-util", + "pin-project-lite", + "rustls", + "rustls-native-certs", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tower", + "tracing", +] + +[[package]] +name = "aws-smithy-json" +version = "0.65.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b45f9fc77ffe92c92c341165a01a565bd30c39ecf5aa8c0719b41719009604f9" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", +] + +[[package]] +name = "aws-smithy-observability" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a416257d7b1b0f32a517a470a8c1dd354d8c4aa697ccab81a11a1e6cfb7418e3" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-types", + "tracing", +] + +[[package]] +name = "aws-smithy-runtime" +version = "1.16.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cb80fb19c084da5fca9d797845735e184445e5b958599440d67529bf6ef1b64" +dependencies = [ + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-http-client", + "aws-smithy-observability", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "bytes", + "fastrand", + "http", + "http-body", + "http-body-util", + "pin-project-lite", + "tokio", + "tracing", +] + +[[package]] +name = "aws-smithy-runtime-api" +version = "1.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d06a8615310f03c73ea8eea607ee1a9cb17b7ebe055b9152feda1d86831c215c" +dependencies = [ + "aws-smithy-async", + "aws-smithy-runtime-api-macros", + "aws-smithy-types", + "bytes", + "http", + "pin-project-lite", + "tokio", + "tracing", + "zeroize", +] + +[[package]] +name = "aws-smithy-runtime-api-macros" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce237f890a2283c9b328955c3b0918d2d5bf66a1cb6ceccd36a0047246fbc6e6" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "aws-smithy-schema" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7ef30d39e131651359b60a1c560de9a9502a4c3754172ab2d6e8734fe21bcba" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-types", + "http", +] + +[[package]] +name = "aws-smithy-types" +version = "1.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c3a040d8b963838a01d7d239b0118357388d0ab881857bb5fc2dd780bb87957" +dependencies = [ + "base64-simd", + "bytes", + "bytes-utils", + "futures-core", + "http", + "http-body", + "http-body-util", + "itoa", + "pin-project-lite", + "ryu", + "serde", + "time", + "tokio", + "tokio-util", +] + +[[package]] +name = "aws-types" +version = "1.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ba0d970d3947bb064d2ae1d823efa186c1dbde88f88afa8f6549946e5d8deee" +dependencies = [ + "aws-credential-types", + "aws-smithy-async", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "rustc_version", + "tracing", +] + +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + +[[package]] +name = "base64-simd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "339abbe78e73178762e23bea9dfd08e697eb3f3301cd4be981c0f78ba5859195" +dependencies = [ + "outref", + "vsimd", +] + +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "bytes-utils" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dafe3a8757b027e2be6e4e5601ed563c55989fcf1546e933c66c8eb3a058d35" +dependencies = [ + "bytes", + "either", +] + +[[package]] +name = "cc" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ff2a8126fff0f8c839080d619675a585e61720bb12d2baceae45769fd66f155" +dependencies = [ + "find-msvc-tools", + "jobserver", + "libc", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" + +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", +] + +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "ctutils" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "261af4dc7a06ea8a527a8717d8cbba3417530d6ea14e1edb3de6b67806be474f" +dependencies = [ + "cmov", +] + +[[package]] +name = "deranged" +version = "0.5.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e9de72ce2ad1f90dc62fa25f0f430ef85eb4b0d8fa0be4f30373bc40a21d28e" + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer", + "const-oid", + "crypto-common", + "ctutils", +] + +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + +[[package]] +name = "either" +version = "1.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be" + +[[package]] +name = "equivalent" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826" + +[[package]] +name = "fastrand" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" + +[[package]] +name = "find-msvc-tools" +version = "0.1.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c66915cc5391097de7a2b0b972e8505aa33c0df401fd3a6811dbdb32881ba48" + +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", +] + +[[package]] +name = "h2" +version = "0.4.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d29020232d6aa3fb1daca64c1127cf662cf97f254ae16c18c05b8ab635fc118" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + +[[package]] +name = "hex" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" + +[[package]] +name = "hmac" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" +dependencies = [ + "digest", +] + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "hybrid-array" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" +dependencies = [ + "typenum", +] + +[[package]] +name = "hyper" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c3e324da4c95177d6291d4c8730197c0d1822f8a9766814a4a44fa5ab797c9c" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "h2", + "http", + "http-body", + "httparse", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-rustls" +version = "0.27.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfa8e654703247911e29c23fbeaa261834bd9bb74efba2f9acddc37bfb127f53" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "rustls-native-certs", + "tokio", + "tokio-rustls", + "tower-service", +] + +[[package]] +name = "hyper-util" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff" +dependencies = [ + "base64", + "bytes", + "futures-channel", + "futures-util", + "http", + "http-body", + "httparse", + "hyper", + "ipnet", + "libc", + "percent-encoding", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", +] + +[[package]] +name = "indexmap" +version = "2.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855" +dependencies = [ + "equivalent", + "hashbrown", +] + +[[package]] +name = "ipnet" +version = "2.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom 0.4.3", + "libc", +] + +[[package]] +name = "js-sys" +version = "0.3.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8dc09d2154098bbf4ef54eab6b117a3c24cbbffe94046b16ee6d6d4510584d3a" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + +[[package]] +name = "libc" +version = "0.2.191" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a" + +[[package]] +name = "mio" +version = "1.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + +[[package]] +name = "num-conv" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "openssl-probe" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e036b411754132fd6b74e94b74e9a13c5cf606fa93d1885c35860d3060820b1" + +[[package]] +name = "outref" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" + +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "pkg-config" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" + +[[package]] +name = "powerfmt" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a6394b9e965e73d0a289ee54f589087e2c676aedf60885baf52c76b771e4958" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "regex-lite" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973" + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + +[[package]] +name = "rustls" +version = "0.23.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf" +dependencies = [ + "aws-lc-rs", + "once_cell", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-native-certs" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a073f5dc7469f984c52ad2752b63b0807745133b6de880b7b64c1ac4c48aec4" +dependencies = [ + "openssl-probe", + "rustls-pki-types", + "schannel", + "security-framework", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" +dependencies = [ + "aws-lc-rs", + "ring", + "rustls-pki-types", + "untrusted", +] + +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + +[[package]] +name = "schannel" +version = "0.1.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9cb23c424d8ee841bb4d813677ca237e2a610067d594fec130f87d6093495ab5" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "security-framework" +version = "3.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" +dependencies = [ + "bitflags", + "core-foundation", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "semver" +version = "1.0.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "sha2" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d7069beb7d6ac7b9acd1039986e73443f24234f41074da099d6f994ac9ad19" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.16.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "149309498a5f7334b12ddae0cad8aba81ce785fe91755068867e017cf98fee46" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + +[[package]] +name = "storage-queue" +version = "0.1.0" +dependencies = [ + "aws-sdk-sqs", + "thiserror", + "tokio", +] + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thiserror" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "time" +version = "0.3.55" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134" +dependencies = [ + "deranged", + "num-conv", + "powerfmt", + "serde_core", + "time-core", + "time-macros", +] + +[[package]] +name = "time-core" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" + +[[package]] +name = "time-macros" +version = "0.2.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e689342a48d2ea927c87ea50cabf8594854bf940e9310208848d680d668ed85" +dependencies = [ + "num-conv", + "time-core", +] + +[[package]] +name = "tokio" +version = "1.54.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b52fd3255258d26fac7264ab5a76b53c13367136dc011d3d6ad4f8ed070396a3" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9cc2678c2cdd569ef8215e2afd7954ada2ae20b4fdd2c5fe6139a3b02d105db" +dependencies = [ + "rustls", + "tokio", +] + +[[package]] +name = "tokio-util" +version = "0.7.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e464cf451ba96ebfc6f9b6542f17ee8b8956e33f1e40d9690624e59d7a7f8a4b" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "tower-layer", + "tower-service", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + +[[package]] +name = "unicode-ident" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" + +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "uuid" +version = "1.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a06f08a856b009bc7bbaf7d3e79808595d032dd19074c0c2ef26e49e03e6bdc" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "vsimd" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c3082ca00d5a5ef149bb8b555a72ae84c9c59f7250f013ac822ac2e49b19c64" + +[[package]] +name = "want" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec4cdd0dd910afe868b7ef477227d8d538b46b3075031afee8a9f2acb0a2ed0b" +dependencies = [ + "try-lock", +] + +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "wasm-bindgen" +version = "0.2.130" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a0181c14ddd6c7b8b0b0132a5f1ea2f32be0d893cd3ea97b6e3b28886b8a3d1c" +dependencies = [ + "cfg-if", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.130" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e931a02a39abdbf66966fc35902b5ed58344750a70885bed4ebb3d8eaf28ed9" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.130" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ca4b54280e455bd831e33a6b8e734baf432fe63fbc35c42ac14ce67a043a53a" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn 3.0.9", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.130" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b1f9c3798725016c6ee5f2f57869ec0831e8bbc0ab0310a86e787e97f41389e" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "zeroize" +version = "1.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13084392c5e4bc371903e2935a5eaeed24905a7511356b883835e18a78f6879" diff --git a/rust/Cargo.toml b/rust/Cargo.toml new file mode 100644 index 000000000..6541faf58 --- /dev/null +++ b/rust/Cargo.toml @@ -0,0 +1,8 @@ +[workspace] +resolver = "3" +members = ["crates/queue"] + +[workspace.dependencies] +aws-sdk-sqs = { version = "1.119.0", default-features = false, features = ["behavior-version-latest", "default-https-client", "rt-tokio"] } +thiserror = "2.0.21" +tokio = { version = "1.54.1", features = ["rt-multi-thread", "macros", "time"] } diff --git a/rust/README.md b/rust/README.md new file mode 100644 index 000000000..4ad09cd00 --- /dev/null +++ b/rust/README.md @@ -0,0 +1,16 @@ +# Rust workspace + +`crates/queue` (`storage-queue`) provides transport primitives with SQS standard-queue semantics +(at-least-once delivery, visibility timeout, receive count, delay, redrive). It currently implements +the transport for AWS SQS. It does not provide event dispatch or job policies. + +The SQS contract tests are ignored by default and require an endpoint. From `rust/`: + + STORAGE_QUEUE_TEST_SQS_ENDPOINT=http://127.0.0.1:9324 \ + cargo test --test contract -- --ignored + +ElasticMQ (the SQS endpoint) is the `elasticmq` service in `.docker/docker-compose-infra.yml`. + +Start it from the repo root: + + docker compose --project-directory . -f ./.docker/docker-compose-infra.yml up -d elasticmq diff --git a/rust/crates/queue/Cargo.toml b/rust/crates/queue/Cargo.toml new file mode 100644 index 000000000..9a8e939da --- /dev/null +++ b/rust/crates/queue/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "storage-queue" +version = "0.1.0" +edition = "2024" + +[lib] +name = "storage_queue" + +[dependencies] +aws-sdk-sqs.workspace = true +thiserror.workspace = true +tokio.workspace = true diff --git a/rust/crates/queue/src/lib.rs b/rust/crates/queue/src/lib.rs new file mode 100644 index 000000000..3e0f6603d --- /dev/null +++ b/rust/crates/queue/src/lib.rs @@ -0,0 +1,129 @@ +use std::time::Duration; + +pub mod sqs; + +const MAX_RECEIVE: usize = 10; +const MAX_WAIT_SECS: u64 = 20; +const MAX_DELAY_SECS: u64 = 15 * 60; +const MAX_VISIBILITY_SECS: u64 = 12 * 60 * 60; + +#[derive(Debug, Clone)] +pub struct Outgoing { + pub body: String, + pub delay: Duration, + pub group_id: Option, +} + +#[derive(Debug, Clone)] +pub struct Received { + pub receipt: String, + pub body: String, + pub receive_count: u32, + pub group_id: Option, +} + +pub trait Transport { + /// Enqueue, chunking as needed. Batch errors report an outcome for each input. + /// Unknown outcomes may have succeeded; retrying them can produce duplicates. + /// The underlying client may also retry requests, so a failed entry in the final + /// response does not rule out success in an earlier attempt. + /// Durations must be whole seconds. + fn send(&self, messages: &[Outgoing]) -> impl Future> + Send; + /// Claim up to `max` messages (1..=10), long-polling up to `wait` (<= 20 s). Claimed messages are + /// invisible for `visibility` (0 s ..= 12 h) and have receive_count incremented. + /// Durations must be whole seconds. If any returned message is malformed, the + /// whole receive fails; claimed messages remain invisible until their timeout. + fn receive( + &self, + max: usize, + wait: Duration, + visibility: Duration, + ) -> impl Future, Error>> + Send; + /// Set the message's visibility timeout to `visibility` measured from now (0 = make visible now, i.e. nack). + /// Returns Error::StaleReceipt when the backend identifies the message as no longer in flight. + /// Invalid handles and other provider errors are returned separately; a provider may not + /// distinguish every stale handle from an invalid one. + /// The duration must be whole seconds. + fn extend( + &self, + receipt: &str, + visibility: Duration, + ) -> impl Future> + Send; + /// Delete (ack). SQS may report success for an old receipt without deleting the message; + /// rejected receipts are reported as failures. + /// Batch errors report an outcome for each input. + fn delete(&self, receipts: &[&str]) -> impl Future> + Send; +} + +/// The outcome of one input in a failed batch operation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum BatchOutcome { + Succeeded, + /// The final response rejected this entry. Earlier request attempts may have succeeded. + Failed { + code: String, + message: Option, + sender_fault: bool, + }, + /// The request may have been sent, but its result could not be determined. + Unknown, + /// No request containing this input was made. + NotAttempted, +} + +#[derive(Debug, thiserror::Error)] +#[error("{operation} batch failed")] +pub struct BatchError { + pub operation: &'static str, + /// Outcomes are ordered to match the original input slice. + pub outcomes: Vec, + #[source] + pub source: Option>, +} + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("invalid argument: {0}")] + InvalidArgument(&'static str), + #[error("stale receipt")] + StaleReceipt, + #[error("invalid SQS response: {0}")] + InvalidResponse(&'static str), + #[error(transparent)] + Batch(#[from] BatchError), + #[error(transparent)] + Sqs(Box), +} + +pub(crate) fn check_send(messages: &[Outgoing]) -> Result<(), Error> { + if messages.iter().any(|m| m.delay.subsec_nanos() != 0) { + return Err(Error::InvalidArgument("delay must be whole seconds")); + } + if messages.iter().any(|m| m.delay.as_secs() > MAX_DELAY_SECS) { + return Err(Error::InvalidArgument("delay exceeds 15 minutes")); + } + Ok(()) +} + +pub(crate) fn check_receive(max: usize, wait: Duration, visibility: Duration) -> Result<(), Error> { + if !(1..=MAX_RECEIVE).contains(&max) { + return Err(Error::InvalidArgument("max must be 1..=10")); + } + if wait.subsec_nanos() != 0 { + return Err(Error::InvalidArgument("wait must be whole seconds")); + } + if wait.as_secs() > MAX_WAIT_SECS { + return Err(Error::InvalidArgument("wait exceeds 20 seconds")); + } + check_visibility(visibility) +} + +pub(crate) fn check_visibility(visibility: Duration) -> Result<(), Error> { + if visibility.subsec_nanos() != 0 { + return Err(Error::InvalidArgument("visibility must be whole seconds")); + } + if visibility.as_secs() > MAX_VISIBILITY_SECS { + return Err(Error::InvalidArgument("visibility exceeds 12 hours")); + } + Ok(()) +} diff --git a/rust/crates/queue/src/sqs.rs b/rust/crates/queue/src/sqs.rs new file mode 100644 index 000000000..2be95af37 --- /dev/null +++ b/rust/crates/queue/src/sqs.rs @@ -0,0 +1,322 @@ +use std::time::Duration; + +use aws_sdk_sqs::{ + Client, + error::ProvideErrorMetadata, + types::{ + DeleteMessageBatchRequestEntry, Message, MessageSystemAttributeName, + SendMessageBatchRequestEntry, + }, +}; + +use crate::{ + BatchError, BatchOutcome, Error, Outgoing, Received, Transport, check_receive, check_send, + check_visibility, +}; + +const BATCH: usize = 10; +const MAX_BATCH_BYTES: usize = 1024 * 1024; + +pub struct SqsQueue { + client: Client, + queue_url: String, +} + +impl SqsQueue { + pub fn new(client: Client, queue_url: impl Into) -> Self { + Self { + client, + queue_url: queue_url.into(), + } + } +} + +fn sqs(e: E) -> Error { + Error::Sqs(Box::new(e)) +} + +fn secs(d: Duration) -> i32 { + d.as_secs() as i32 +} + +fn batch_error( + operation: &'static str, + outcomes: Vec, + source: Option>, +) -> Error { + BatchError { + operation, + outcomes, + source, + } + .into() +} + +fn record_outcome( + outcomes: &mut [BatchOutcome], + start: usize, + end: usize, + id: &str, + outcome: BatchOutcome, +) -> Result<(), Error> { + let index = id + .parse::() + .map_err(|_| Error::InvalidResponse("invalid batch entry id"))?; + let slot = outcomes + .get_mut(index) + .ok_or(Error::InvalidResponse("batch entry id out of range"))?; + if index < start || index >= end || !matches!(slot, BatchOutcome::Unknown) { + return Err(Error::InvalidResponse( + "unexpected or duplicate batch entry id", + )); + } + *slot = outcome; + Ok(()) +} + +fn received(m: &Message) -> Result { + let attrs = m.attributes(); + let receive_count = attrs + .and_then(|a| a.get(&MessageSystemAttributeName::ApproximateReceiveCount)) + .ok_or(Error::InvalidResponse("missing receive count"))? + .parse::() + .ok() + .filter(|count| *count > 0) + .ok_or(Error::InvalidResponse("invalid receive count"))?; + Ok(Received { + receipt: m + .receipt_handle() + .ok_or(Error::InvalidResponse("missing receipt handle"))? + .to_owned(), + body: m + .body() + .ok_or(Error::InvalidResponse("missing message body"))? + .to_owned(), + receive_count, + group_id: attrs + .and_then(|a| a.get(&MessageSystemAttributeName::MessageGroupId)) + .cloned(), + }) +} + +impl Transport for SqsQueue { + async fn send(&self, messages: &[Outgoing]) -> Result<(), Error> { + check_send(messages)?; + if messages.iter().any(|m| m.body.len() > MAX_BATCH_BYTES) { + return Err(Error::InvalidArgument("message body exceeds 1 MiB")); + } + let mut outcomes = vec![BatchOutcome::NotAttempted; messages.len()]; + let mut start = 0; + while start < messages.len() { + let mut end = start; + let mut bytes = 0; + while end < messages.len() && end - start < BATCH { + let size = messages[end].body.len(); + if size > MAX_BATCH_BYTES - bytes { + break; + } + bytes += size; + end += 1; + } + let chunk = &messages[start..end]; + let entries = chunk + .iter() + .enumerate() + .map(|(i, m)| { + let mut e = SendMessageBatchRequestEntry::builder() + .id((start + i).to_string()) + .message_body(&m.body) + .delay_seconds(secs(m.delay)); + if let Some(g) = &m.group_id { + e = e.message_group_id(g); + } + e.build() + }) + .collect::, _>>() + .map_err(|e| batch_error("send", outcomes.clone(), Some(Box::new(e))))?; + outcomes[start..end].fill(BatchOutcome::Unknown); + let out = self + .client + .send_message_batch() + .queue_url(&self.queue_url) + .set_entries(Some(entries)) + .send() + .await + .map_err(|e| batch_error("send", outcomes.clone(), Some(Box::new(e))))?; + let result = (|| { + for entry in out.successful() { + record_outcome( + &mut outcomes, + start, + end, + entry.id(), + BatchOutcome::Succeeded, + )?; + } + for entry in out.failed() { + record_outcome( + &mut outcomes, + start, + end, + entry.id(), + BatchOutcome::Failed { + code: entry.code().to_owned(), + message: entry.message().map(str::to_owned), + sender_fault: entry.sender_fault(), + }, + )?; + } + Ok::<_, Error>(()) + })(); + if let Err(e) = result { + outcomes[start..end].fill(BatchOutcome::Unknown); + return Err(batch_error("send", outcomes, Some(Box::new(e)))); + } + if outcomes[start..end] + .iter() + .any(|o| !matches!(o, BatchOutcome::Succeeded)) + { + return Err(batch_error("send", outcomes, None)); + } + start = end; + } + Ok(()) + } + + async fn receive( + &self, + max: usize, + wait: Duration, + visibility: Duration, + ) -> Result, Error> { + check_receive(max, wait, visibility)?; + let out = self + .client + .receive_message() + .queue_url(&self.queue_url) + .max_number_of_messages(max as i32) + .wait_time_seconds(secs(wait)) + .visibility_timeout(secs(visibility)) + .message_system_attribute_names(MessageSystemAttributeName::ApproximateReceiveCount) + .message_system_attribute_names(MessageSystemAttributeName::MessageGroupId) + .send() + .await + .map_err(sqs)?; + out.messages().iter().map(received).collect() + } + + async fn extend(&self, receipt: &str, visibility: Duration) -> Result<(), Error> { + check_visibility(visibility)?; + self.client + .change_message_visibility() + .queue_url(&self.queue_url) + .receipt_handle(receipt) + .visibility_timeout(secs(visibility)) + .send() + .await + .map_err(|e| { + let not_inflight = e.as_service_error().is_some_and(|e| e.is_message_not_inflight()) + || e.code() == Some("MessageNotInflight"); + // AWS has also returned this unmodeled error for a deleted or unavailable message. + let unavailable = e.code() == Some("InvalidParameterValue") + && e.message().is_some_and(|message| message.contains("for parameter ReceiptHandle is invalid") + && message.contains("Message does not exist or is not available for visibility timeout change")); + if not_inflight || unavailable { Error::StaleReceipt } else { sqs(e) } + })?; + Ok(()) + } + + async fn delete(&self, receipts: &[&str]) -> Result<(), Error> { + let mut outcomes = vec![BatchOutcome::NotAttempted; receipts.len()]; + for (chunk_index, chunk) in receipts.chunks(BATCH).enumerate() { + let start = chunk_index * BATCH; + let end = start + chunk.len(); + let entries = chunk + .iter() + .enumerate() + .map(|(i, r)| { + DeleteMessageBatchRequestEntry::builder() + .id((start + i).to_string()) + .receipt_handle(*r) + .build() + }) + .collect::, _>>() + .map_err(|e| batch_error("delete", outcomes.clone(), Some(Box::new(e))))?; + outcomes[start..end].fill(BatchOutcome::Unknown); + let out = self + .client + .delete_message_batch() + .queue_url(&self.queue_url) + .set_entries(Some(entries)) + .send() + .await + .map_err(|e| batch_error("delete", outcomes.clone(), Some(Box::new(e))))?; + let result = (|| { + for entry in out.successful() { + record_outcome( + &mut outcomes, + start, + end, + entry.id(), + BatchOutcome::Succeeded, + )?; + } + for entry in out.failed() { + let outcome = BatchOutcome::Failed { + code: entry.code().to_owned(), + message: entry.message().map(str::to_owned), + sender_fault: entry.sender_fault(), + }; + record_outcome(&mut outcomes, start, end, entry.id(), outcome)?; + } + Ok::<_, Error>(()) + })(); + if let Err(e) = result { + outcomes[start..end].fill(BatchOutcome::Unknown); + return Err(batch_error("delete", outcomes, Some(Box::new(e)))); + } + if outcomes[start..end] + .iter() + .any(|o| !matches!(o, BatchOutcome::Succeeded)) + { + return Err(batch_error("delete", outcomes, None)); + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn received_requires_message_fields_and_a_valid_count() { + let make = |receipt: Option<&str>, body: Option<&str>, count: Option<&str>| { + let mut builder = Message::builder() + .set_receipt_handle(receipt.map(str::to_owned)) + .set_body(body.map(str::to_owned)); + if let Some(count) = count { + builder = + builder.attributes(MessageSystemAttributeName::ApproximateReceiveCount, count); + } + builder.build() + }; + assert!(matches!( + received(&make(None, Some("body"), Some("1"))), + Err(Error::InvalidResponse("missing receipt handle")) + )); + assert!(matches!( + received(&make(Some("receipt"), None, Some("1"))), + Err(Error::InvalidResponse("missing message body")) + )); + for count in [None, Some("invalid"), Some("0")] { + assert!(matches!( + received(&make(Some("receipt"), Some("body"), count)), + Err(Error::InvalidResponse(_)) + )); + } + let result = received(&make(Some("receipt"), Some("body"), Some("3"))).unwrap(); + assert_eq!(result.receive_count, 3); + } +} diff --git a/rust/crates/queue/tests/contract.rs b/rust/crates/queue/tests/contract.rs new file mode 100644 index 000000000..024471f40 --- /dev/null +++ b/rust/crates/queue/tests/contract.rs @@ -0,0 +1,334 @@ +use std::{ + collections::HashSet, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; + +use aws_sdk_sqs::{ + Client, + config::{BehaviorVersion, Credentials, Region}, + types::QueueAttributeName, +}; +use storage_queue::{BatchOutcome, Error, Outgoing, Received, Transport, sqs::SqsQueue}; +use tokio::time::sleep; + +const SQS_ENDPOINT: &str = "STORAGE_QUEUE_TEST_SQS_ENDPOINT"; + +static COUNTER: AtomicUsize = AtomicUsize::new(0); + +fn unique(test: &str) -> String { + format!( + "{test}-{}-{}", + std::process::id(), + COUNTER.fetch_add(1, Ordering::Relaxed) + ) +} + +fn sqs_client(endpoint: &str) -> Client { + let config = aws_sdk_sqs::Config::builder() + .behavior_version(BehaviorVersion::latest()) + .endpoint_url(endpoint) + .region(Region::new("us-east-1")) + .credentials_provider(Credentials::new("x", "x", None, None, "test")) + .build(); + Client::from_conf(config) +} + +async fn create_queue(client: &Client, name: &str, redrive: Option<(&str, u32)>) -> String { + let mut req = client.create_queue().queue_name(name); + if let Some((dlq_url, max)) = redrive { + let arn = client + .get_queue_attributes() + .queue_url(dlq_url) + .attribute_names(QueueAttributeName::QueueArn) + .send() + .await + .unwrap() + .attributes() + .unwrap() + .get(&QueueAttributeName::QueueArn) + .unwrap() + .clone(); + req = req.attributes( + QueueAttributeName::RedrivePolicy, + format!(r#"{{"deadLetterTargetArn":"{arn}","maxReceiveCount":"{max}"}}"#), + ); + } + req.send().await.unwrap().queue_url.unwrap() +} + +async fn sqs_queue(test: &str) -> SqsQueue { + let client = sqs_client( + &std::env::var(SQS_ENDPOINT).expect("STORAGE_QUEUE_TEST_SQS_ENDPOINT must be set"), + ); + let url = create_queue(&client, &unique(test), None).await; + SqsQueue::new(client, url) +} + +async fn sqs_redrive(test: &str, max_receive_count: u32) -> (SqsQueue, SqsQueue) { + let client = sqs_client( + &std::env::var(SQS_ENDPOINT).expect("STORAGE_QUEUE_TEST_SQS_ENDPOINT must be set"), + ); + let dlq_url = create_queue(&client, &unique(&format!("{test}-dlq")), None).await; + let url = create_queue(&client, &unique(test), Some((&dlq_url, max_receive_count))).await; + ( + SqsQueue::new(client.clone(), url), + SqsQueue::new(client, dlq_url), + ) +} + +macro_rules! sqs_contract { + ($f:ident) => { + mod $f { + #[tokio::test] + #[ignore = "requires an SQS-compatible endpoint"] + async fn sqs() { + let q = super::sqs_queue(stringify!($f)).await; + super::$f(&q).await; + } + } + }; +} + +fn msg(body: &str) -> Outgoing { + Outgoing { + body: body.into(), + delay: Duration::ZERO, + group_id: None, + } +} + +const S1: Duration = Duration::from_secs(1); +const S2: Duration = Duration::from_secs(2); +const NOW: Duration = Duration::ZERO; + +async fn receive_one(q: &T, visibility: Duration) -> Received { + let mut got = q.receive(1, NOW, visibility).await.unwrap(); + assert_eq!(got.len(), 1); + got.remove(0) +} + +async fn assert_empty(q: &T) { + assert!(q.receive(10, NOW, S1).await.unwrap().is_empty()); +} + +async fn round_trip(q: &T) { + q.send(&[Outgoing { + body: "hello".into(), + delay: NOW, + group_id: Some("g1".into()), + }]) + .await + .unwrap(); + let got = receive_one(q, S2).await; + assert_eq!(got.body, "hello"); + assert_eq!(got.receive_count, 1); + assert_eq!(got.group_id.as_deref(), Some("g1")); +} +sqs_contract!(round_trip); + +async fn redelivery(q: &T) { + q.send(&[msg("a")]).await.unwrap(); + receive_one(q, S2).await; + assert_empty(q).await; + sleep(Duration::from_millis(2500)).await; + let got = receive_one(q, S2).await; + assert_eq!((got.body.as_str(), got.receive_count), ("a", 2)); +} +sqs_contract!(redelivery); + +async fn delete_acks(q: &T) { + q.send(&[msg("a")]).await.unwrap(); + let got = receive_one(q, S1).await; + q.delete(&[&got.receipt]).await.unwrap(); + sleep(Duration::from_millis(1500)).await; + assert_empty(q).await; +} +sqs_contract!(delete_acks); + +async fn extend_and_nack(q: &T) { + q.send(&[msg("a")]).await.unwrap(); + let got = receive_one(q, S1).await; + q.extend(&got.receipt, Duration::from_secs(4)) + .await + .unwrap(); + sleep(Duration::from_millis(1500)).await; + assert_empty(q).await; + q.extend(&got.receipt, NOW).await.unwrap(); + let again = receive_one(q, S2).await; + assert_eq!((again.body.as_str(), again.receive_count), ("a", 2)); +} +sqs_contract!(extend_and_nack); + +async fn stale_receipt(q: &T) { + q.send(&[msg("a")]).await.unwrap(); + let first = receive_one(q, S1).await; + sleep(Duration::from_millis(1500)).await; + let second = receive_one(q, Duration::from_secs(10)).await; + assert_eq!(second.receive_count, 2); + // Providers can reject a stale handle as invalid rather than as not-in-flight. + assert!(matches!( + q.extend(&first.receipt, S2).await, + Err(Error::StaleReceipt | Error::Sqs(_)) + )); + match q.delete(&[&first.receipt]).await { + Ok(()) => (), + Err(Error::Batch(e)) => assert!( + matches!(&e.outcomes[..], [BatchOutcome::Failed { code, .. }] if code == "ReceiptHandleIsInvalid") + ), + Err(e) => panic!("unexpected error: {e}"), + } + q.extend(&second.receipt, NOW).await.unwrap(); + let third = receive_one(q, S2).await; + assert_eq!(third.receive_count, 3); +} +sqs_contract!(stale_receipt); + +async fn delay(q: &T) { + q.send(&[Outgoing { + delay: S2, + ..msg("a") + }]) + .await + .unwrap(); + assert_empty(q).await; + sleep(Duration::from_millis(2500)).await; + assert_eq!(receive_one(q, S2).await.body, "a"); +} +sqs_contract!(delay); + +#[tokio::test] +#[ignore = "requires an SQS-compatible endpoint"] +async fn zero_delay_overrides_queue_default() { + let endpoint = + std::env::var(SQS_ENDPOINT).expect("STORAGE_QUEUE_TEST_SQS_ENDPOINT must be set"); + let client = sqs_client(&endpoint); + let url = client + .create_queue() + .queue_name(unique("default-delay")) + .attributes(QueueAttributeName::DelaySeconds, "10") + .send() + .await + .unwrap() + .queue_url + .unwrap(); + let q = SqsQueue::new(client, url); + q.send(&[msg("immediate")]).await.unwrap(); + let got = q.receive(1, S1, S2).await.unwrap(); + assert_eq!(got.len(), 1); + assert_eq!(got[0].body, "immediate"); +} + +async fn chunked_send(q: &T) { + let all: Vec<_> = (0..25).map(|i| msg(&format!("m{i}"))).collect(); + q.send(&all).await.unwrap(); + let mut seen = HashSet::new(); + for _ in 0..50 { + if seen.len() == 25 { + break; + } + for m in q.receive(10, S1, Duration::from_secs(30)).await.unwrap() { + assert!(seen.insert(m.body)); + } + } + assert_eq!(seen.len(), 25); +} +sqs_contract!(chunked_send); + +async fn concurrent_receivers(q: T) { + let q = Arc::new(q); + let all: Vec<_> = (0..20).map(|i| msg(&format!("m{i}"))).collect(); + q.send(&all).await.unwrap(); + let drain = |q: Arc| { + tokio::spawn(async move { + let mut got = Vec::new(); + loop { + let batch = q.receive(3, S1, Duration::from_secs(30)).await.unwrap(); + if batch.is_empty() { + return got; + } + got.extend(batch); + } + }) + }; + let (a, b) = tokio::join!(drain(q.clone()), drain(q.clone())); + let all: Vec = [a.unwrap(), b.unwrap()].concat(); + assert_eq!(all.len(), 20); + assert_eq!( + all.iter().map(|m| &m.body).collect::>().len(), + 20 + ); + assert_eq!( + all.iter().map(|m| &m.receipt).collect::>().len(), + 20 + ); +} + +#[tokio::test] +#[ignore = "requires an SQS-compatible endpoint"] +async fn concurrent_receivers_sqs() { + concurrent_receivers(sqs_queue("concurrent").await).await; +} + +async fn redrive(main: &T, dlq: &T) { + main.send(&[msg("a")]).await.unwrap(); + for n in 1..=2 { + assert_eq!(receive_one(main, S1).await.receive_count, n); + sleep(Duration::from_millis(1500)).await; + } + assert_empty(main).await; + let dead = receive_one(dlq, S2).await; + assert_eq!(dead.body, "a"); +} + +#[tokio::test] +#[ignore = "requires an SQS-compatible endpoint"] +async fn redrive_sqs() { + let (main, dlq) = sqs_redrive("redrive", 2).await; + redrive(&main, &dlq).await; +} + +async fn validation(q: &T) { + let bad = |r: Result<_, Error>| assert!(matches!(r, Err(Error::InvalidArgument(_)))); + bad(q.receive(0, NOW, S1).await.map(drop)); + bad(q.receive(11, NOW, S1).await.map(drop)); + bad(q.receive(1, Duration::from_secs(21), S1).await.map(drop)); + bad(q.receive(1, Duration::from_millis(500), S1).await.map(drop)); + bad(q + .receive(1, NOW, Duration::from_millis(500)) + .await + .map(drop)); + bad(q + .receive(1, NOW, Duration::from_secs(13 * 3600)) + .await + .map(drop)); + bad(q.extend("r", Duration::from_secs(13 * 3600)).await); + bad(q.extend("r", Duration::from_millis(500)).await); + bad(q + .send(&[Outgoing { + delay: Duration::from_millis(500), + ..msg("a") + }]) + .await); + bad(q + .send(&[Outgoing { + delay: Duration::from_millis(900_001), + ..msg("a") + }]) + .await); + bad(q + .send(&[Outgoing { + delay: Duration::from_secs(16 * 60), + ..msg("a") + }]) + .await); +} + +#[tokio::test] +async fn validation_sqs() { + let q = SqsQueue::new(sqs_client("http://127.0.0.1:1"), "http://127.0.0.1:1/q"); + validation(&q).await; +} diff --git a/rust/crates/queue/tests/failure_paths.rs b/rust/crates/queue/tests/failure_paths.rs new file mode 100644 index 000000000..7763a3fcc --- /dev/null +++ b/rust/crates/queue/tests/failure_paths.rs @@ -0,0 +1,317 @@ +use std::{ + io::{Read, Write}, + net::TcpListener, + thread, + time::Duration, +}; + +use aws_sdk_sqs::{ + Client, + config::{BehaviorVersion, Credentials, Region, retry::RetryConfig}, +}; +use storage_queue::{BatchOutcome, Error, Outgoing, Transport, sqs::SqsQueue}; + +// A small HTTP endpoint exercises the SDK and transport without depending on emulator failure injection. +fn endpoint(responses: Vec<(u16, String)>) -> (SqsQueue, thread::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let handle = thread::spawn(move || { + let mut requests = Vec::new(); + for (status, body) in responses { + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut bytes = Vec::new(); + let mut buffer = [0; 4096]; + loop { + let n = stream.read(&mut buffer).unwrap(); + assert!(n > 0); + bytes.extend_from_slice(&buffer[..n]); + if let Some(pos) = bytes.windows(4).position(|w| w == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&bytes[..pos]); + let length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().unwrap()) + }) + .unwrap(); + if bytes.len() >= pos + 4 + length { + requests.push( + String::from_utf8(bytes[pos + 4..pos + 4 + length].to_vec()).unwrap(), + ); + break; + } + } + } + write!(stream, "HTTP/1.1 {status} Test\r\nContent-Type: application/x-amz-json-1.0\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap(); + } + requests + }); + let config = aws_sdk_sqs::Config::builder() + .behavior_version(BehaviorVersion::latest()) + .endpoint_url(&url) + .region(Region::new("us-east-1")) + .credentials_provider(Credentials::new("x", "x", None, None, "test")) + .retry_config(RetryConfig::standard().with_max_attempts(1)) + .build(); + (SqsQueue::new(Client::from_conf(config), &url), handle) +} + +fn successful(start: usize, end: usize) -> String { + (start..end) + .map(|i| format!(r#"{{"Id":"{i}","MessageId":"m{i}","MD5OfMessageBody":"x"}}"#)) + .collect::>() + .join(",") +} + +fn messages(n: usize) -> Vec { + (0..n) + .map(|_| Outgoing { + body: "body".into(), + delay: Duration::ZERO, + group_id: None, + }) + .collect() +} + +#[tokio::test] +async fn send_reports_successes_failures_and_unsent_entries_across_chunks() { + let first = format!(r#"{{"Successful":[{}],"Failed":[]}}"#, successful(0, 10)); + let second = format!( + r#"{{"Successful":[{}],"Failed":[{{"Id":"10","Code":"bad","Message":"rejected","SenderFault":true}}]}}"#, + successful(11, 20) + ); + let (queue, server) = endpoint(vec![(200, first), (200, second)]); + let Error::Batch(error) = queue.send(&messages(21)).await.unwrap_err() else { + panic!("expected batch error") + }; + server.join().unwrap(); + assert_eq!(error.operation, "send"); + assert!( + error.outcomes[..10] + .iter() + .all(|o| *o == BatchOutcome::Succeeded) + ); + assert_eq!( + error.outcomes[10], + BatchOutcome::Failed { + code: "bad".into(), + message: Some("rejected".into()), + sender_fault: true + } + ); + assert!( + error.outcomes[11..20] + .iter() + .all(|o| *o == BatchOutcome::Succeeded) + ); + assert_eq!(error.outcomes[20], BatchOutcome::NotAttempted); +} + +#[tokio::test] +async fn request_error_keeps_current_chunk_unknown_and_later_chunks_unsent() { + let first = format!(r#"{{"Successful":[{}],"Failed":[]}}"#, successful(0, 10)); + let (queue, server) = endpoint(vec![ + (200, first), + ( + 500, + r#"{"__type":"InternalError","message":"failed"}"#.into(), + ), + ]); + let Error::Batch(error) = queue.send(&messages(21)).await.unwrap_err() else { + panic!("expected batch error") + }; + server.join().unwrap(); + assert!(error.source.is_some()); + assert!( + error.outcomes[..10] + .iter() + .all(|o| *o == BatchOutcome::Succeeded) + ); + assert!( + error.outcomes[10..20] + .iter() + .all(|o| *o == BatchOutcome::Unknown) + ); + assert_eq!(error.outcomes[20], BatchOutcome::NotAttempted); +} + +#[tokio::test] +async fn delete_reports_invalid_receipts_and_other_rejections() { + let body = r#"{"Successful":[{"Id":"0"}],"Failed":[{"Id":"1","Code":"ReceiptHandleIsInvalid","SenderFault":true},{"Id":"2","Code":"Throttled","SenderFault":false}]}"#; + let (queue, server) = endpoint(vec![(200, body.into())]); + let Error::Batch(error) = queue.delete(&["a", "b", "c"]).await.unwrap_err() else { + panic!("expected batch error") + }; + server.join().unwrap(); + assert_eq!(error.operation, "delete"); + assert_eq!( + error.outcomes, + vec![ + BatchOutcome::Succeeded, + BatchOutcome::Failed { + code: "ReceiptHandleIsInvalid".into(), + message: None, + sender_fault: true + }, + BatchOutcome::Failed { + code: "Throttled".into(), + message: None, + sender_fault: false + } + ] + ); +} + +#[tokio::test] +async fn send_splits_on_utf8_bytes_and_explicitly_sends_zero_delay() { + let responses = vec![ + ( + 200, + format!(r#"{{"Successful":[{}],"Failed":[]}}"#, successful(0, 2)), + ), + ( + 200, + format!(r#"{{"Successful":[{}],"Failed":[]}}"#, successful(2, 3)), + ), + ]; + let (queue, server) = endpoint(responses); + let input = vec![ + Outgoing { + body: "é".repeat(262_144), + delay: Duration::ZERO, + group_id: None, + }, + Outgoing { + body: "x".repeat(524_288), + delay: Duration::from_secs(3), + group_id: None, + }, + Outgoing { + body: "z".into(), + delay: Duration::ZERO, + group_id: None, + }, + ]; + queue.send(&input).await.unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests.len(), 2); + assert!(requests[0].contains(r#""Id":"0""#)); + assert!(requests[0].contains(r#""Id":"1""#)); + assert!(!requests[0].contains(r#""Id":"2""#)); + assert!(requests[0].contains(r#""DelaySeconds":0"#)); + assert!(requests[0].contains(r#""DelaySeconds":3"#)); + assert!(requests[1].contains(r#""Id":"2""#)); + assert!(requests[1].contains(r#""DelaySeconds":0"#)); +} + +#[tokio::test] +async fn oversized_message_is_rejected_before_any_send() { + let (queue, server) = endpoint(vec![]); + let mut input = messages(2); + input[1].body = "x".repeat(1_048_577); + assert!(matches!( + queue.send(&input).await, + Err(Error::InvalidArgument("message body exceeds 1 MiB")) + )); + assert!(server.join().unwrap().is_empty()); +} + +#[tokio::test] +async fn byte_sized_chunks_preserve_original_indices_on_partial_failure() { + let first = format!(r#"{{"Successful":[{}],"Failed":[]}}"#, successful(0, 2)); + let second = r#"{"Successful":[],"Failed":[{"Id":"2","Code":"InvalidMessageContents","SenderFault":true}]}"#; + let (queue, server) = endpoint(vec![(200, first), (200, second.into())]); + let mut input = messages(4); + input[0].body = "x".repeat(524_288); + input[1].body = "x".repeat(524_288); + input[2].body = "x".repeat(1_048_576); + let Error::Batch(error) = queue.send(&input).await.unwrap_err() else { + panic!("expected batch error") + }; + server.join().unwrap(); + assert_eq!( + error.outcomes, + vec![ + BatchOutcome::Succeeded, + BatchOutcome::Succeeded, + BatchOutcome::Failed { + code: "InvalidMessageContents".into(), + message: None, + sender_fault: true + }, + BatchOutcome::NotAttempted, + ] + ); +} + +#[tokio::test] +async fn extend_distinguishes_unavailable_messages_from_invalid_handles() { + let cases = [ + ( + "AWS.SimpleQueueService.MessageNotInflight", + "not in flight", + true, + ), + ("MessageNotInflight", "not in flight", true), + ( + "InvalidParameterValue", + "Value example for parameter ReceiptHandle is invalid. Reason: Message does not exist or is not available for visibility timeout change.", + true, + ), + ("ReceiptHandleIsInvalid", "invalid receipt", false), + ("InvalidParameterValue", "invalid visibility timeout", false), + ]; + for (code, message, stale) in cases { + let response = format!(r#"{{"__type":"{code}","message":"{message}"}}"#); + let (queue, server) = endpoint(vec![(400, response)]); + let result = queue.extend("receipt", Duration::from_secs(1)).await; + server.join().unwrap(); + if stale { + assert!( + matches!(result, Err(Error::StaleReceipt)), + "{code}: {result:?}" + ); + } else { + assert!(matches!(result, Err(Error::Sqs(_))), "{code}: {result:?}"); + } + } +} + +#[tokio::test] +async fn incomplete_batch_response_reports_unknown_outcome() { + let (queue, server) = endpoint(vec![( + 200, + format!(r#"{{"Successful":[{}],"Failed":[]}}"#, successful(0, 1)), + )]); + let Error::Batch(error) = queue.send(&messages(2)).await.unwrap_err() else { + panic!("expected batch error") + }; + server.join().unwrap(); + assert_eq!( + error.outcomes, + vec![BatchOutcome::Succeeded, BatchOutcome::Unknown] + ); +} + +#[tokio::test] +async fn duplicate_batch_ids_leave_the_chunk_unknown() { + let response = format!( + r#"{{"Successful":[{},{}],"Failed":[]}}"#, + successful(0, 1), + successful(0, 1) + ); + let (queue, server) = endpoint(vec![(200, response)]); + let Error::Batch(error) = queue.send(&messages(2)).await.unwrap_err() else { + panic!("expected batch error") + }; + server.join().unwrap(); + assert!(error.source.is_some()); + assert_eq!( + error.outcomes, + vec![BatchOutcome::Unknown, BatchOutcome::Unknown] + ); +} From 185c8469ceab0a5d82f44ebc6687adcb1d575ec3 Mon Sep 17 00:00:00 2001 From: Ferhat Elmas Date: Mon, 9 Nov 2026 16:59:47 +0100 Subject: [PATCH 2/2] fix: expose visibility change errors and document limit Signed-off-by: Ferhat Elmas --- rust/crates/queue/src/lib.rs | 13 ++++++++- rust/crates/queue/src/sqs.rs | 10 ++++++- rust/crates/queue/tests/contract.rs | 2 +- rust/crates/queue/tests/failure_paths.rs | 35 +++++++++++++++++++++++- 4 files changed, 56 insertions(+), 4 deletions(-) diff --git a/rust/crates/queue/src/lib.rs b/rust/crates/queue/src/lib.rs index 3e0f6603d..fbff50dbe 100644 --- a/rust/crates/queue/src/lib.rs +++ b/rust/crates/queue/src/lib.rs @@ -33,6 +33,7 @@ pub trait Transport { /// invisible for `visibility` (0 s ..= 12 h) and have receive_count incremented. /// Durations must be whole seconds. If any returned message is malformed, the /// whole receive fails; claimed messages remain invisible until their timeout. + /// SQS limits visibility to 12 hours from receiving the original receive request. fn receive( &self, max: usize, @@ -43,7 +44,10 @@ pub trait Transport { /// Returns Error::StaleReceipt when the backend identifies the message as no longer in flight. /// Invalid handles and other provider errors are returned separately; a provider may not /// distinguish every stale handle from an invalid one. - /// The duration must be whole seconds. + /// The duration must be whole seconds and at most 12 hours. SQS also rejects a duration + /// beyond the time remaining in the 12-hour window from the original receive request; + /// the remaining window is determined by SQS, not this client. Such errors are returned + /// as Error::VisibilityChange with the provider details; the timeout is not clamped. fn extend( &self, receipt: &str, @@ -89,6 +93,13 @@ pub enum Error { StaleReceipt, #[error("invalid SQS response: {0}")] InvalidResponse(&'static str), + #[error("visibility change failed (code: {code:?}, message: {message:?})")] + VisibilityChange { + code: Option, + message: Option, + #[source] + source: Box, + }, #[error(transparent)] Batch(#[from] BatchError), #[error(transparent)] diff --git a/rust/crates/queue/src/sqs.rs b/rust/crates/queue/src/sqs.rs index 2be95af37..26c5a9901 100644 --- a/rust/crates/queue/src/sqs.rs +++ b/rust/crates/queue/src/sqs.rs @@ -221,7 +221,15 @@ impl Transport for SqsQueue { let unavailable = e.code() == Some("InvalidParameterValue") && e.message().is_some_and(|message| message.contains("for parameter ReceiptHandle is invalid") && message.contains("Message does not exist or is not available for visibility timeout change")); - if not_inflight || unavailable { Error::StaleReceipt } else { sqs(e) } + if not_inflight || unavailable { + Error::StaleReceipt + } else { + Error::VisibilityChange { + code: e.code().map(str::to_owned), + message: e.message().map(str::to_owned), + source: Box::new(e), + } + } })?; Ok(()) } diff --git a/rust/crates/queue/tests/contract.rs b/rust/crates/queue/tests/contract.rs index 024471f40..aacd5efab 100644 --- a/rust/crates/queue/tests/contract.rs +++ b/rust/crates/queue/tests/contract.rs @@ -172,7 +172,7 @@ async fn stale_receipt(q: &T) { // Providers can reject a stale handle as invalid rather than as not-in-flight. assert!(matches!( q.extend(&first.receipt, S2).await, - Err(Error::StaleReceipt | Error::Sqs(_)) + Err(Error::StaleReceipt | Error::VisibilityChange { .. }) )); match q.delete(&[&first.receipt]).await { Ok(()) => (), diff --git a/rust/crates/queue/tests/failure_paths.rs b/rust/crates/queue/tests/failure_paths.rs index 7763a3fcc..6356f897f 100644 --- a/rust/crates/queue/tests/failure_paths.rs +++ b/rust/crates/queue/tests/failure_paths.rs @@ -276,11 +276,44 @@ async fn extend_distinguishes_unavailable_messages_from_invalid_handles() { "{code}: {result:?}" ); } else { - assert!(matches!(result, Err(Error::Sqs(_))), "{code}: {result:?}"); + assert!( + matches!(result, Err(Error::VisibilityChange { .. })), + "{code}: {result:?}" + ); } } } +#[tokio::test] +async fn extend_preserves_provider_error_details() { + use std::error::Error as _; + + let body = r#"{"__type":"InvalidParameterValue","message":"requested visibility exceeds remaining time"}"#; + let (queue, server) = endpoint(vec![(400, body.into())]); + let error = queue + .extend("receipt", Duration::from_secs(3600)) + .await + .unwrap_err(); + server.join().unwrap(); + assert!(error.source().is_some()); + assert!( + matches!(error, Error::VisibilityChange { code: Some(code), message: Some(message), .. } + if code == "InvalidParameterValue" && message == "requested visibility exceeds remaining time") + ); +} + +#[tokio::test] +async fn extend_forwards_valid_visibility_without_clamping() { + let (queue, server) = endpoint(vec![(200, "{}".into())]); + queue + .extend("receipt", Duration::from_secs(43_195)) + .await + .unwrap(); + let requests = server.join().unwrap(); + assert_eq!(requests.len(), 1); + assert!(requests[0].contains(r#""VisibilityTimeout":43195"#)); +} + #[tokio::test] async fn incomplete_batch_response_reports_unknown_outcome() { let (queue, server) = endpoint(vec![(