diff --git a/crates/e2e_test/src/notification_webhook_test.rs b/crates/e2e_test/src/notification_webhook_test.rs index dc55b111b..b436f39a9 100644 --- a/crates/e2e_test/src/notification_webhook_test.rs +++ b/crates/e2e_test/src/notification_webhook_test.rs @@ -308,7 +308,7 @@ async fn configure_webhook_target(env: &RustFSTestEnvironment, target_name: &str 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 { + for _ in 0..80 { 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?)?; @@ -537,16 +537,28 @@ async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult { 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?; + // Configure the target while its endpoint is reachable so activation and + // ARN registration complete deterministically. Registering against a dead + // endpoint stalls behind the reachability probe's timeout and flakes the + // registration wait on loaded runners (observed on the merge run of + // rustfs#4821). + let listener = TcpListener::bind("0.0.0.0:0").await?; + let port = listener.local_addr()?.port(); let endpoint_ip = local_ip()?; let endpoint = format!("http://{endpoint_ip}.nip.io:{port}/events"); + let (setup_tx, _setup_rx) = mpsc::unbounded_channel(); + let setup_handle = serve_event_collector(listener, setup_tx); + configure_webhook_target(&env, target, &endpoint).await?; wait_for_target_registered(&env, target).await?; put_notification_config(&client, bucket, target, "uploads/", ".dat").await?; + // Take the endpoint down (drops the listener, so connections are refused — + // a retryable NotConnected), then PUT: the event cannot be delivered and + // must survive on the durable queue store. + setup_handle.abort(); + let _ = setup_handle.await; + let key = "uploads/redeliver.dat"; client .put_object() @@ -556,7 +568,12 @@ async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult { .send() .await?; - // Bring the endpoint up on the reserved port; the replay worker retries with + // Hold the endpoint down long enough for at least one replay attempt to + // fail (the replay worker scans the store every 500ms), so recovery below + // exercises real redelivery rather than a first-attempt success. + tokio::time::sleep(Duration::from_secs(2)).await; + + // Bring the endpoint back on the same 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();