mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
test(e2e): deflake webhook redelivery target registration (#4848)
Register the webhook target while its endpoint is reachable, then drop the listener before the PUT: registering against a dead endpoint stalls behind the reachability probe timeout and flaked the ARN wait on the rustfs#4821 merge run. Also widen the registration wait to 20s.
This commit is contained in:
@@ -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<String> = 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();
|
||||
|
||||
Reference in New Issue
Block a user