From 52ef30f16715fabe2cc4e27b2f6ff0e36256f3cc Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 15 Jul 2026 13:01:20 +0800 Subject: [PATCH] test(e2e): add S3 event-notification webhook delivery regression net (#4821) Cover the previously untested configure-target -> put-notification-config -> object operation -> webhook delivery chain (backlog#1154 peri-1): PUT/multipart-complete/DELETE event fields, prefix/suffix filter negatives, and store-queue redelivery after target recovery. Wire both tests into the e2e-smoke nextest profile (ci-4 mechanism). --- .config/nextest.toml | 2 +- crates/e2e_test/src/lib.rs | 6 + .../e2e_test/src/notification_webhook_test.rs | 576 ++++++++++++++++++ 3 files changed, 583 insertions(+), 1 deletion(-) create mode 100644 crates/e2e_test/src/notification_webhook_test.rs diff --git a/.config/nextest.toml b/.config/nextest.toml index 6cb716248..027d52e21 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -165,7 +165,7 @@ test-group = 'e2e-reliability' [profile.e2e-smoke] default-filter = """ package(e2e_test) & ( - test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_source_invalid_date|content_encoding|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth)_test::/) + test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_source_invalid_date|content_encoding|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook)_test::/) | test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/) | test(/^reliant::lifecycle::/) ) diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index dbb6b6800..b745cfb62 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -191,6 +191,12 @@ mod stale_multipart_cleanup_cluster_test; #[cfg(test)] mod object_lambda_test; +// S3 event-notification webhook delivery end-to-end (backlog#1154 peri-1): +// configure webhook target -> PutBucketNotificationConfiguration -> object +// operation -> event delivered, plus filter negatives and store-queue redelivery. +#[cfg(test)] +mod notification_webhook_test; + // Replication extension end-to-end regression tests #[cfg(test)] mod replication_extension_test; diff --git a/crates/e2e_test/src/notification_webhook_test.rs b/crates/e2e_test/src/notification_webhook_test.rs new file mode 100644 index 000000000..dc55b111b --- /dev/null +++ b/crates/e2e_test/src/notification_webhook_test.rs @@ -0,0 +1,576 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! End-to-end regression net for the S3 event-notification pipeline +//! (backlog#1154 peri-1). Proves the full "configure webhook target -> +//! PutBucketNotificationConfiguration -> object operation -> event delivered" +//! chain against a real rustfs binary and a real HTTP receiver, which no other +//! test covers: the target-plugin suites stop at broker integration and the +//! object-lambda suite only exercises the webhook target as a transform +//! function, never as an S3 event sink. +//! +//! Coverage: +//! * PUT / multipart-complete / DELETE each deliver one event with the correct +//! eventName, bucket, key, versionId and eTag. +//! * prefix/suffix filters drop non-matching keys (rule-engine gate). +//! * an event queued while the target endpoint is unreachable is redelivered +//! from the on-disk store once the endpoint recovers (store-and-forward). + +use crate::common::{RustFSTestEnvironment, init_logging}; +use aws_sdk_s3::Client; +use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::types::{ + BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, Event, FilterRule, FilterRuleName, + NotificationConfiguration, NotificationConfigurationFilter, QueueConfiguration, S3KeyFilter, VersioningConfiguration, +}; +use http::header::{CONTENT_TYPE, HOST}; +use local_ip_address::local_ip; +use reqwest::StatusCode; +use rustfs_signer::constants::UNSIGNED_PAYLOAD; +use rustfs_signer::sign_v4; +use s3s::Body; +use serde_json::Value; +use serial_test::serial; +use std::error::Error; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tokio::time::{Duration, Instant, timeout}; + +type TestResult = Result<(), Box>; +type BoxError = Box; + +/// Region embedded in the target ARN. Only the `id:name` tail of the ARN is used +/// for target lookup (see `process_queue_configurations`), so the region is +/// nominal, but it must be present for `ARN::parse` to succeed. +const NOTIFY_REGION: &str = "us-east-1"; + +/// Webhook targets are registered as `TargetID { id: , name: "webhook" }`, +/// so the ARN a notification rule references is +/// `arn:rustfs:sqs:::webhook`. +fn target_arn(target_name: &str) -> String { + format!("arn:rustfs:sqs:{NOTIFY_REGION}:{target_name}:webhook") +} + +// --------------------------------------------------------------------------- +// In-test HTTP event receiver +// --------------------------------------------------------------------------- + +/// Reads one HTTP request off the stream, returning its method and body. Handles +/// the target's HEAD reachability probe (no body) and POST event deliveries. +async fn read_http_message(stream: &mut tokio::net::TcpStream) -> Result<(String, Vec), BoxError> { + let mut buffer = Vec::new(); + let mut chunk = [0_u8; 4096]; + + let header_end = loop { + let read = stream.read(&mut chunk).await?; + if read == 0 { + return Err("connection closed before request headers were complete".into()); + } + buffer.extend_from_slice(&chunk[..read]); + if let Some(pos) = buffer.windows(4).position(|w| w == b"\r\n\r\n") { + break pos; + } + }; + + let header_text = std::str::from_utf8(&buffer[..header_end])?; + let mut lines = header_text.split("\r\n"); + let request_line = lines.next().ok_or("missing request line")?; + let method = request_line.split_whitespace().next().ok_or("missing method")?.to_string(); + + let mut content_length = 0usize; + for line in lines { + if let Some((name, value)) = line.split_once(':') + && name.trim().eq_ignore_ascii_case("content-length") + { + content_length = value.trim().parse::().unwrap_or(0); + } + } + + let body_offset = header_end + 4; + while buffer.len().saturating_sub(body_offset) < content_length { + let read = stream.read(&mut chunk).await?; + if read == 0 { + return Err("connection closed before request body was complete".into()); + } + buffer.extend_from_slice(&chunk[..read]); + } + + Ok((method, buffer[body_offset..body_offset + content_length].to_vec())) +} + +/// Drives an accepted listener: answers every request 200 OK (so the target's +/// HEAD reachability probe reports it online) and forwards each non-empty POST +/// body, parsed as the S3 event envelope JSON, to `tx`. +fn serve_event_collector(listener: TcpListener, tx: mpsc::UnboundedSender) -> JoinHandle<()> { + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + let tx = tx.clone(); + tokio::spawn(async move { + if let Ok(Ok((method, body))) = timeout(Duration::from_secs(5), read_http_message(&mut stream)).await { + let _ = stream + .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n") + .await; + let _ = stream.shutdown().await; + if method == "POST" + && !body.is_empty() + && let Ok(value) = serde_json::from_slice::(&body) + { + let _ = tx.send(value); + } + } + }); + } + }) +} + +/// Binds a fresh collector on a random port. Returns its `/events` +/// endpoint URL, a receiver of parsed event envelopes, and the serving task. +async fn spawn_event_collector() -> Result<(String, mpsc::UnboundedReceiver, JoinHandle<()>), BoxError> { + let listener = TcpListener::bind("0.0.0.0:0").await?; + let port = listener.local_addr()?.port(); + let endpoint_ip = local_ip()?; + let (tx, rx) = mpsc::unbounded_channel(); + let handle = serve_event_collector(listener, tx); + Ok((format!("http://{endpoint_ip}.nip.io:{port}/events"), rx, handle)) +} + +/// Decoded object key of the first record in an event envelope. +fn event_key(envelope: &Value) -> Option { + let raw = envelope["Records"][0]["s3"]["object"]["key"].as_str()?; + Some( + urlencoding::decode(raw) + .map(|d| d.into_owned()) + .unwrap_or_else(|_| raw.to_string()), + ) +} + +/// Waits until an event targeting `key` whose `EventName` starts with +/// `event_prefix` arrives, returning the full envelope. Other envelopes are +/// discarded, so a lingering duplicate of an earlier event (delivery is +/// at-least-once) or the create event of a key that is later deleted never +/// satisfies a wait for the opposite event type. +async fn wait_for_event( + rx: &mut mpsc::UnboundedReceiver, + key: &str, + event_prefix: &str, + within: Duration, +) -> Result { + let deadline = Instant::now() + within; + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(format!("no {event_prefix}* event for key {key} within {within:?}").into()); + } + match timeout(remaining, rx.recv()).await { + Ok(Some(envelope)) => { + let key_matches = event_key(&envelope).as_deref() == Some(key); + let name_matches = envelope["EventName"].as_str().is_some_and(|n| n.starts_with(event_prefix)); + if key_matches && name_matches { + return Ok(envelope); + } + } + Ok(None) => return Err("event collector channel closed".into()), + Err(_) => return Err(format!("no {event_prefix}* event for key {key} within {within:?}").into()), + } + } +} + +/// Collects every event until `stop_key` is seen (plus a short grace window to +/// catch stragglers), or until `max_wait` elapses. Used to prove a negative: +/// non-matching keys must never appear even though a matching sentinel does. +async fn collect_until( + rx: &mut mpsc::UnboundedReceiver, + stop_key: &str, + max_wait: Duration, + grace: Duration, +) -> Vec { + let hard_deadline = Instant::now() + max_wait; + let mut soft_deadline: Option = None; + let mut collected = Vec::new(); + loop { + let deadline = soft_deadline.unwrap_or(hard_deadline).min(hard_deadline); + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + match timeout(remaining, rx.recv()).await { + Ok(Some(envelope)) => { + let matched = event_key(&envelope).as_deref() == Some(stop_key); + collected.push(envelope); + if matched && soft_deadline.is_none() { + soft_deadline = Some(Instant::now() + grace); + } + } + Ok(None) => break, + Err(_) => break, + } + } + collected +} + +// --------------------------------------------------------------------------- +// Admin target configuration (signed admin HTTP) +// --------------------------------------------------------------------------- + +async fn signed_admin_request( + env: &RustFSTestEnvironment, + method: http::Method, + url: &str, + body: Option>, +) -> Result { + let uri = url.parse::()?; + let authority = uri.authority().ok_or("admin URL missing authority")?.to_string(); + let mut builder = http::Request::builder() + .method(method.clone()) + .uri(uri) + .header(HOST, authority) + .header("x-amz-content-sha256", UNSIGNED_PAYLOAD); + if body.is_some() { + builder = builder.header(CONTENT_TYPE, "application/json"); + } + + let content_len = body.as_ref().map(|b| b.len() as i64).unwrap_or_default(); + let signed = sign_v4( + builder.body(Body::empty())?, + content_len, + &env.access_key, + &env.secret_key, + "", + "us-east-1", + ); + + let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; + let mut request = crate::common::local_http_client().request(reqwest_method, url); + for (name, value) in signed.headers() { + request = request.header(name, value); + } + if let Some(body) = body { + request = request.body(body); + } + Ok(request.send().await?) +} + +async fn enable_notify_module(env: &RustFSTestEnvironment) -> TestResult { + let payload = serde_json::json!({ + "notify_enabled": true, + "audit_enabled": false, + }); + let url = format!("{}/rustfs/admin/v3/module-switches", env.url); + let response = signed_admin_request(env, http::Method::PUT, &url, Some(payload.to_string().into_bytes())).await?; + if response.status() != StatusCode::OK { + let status = response.status(); + let text = response.text().await.unwrap_or_default(); + return Err(format!("failed to enable notify module: {status} {text}").into()); + } + Ok(()) +} + +/// Registers a webhook notification target with a persistent queue directory, so +/// delivery goes through the durable store-and-forward path. +async fn configure_webhook_target(env: &RustFSTestEnvironment, target_name: &str, endpoint: &str) -> TestResult { + let queue_dir = format!("{}/notify-queue-{target_name}", env.temp_dir); + tokio::fs::create_dir_all(&queue_dir).await?; + let payload = serde_json::json!({ + "key_values": [ + { "key": "endpoint", "value": endpoint }, + { "key": "queue_dir", "value": queue_dir }, + ] + }); + let url = format!("{}/rustfs/admin/v3/target/notify_webhook/{target_name}", env.url); + let response = signed_admin_request(env, http::Method::PUT, &url, Some(payload.to_string().into_bytes())).await?; + let status = response.status(); + if status != StatusCode::OK { + let text = response.text().await.unwrap_or_default(); + return Err(format!("failed to configure webhook target {target_name}: {status} {text}").into()); + } + Ok(()) +} + +/// Polls the admin target ARN list until the freshly configured target is +/// registered in the runtime, so notification rules and object events do not +/// race target activation. +async fn wait_for_target_registered(env: &RustFSTestEnvironment, target_name: &str) -> TestResult { + let suffix = format!(":{target_name}:webhook"); + let url = format!("{}/rustfs/admin/v3/target/arns", env.url); + for _ in 0..40 { + let response = signed_admin_request(env, http::Method::GET, &url, None).await?; + if response.status() == StatusCode::OK { + let arns: Vec = serde_json::from_slice(&response.bytes().await?)?; + if arns.iter().any(|arn| arn.ends_with(&suffix)) { + return Ok(()); + } + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + Err(format!("target {target_name} was not registered in admin ARNs").into()) +} + +/// Binds a bucket to a webhook target for ObjectCreated:*/ObjectRemoved:* events, +/// filtered to `prefix` + `suffix`. +async fn put_notification_config(client: &Client, bucket: &str, target_name: &str, prefix: &str, suffix: &str) -> TestResult { + let key_filter = S3KeyFilter::builder() + .filter_rules(FilterRule::builder().name(FilterRuleName::Prefix).value(prefix).build()) + .filter_rules(FilterRule::builder().name(FilterRuleName::Suffix).value(suffix).build()) + .build(); + let queue = QueueConfiguration::builder() + .id(format!("{target_name}-rule")) + .queue_arn(target_arn(target_name)) + .events(Event::from("s3:ObjectCreated:*")) + .events(Event::from("s3:ObjectRemoved:*")) + .filter(NotificationConfigurationFilter::builder().key(key_filter).build()) + .build()?; + let config = NotificationConfiguration::builder().queue_configurations(queue).build(); + client + .put_bucket_notification_configuration() + .bucket(bucket) + .notification_configuration(config) + .send() + .await?; + Ok(()) +} + +async fn enable_versioning(client: &Client, bucket: &str) -> TestResult { + client + .put_bucket_versioning() + .bucket(bucket) + .versioning_configuration( + VersioningConfiguration::builder() + .status(BucketVersioningStatus::Enabled) + .build(), + ) + .send() + .await?; + Ok(()) +} + +fn trimmed_etag(value: Option<&str>) -> Option { + value.map(|e| e.trim_matches('"').to_string()) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +/// PUT / multipart-complete / DELETE each deliver one event with correct fields, +/// and the prefix/suffix filter drops non-matching keys. +#[tokio::test] +#[serial] +async fn test_webhook_event_delivery_and_filtering() -> TestResult { + init_logging(); + + let (endpoint, mut rx, handle) = spawn_event_collector().await?; + + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server(vec![]).await?; + enable_notify_module(&env).await?; + + let bucket = "peri1-events"; + let target = "peri1events"; + let client = env.create_s3_client(); + client.create_bucket().bucket(bucket).send().await?; + enable_versioning(&client, bucket).await?; + + configure_webhook_target(&env, target, &endpoint).await?; + wait_for_target_registered(&env, target).await?; + put_notification_config(&client, bucket, target, "uploads/", ".dat").await?; + + // --- PUT: ObjectCreated:Put with exact eTag + versionId ------------------ + let put_key = "uploads/report.dat"; + let put = client + .put_object() + .bucket(bucket) + .key(put_key) + .body(ByteStream::from_static(b"peri-1 put body")) + .send() + .await?; + let put_version = put + .version_id() + .ok_or("PUT response missing versionId (versioning not enabled?)")?; + + let created = wait_for_event(&mut rx, put_key, "s3:ObjectCreated:", Duration::from_secs(20)).await?; + let record = &created["Records"][0]; + let object = &record["s3"]["object"]; + assert_eq!( + created["EventName"].as_str(), + Some("s3:ObjectCreated:Put"), + "envelope EventName: {created}" + ); + assert!( + record["eventName"] + .as_str() + .is_some_and(|n| n.starts_with("s3:ObjectCreated:")), + "record eventName: {record}" + ); + assert_eq!(record["s3"]["bucket"]["name"].as_str(), Some(bucket), "bucket in event: {record}"); + assert_eq!(object["versionId"].as_str(), Some(put_version), "versionId in event: {object}"); + assert_eq!( + trimmed_etag(object["eTag"].as_str()), + trimmed_etag(put.e_tag()), + "eTag in event: {object}" + ); + + // --- Multipart complete: ObjectCreated:CompleteMultipartUpload ----------- + let mp_key = "uploads/multi.dat"; + let created_mp = client.create_multipart_upload().bucket(bucket).key(mp_key).send().await?; + let upload_id = created_mp.upload_id().ok_or("missing upload id")?; + let part = client + .upload_part() + .bucket(bucket) + .key(mp_key) + .upload_id(upload_id) + .part_number(1) + .body(ByteStream::from_static(b"peri-1 multipart part body")) + .send() + .await?; + let complete = client + .complete_multipart_upload() + .bucket(bucket) + .key(mp_key) + .upload_id(upload_id) + .multipart_upload( + CompletedMultipartUpload::builder() + .parts( + CompletedPart::builder() + .part_number(1) + .e_tag(part.e_tag().unwrap_or_default()) + .build(), + ) + .build(), + ) + .send() + .await?; + + let mp_event = wait_for_event(&mut rx, mp_key, "s3:ObjectCreated:", Duration::from_secs(20)).await?; + let mp_record = &mp_event["Records"][0]; + assert_eq!( + mp_event["EventName"].as_str(), + Some("s3:ObjectCreated:CompleteMultipartUpload"), + "multipart EventName: {mp_event}" + ); + assert_eq!(mp_record["s3"]["bucket"]["name"].as_str(), Some(bucket)); + assert_eq!( + trimmed_etag(mp_record["s3"]["object"]["eTag"].as_str()), + trimmed_etag(complete.e_tag()), + "multipart eTag in event: {mp_record}" + ); + + // --- Filter: non-matching prefix and suffix must never be delivered ------ + let wrong_prefix = "logs/report.dat"; // right suffix, wrong prefix + let wrong_suffix = "uploads/report.txt"; // right prefix, wrong suffix + let sentinel = "uploads/keep.dat"; + for key in [wrong_prefix, wrong_suffix, sentinel] { + client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from_static(b"filter probe")) + .send() + .await?; + } + let collected = collect_until(&mut rx, sentinel, Duration::from_secs(20), Duration::from_secs(2)).await; + let seen: Vec = collected.iter().filter_map(event_key).collect(); + assert!( + seen.iter().any(|k| k == sentinel), + "matching sentinel {sentinel} was not delivered; saw {seen:?}" + ); + assert!( + !seen.iter().any(|k| k == wrong_prefix), + "wrong-prefix key {wrong_prefix} bypassed the filter; saw {seen:?}" + ); + assert!( + !seen.iter().any(|k| k == wrong_suffix), + "wrong-suffix key {wrong_suffix} bypassed the filter; saw {seen:?}" + ); + + // --- DELETE on a versioned bucket: ObjectRemoved:* with delete-marker version + let delete = client.delete_object().bucket(bucket).key(put_key).send().await?; + let removed = wait_for_event(&mut rx, put_key, "s3:ObjectRemoved:", Duration::from_secs(20)).await?; + let removed_record = &removed["Records"][0]; + assert!( + removed["EventName"] + .as_str() + .is_some_and(|n| n.starts_with("s3:ObjectRemoved:")), + "delete EventName: {removed}" + ); + if let Some(marker_version) = delete.version_id() { + assert_eq!( + removed_record["s3"]["object"]["versionId"].as_str(), + Some(marker_version), + "delete-marker versionId in event: {removed_record}" + ); + } + + env.stop_server(); + handle.abort(); + Ok(()) +} + +/// An event queued while the target endpoint is unreachable survives on the +/// durable store and is redelivered once the endpoint comes back. +#[tokio::test] +#[serial] +async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult { + init_logging(); + + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server(vec![]).await?; + enable_notify_module(&env).await?; + + let bucket = "peri1-redeliver"; + let target = "peri1redeliver"; + let client = env.create_s3_client(); + client.create_bucket().bucket(bucket).send().await?; + + // Reserve a port but leave it unbound: the webhook endpoint is + // unreachable (connection refused -> retryable NotConnected), so the first + // delivery attempts fail and the event is persisted to the queue store. + let port = RustFSTestEnvironment::find_available_port().await?; + let endpoint_ip = local_ip()?; + let endpoint = format!("http://{endpoint_ip}.nip.io:{port}/events"); + configure_webhook_target(&env, target, &endpoint).await?; + wait_for_target_registered(&env, target).await?; + put_notification_config(&client, bucket, target, "uploads/", ".dat").await?; + + let key = "uploads/redeliver.dat"; + client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from_static(b"queued while target down")) + .send() + .await?; + + // Bring the endpoint up on the reserved port; the replay worker retries with + // exponential backoff and delivers the queued event. + let listener = TcpListener::bind(("0.0.0.0", port)).await?; + let (tx, mut rx) = mpsc::unbounded_channel(); + let handle = serve_event_collector(listener, tx); + + let redelivered = wait_for_event(&mut rx, key, "s3:ObjectCreated:", Duration::from_secs(45)).await?; + assert_eq!( + redelivered["EventName"].as_str(), + Some("s3:ObjectCreated:Put"), + "redelivered EventName: {redelivered}" + ); + assert_eq!(redelivered["Records"][0]["s3"]["bucket"]["name"].as_str(), Some(bucket)); + + env.stop_server(); + handle.abort(); + Ok(()) +}