mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-25 13:36:50 +00:00
test(e2e): restore webhook redelivery regression coverage (#5690)
test(e2e): unquarantine webhook redelivery regression
This commit is contained in:
@@ -24,7 +24,7 @@
|
|||||||
//! * PUT / multipart-complete / DeleteObject / DeleteObjects each deliver one event with the correct
|
//! * PUT / multipart-complete / DeleteObject / DeleteObjects each deliver one event with the correct
|
||||||
//! eventName, bucket, key, versionId and eTag.
|
//! eventName, bucket, key, versionId and eTag.
|
||||||
//! * prefix/suffix filters drop non-matching keys (rule-engine gate).
|
//! * prefix/suffix filters drop non-matching keys (rule-engine gate).
|
||||||
//! * an event queued while the target endpoint is unreachable is redelivered
|
//! * an event queued while the target endpoint rejects delivery is redelivered
|
||||||
//! from the on-disk store once the endpoint recovers (store-and-forward).
|
//! from the on-disk store once the endpoint recovers (store-and-forward).
|
||||||
//! * responseElements and the S3 response use the canonical request ID while
|
//! * responseElements and the S3 response use the canonical request ID while
|
||||||
//! requestParameters preserve a conflicting client-supplied value.
|
//! requestParameters preserve a conflicting client-supplied value.
|
||||||
@@ -897,11 +897,10 @@ async fn test_webhook_event_delivery_and_filtering() -> TestResult {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// An event queued while the target endpoint is unreachable survives on the
|
/// An event queued while the target endpoint rejects delivery survives on the
|
||||||
/// durable store and is redelivered once the endpoint comes back.
|
/// durable store and is redelivered once the endpoint comes back.
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
#[serial]
|
#[serial]
|
||||||
#[ignore = "FAILING deterministically on main since it landed (#4821): the target is created but never appears in /rustfs/admin/v3/target/arns, so wait_for_target_registered times out. Quarantined per the flake policy; remove with the fix for rustfs#4852"]
|
|
||||||
async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult {
|
async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult {
|
||||||
init_logging();
|
init_logging();
|
||||||
|
|
||||||
@@ -932,28 +931,55 @@ async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult {
|
|||||||
wait_for_target_registered(&env, target).await?;
|
wait_for_target_registered(&env, target).await?;
|
||||||
put_notification_config(&client, bucket, target, "uploads/", ".dat").await?;
|
put_notification_config(&client, bucket, target, "uploads/", ".dat").await?;
|
||||||
|
|
||||||
// Take the endpoint down (drops the listener, so connections are refused —
|
// Replace the healthy setup listener with one that rejects the first POST.
|
||||||
// a retryable NotConnected), then PUT: the event cannot be delivered and
|
// Waiting for that response below proves the queued event reached a failed
|
||||||
// must survive on the durable queue store.
|
// delivery attempt before the endpoint recovers.
|
||||||
setup_handle.abort();
|
setup_handle.abort();
|
||||||
let _ = setup_handle.await;
|
let _ = setup_handle.await;
|
||||||
|
|
||||||
|
let listener = TcpListener::bind(("0.0.0.0", port)).await?;
|
||||||
let key = "uploads/redeliver.dat";
|
let key = "uploads/redeliver.dat";
|
||||||
client
|
client
|
||||||
.put_object()
|
.put_object()
|
||||||
.bucket(bucket)
|
.bucket(bucket)
|
||||||
.key(key)
|
.key(key)
|
||||||
.body(ByteStream::from_static(b"queued while target down"))
|
.body(ByteStream::from_static(b"queued while target rejects"))
|
||||||
.send()
|
.send()
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
// Hold the endpoint down long enough for at least one replay attempt to
|
let mut failure_handle = tokio::spawn(async move {
|
||||||
// fail (the replay worker scans the store every 500ms), so recovery below
|
loop {
|
||||||
// exercises real redelivery rather than a first-attempt success.
|
let (mut stream, _) = listener.accept().await?;
|
||||||
tokio::time::sleep(Duration::from_secs(2)).await;
|
let (method, _) = timeout(Duration::from_secs(5), read_http_message(&mut stream)).await??;
|
||||||
|
if method == "HEAD" {
|
||||||
|
stream
|
||||||
|
.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
|
||||||
|
.await?;
|
||||||
|
stream.shutdown().await?;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if method == "POST" {
|
||||||
|
stream
|
||||||
|
.write_all(b"HTTP/1.1 503 Service Unavailable\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
|
||||||
|
.await?;
|
||||||
|
stream.shutdown().await?;
|
||||||
|
return Ok::<(), BoxError>(());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
// Bring the endpoint back on the same port; the replay worker retries with
|
let rejected = match timeout(Duration::from_secs(20), &mut failure_handle).await {
|
||||||
// exponential backoff and delivers the queued event.
|
Ok(rejected) => rejected,
|
||||||
|
Err(_) => {
|
||||||
|
failure_handle.abort();
|
||||||
|
let _ = failure_handle.await;
|
||||||
|
return Err("webhook replay did not reach the rejecting endpoint".into());
|
||||||
|
}
|
||||||
|
};
|
||||||
|
rejected??;
|
||||||
|
|
||||||
|
// Bring the endpoint back on the same port; the replay worker rescans the
|
||||||
|
// durable queue and delivers the retained event.
|
||||||
let listener = TcpListener::bind(("0.0.0.0", port)).await?;
|
let listener = TcpListener::bind(("0.0.0.0", port)).await?;
|
||||||
let (tx, mut rx) = mpsc::unbounded_channel();
|
let (tx, mut rx) = mpsc::unbounded_channel();
|
||||||
let handle = serve_event_collector(listener, tx);
|
let handle = serve_event_collector(listener, tx);
|
||||||
|
|||||||
@@ -66,7 +66,7 @@
|
|||||||
| multipart_storage_class_test | 3 | ✅ |
|
| multipart_storage_class_test | 3 | ✅ |
|
||||||
| namespace_lock_quorum_test | 2 | |
|
| namespace_lock_quorum_test | 2 | |
|
||||||
| negative_sigv4_test | 6 | ✅ |
|
| negative_sigv4_test | 6 | ✅ |
|
||||||
| notification_webhook_test | 2 | ✅ |
|
| notification_webhook_test | 3 | ✅ |
|
||||||
| object_lambda_test | 16 | |
|
| object_lambda_test | 16 | |
|
||||||
| object_lock | 33 | |
|
| object_lock | 33 | |
|
||||||
| overwrite_cleanup_regression_test | 1 | |
|
| overwrite_cleanup_regression_test | 1 | |
|
||||||
@@ -88,6 +88,4 @@
|
|||||||
| tls_hot_reload_test | 1 | ✅ |
|
| tls_hot_reload_test | 1 | ✅ |
|
||||||
| version_id_regression_test | 10 | ✅ |
|
| version_id_regression_test | 10 | ✅ |
|
||||||
|
|
||||||
`notification_webhook_test` also has 1 ignored store-and-forward regression tracked by rustfs#4852; ignored tests are excluded from the active counts above.
|
**Total listed: 528 tests across 70 modules · PR smoke subset: 148 tests / 33 modules** (31 full modules + 18 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 28 tests** · generated 2026-08-04.
|
||||||
|
|
||||||
**Total listed: 527 tests across 70 modules · PR smoke subset: 147 tests / 33 modules** (31 full modules + 18 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 28 tests** · generated 2026-07-30.
|
|
||||||
|
|||||||
Reference in New Issue
Block a user