diff --git a/.github/workflows/backend-test.yml b/.github/workflows/backend-test.yml index fea84e7781..b0b2ebb77a 100644 --- a/.github/workflows/backend-test.yml +++ b/.github/workflows/backend-test.yml @@ -43,6 +43,18 @@ jobs: options: >- --health-cmd "mysqladmin ping -h localhost" --health-interval 10s --health-timeout 5s --health-retries 5 + # SQS for the sqs_trigger e2e test. The AWS SDK crates are bumped by Dependabot + # and the trigger's whole receive/ack path lives behind them, so it needs a broker + # in CI rather than a manual LocalStack someone remembers to start. + localstack: + image: localstack/localstack:3.8 + ports: + - 4566:4566 + env: + SERVICES: sqs + options: >- + --health-cmd "curl -sf http://localhost:4566/_localstack/health" + --health-interval 10s --health-timeout 5s --health-retries 10 steps: - uses: actions/checkout@v4 - uses: actions/setup-dotnet@v4 @@ -236,6 +248,13 @@ jobs: key: ${{ runner.os }}-duckdb-ffi-${{ hashFiles('./backend/windmill-duckdb-ffi-internal/src/**/*.rs', './backend/windmill-duckdb-ffi-internal/Cargo.toml', './backend/windmill-duckdb-ffi-internal/Cargo.lock') }} restore-keys: | ${{ runner.os }}-duckdb-ffi- + - name: Create the SQS test queue + run: | + aws --endpoint-url http://localhost:4566 --region us-east-1 \ + sqs create-queue --queue-name windmill-e2e-test + env: + AWS_ACCESS_KEY_ID: test + AWS_SECRET_ACCESS_KEY: test - name: cargo test timeout-minutes: 30 env: @@ -255,6 +274,11 @@ jobs: RUST_LOG: "off" RUST_LOG_STYLE: never CARGO_NET_GIT_FETCH_WITH_CLI: true + # The SQS trigger builds its client through aws_config, which honours this. + AWS_ENDPOINT_URL: http://localhost:4566 + AWS_ACCESS_KEY_ID: test + AWS_SECRET_ACCESS_KEY: test + AWS_REGION: us-east-1 # Cap parallel rustc/link jobs below the 16 available cores. The tail of # the build links ~128 full-graph test binaries (one per tests/*.rs file # across the workspace); at high parallelism enough heavy codegen+link @@ -295,4 +319,4 @@ jobs: # never reaches it. Pin the target dir (matching the cache step above) so # its own tests run off this compile rather than a second bundled build. (cd windmill-duckdb-ffi-internal && export CARGO_TARGET_DIR="$PWD/target" && ./build_dev.sh && cargo test --release -p windmill_duckdb_ffi_internal) - DENO_PATH=$(which deno) BUN_PATH=$(which bun) NODE_BIN_PATH=$(which node) GO_PATH=$(which go) UV_PATH=$(which uv) PHP_PATH=$(which php) COMPOSER_PATH=$(which composer) RUBY_PATH=$(which ruby) RUBY_BUNDLE_PATH=$(which bundle) RUBY_GEM_PATH=$(which gem) POWERSHELL_PATH=$(which pwsh) DOTNET_PATH=$(which dotnet) cargo test --features enterprise,deno_core,duckdb,license,python,rust,scoped_cache,parquet,private,private_registry_test,csharp,php,ruby,mysql,quickjs,mcp,run_inline --all -- --nocapture --test-threads=10 + DENO_PATH=$(which deno) BUN_PATH=$(which bun) NODE_BIN_PATH=$(which node) GO_PATH=$(which go) UV_PATH=$(which uv) PHP_PATH=$(which php) COMPOSER_PATH=$(which composer) RUBY_PATH=$(which ruby) RUBY_BUNDLE_PATH=$(which bundle) RUBY_GEM_PATH=$(which gem) POWERSHELL_PATH=$(which pwsh) DOTNET_PATH=$(which dotnet) cargo test --features enterprise,deno_core,duckdb,license,python,rust,scoped_cache,parquet,private,private_registry_test,csharp,php,ruby,mysql,quickjs,mcp,run_inline,sqs_trigger --all -- --nocapture --test-threads=10 diff --git a/backend/windmill-api-integration-tests/Cargo.toml b/backend/windmill-api-integration-tests/Cargo.toml index 3aee7e766c..1183901953 100644 --- a/backend/windmill-api-integration-tests/Cargo.toml +++ b/backend/windmill-api-integration-tests/Cargo.toml @@ -16,6 +16,7 @@ deno_core = ["windmill-test-utils/deno_core"] mcp = ["windmill-test-utils/mcp", "dep:rmcp"] run_inline = ["dep:windmill-worker", "windmill-test-utils/run_inline", "windmill-test-utils/duckdb"] postgres_trigger = ["windmill-test-utils/postgres_trigger"] +sqs_trigger = ["windmill-test-utils/sqs_trigger"] [dependencies] windmill-test-utils.workspace = true diff --git a/backend/windmill-api-integration-tests/tests/trigger_e2e.rs b/backend/windmill-api-integration-tests/tests/trigger_e2e.rs index a6b91a66e6..75dd53b346 100644 --- a/backend/windmill-api-integration-tests/tests/trigger_e2e.rs +++ b/backend/windmill-api-integration-tests/tests/trigger_e2e.rs @@ -710,21 +710,21 @@ async fn test_nats_e2e(db: Pool) -> anyhow::Result<()> { /// End-to-end test for SQS trigger (Enterprise only). /// -/// Requires LocalStack with the test queue. Setup: +/// CI provides LocalStack and the queue (see `backend-test.yml`). To run it locally: /// ```bash -/// ./tests/fixtures/start_sqs.sh -/// ``` -/// -/// Run: -/// ```bash -/// AWS_ENDPOINT_URL=http://localhost:4566 \ +/// docker run -d --rm -p 4566:4566 -e SERVICES=sqs localstack/localstack:3.8 +/// aws --endpoint-url http://localhost:4566 --region us-east-1 \ +/// sqs create-queue --queue-name windmill-e2e-test +/// AWS_ENDPOINT_URL=http://localhost:4566 AWS_ACCESS_KEY_ID=test \ +/// AWS_SECRET_ACCESS_KEY=test AWS_REGION=us-east-1 \ /// cargo test --test trigger_e2e test_sqs_e2e \ -/// --features sqs_trigger,enterprise,private -- --ignored --nocapture +/// --features sqs_trigger,enterprise,private -- --nocapture /// ``` #[cfg(all(feature = "enterprise", feature = "private"))] -#[ignore = "requires LocalStack SQS on localhost:4566"] #[sqlx::test(migrations = "../migrations", fixtures("base"))] async fn test_sqs_e2e(db: Pool) -> anyhow::Result<()> { + use aws_sdk_sqs::types::QueueAttributeName; + initialize_tracing().await; // The SQS listener uses aws_config which respects AWS_ENDPOINT_URL for LocalStack. @@ -779,15 +779,55 @@ async fn test_sqs_e2e(db: Pool) -> anyhow::Result<()> { .await; let sqs_client = aws_sdk_sqs::Client::new(&config); + let queue_url = "http://localhost:4566/000000000000/windmill-e2e-test"; sqs_client .send_message() - .queue_url("http://localhost:4566/000000000000/windmill-e2e-test") + .queue_url(queue_url) .message_body("hello from sqs e2e test") .send() .await?; let job = poll_for_trigger_job(&db, script_path, "sqs", Duration::from_secs(30)).await?; - assert!(job.args.is_some(), "job should have args"); + // The handler script has no preprocessor, so the args are exactly the v1 payload. + assert_eq!( + job.args.as_deref(), + Some(&json!({ "msg": "hello from sqs e2e test" })), + "the message body must reach the job unchanged" + ); + + // The ack half: a received message must be deleted, or every message is replayed + // forever once its visibility timeout lapses. Poll because the delete is issued + // after the job is pushed. + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + let attrs = sqs_client + .get_queue_attributes() + .queue_url(queue_url) + .attribute_names(QueueAttributeName::ApproximateNumberOfMessages) + .attribute_names(QueueAttributeName::ApproximateNumberOfMessagesNotVisible) + .send() + .await?; + let count = |name| { + attrs + .attributes() + .and_then(|a| a.get(&name)) + .map(String::as_str) + .unwrap_or("0") + .parse::() + .unwrap_or(0) + }; + let visible = count(QueueAttributeName::ApproximateNumberOfMessages); + let in_flight = count(QueueAttributeName::ApproximateNumberOfMessagesNotVisible); + if visible == 0 && in_flight == 0 { + break; + } + if tokio::time::Instant::now() >= deadline { + panic!( + "message was not deleted from the queue: {visible} visible, {in_flight} in flight" + ); + } + tokio::time::sleep(Duration::from_millis(500)).await; + } Ok(()) } diff --git a/backend/windmill-test-utils/Cargo.toml b/backend/windmill-test-utils/Cargo.toml index 7f025800bc..16e14b09cc 100644 --- a/backend/windmill-test-utils/Cargo.toml +++ b/backend/windmill-test-utils/Cargo.toml @@ -12,6 +12,10 @@ path = "src/lib.rs" default = [] private = ["windmill-api/private"] enterprise = ["windmill-api/enterprise"] +# `run_server` only spawns the SQS listener when windmill-api carries this feature, +# so an e2e test of the trigger has to ask for it rather than rely on another +# workspace member happening to enable it. +sqs_trigger = ["windmill-api/sqs_trigger"] python = ["windmill-common/python"] deno_core = ["dep:windmill-runtime-nativets"] mcp = ["windmill-api/mcp"]