test(e2e): fix cluster nightly oracles (#6397)

.github/workflows/e2e-replication-nightly.yml

@@ -196,6 +196,9 @@ jobs:
           cache-save-if: 'false'
           install-build-packaging-tools: 'false'
 
+      - name: Verify protocol socket oracle
+        run: ss -tn state CLOSE-WAIT >/dev/null
+
       # The suite owns fixed protocol ports and serializes its internal cases.
       - name: Verify protocol e2e membership
         env:

crates/e2e_test/src/object_lambda_test.rs

@@ -16,6 +16,7 @@ use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_lo
 use aws_sdk_s3::primitives::ByteStream;
 use http::header::{CONTENT_TYPE, HOST};
 use reqwest::StatusCode;
+use rustfs_config::{ENV_DRIVE_ACTIVE_CHECK_INTERVAL_SECS, ENV_NOTIFY_ENABLE};
 use rustfs_signer::pre_sign_v4;
 use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS;
 use s3s::Body;
@@ -976,7 +977,8 @@ async fn test_get_object_lambda_rejects_disabled_target() -> Result<(), Box<dyn
     init_logging();
 
     let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    env.start_rustfs_server_with_env(vec![], &[(ENV_NOTIFY_ENABLE, "true")])
+        .await?;
 
     let bucket = "object-lambda-e2e-disabled-target";
     let key = "input.txt";
@@ -992,17 +994,24 @@ async fn test_get_object_lambda_rejects_disabled_target() -> Result<(), Box<dyn
         .send()
         .await?;
 
-    configure_webhook_target_with_key_values(
+    let queue_dir = format!("{}/disabled-target-queue", env.temp_dir);
-        &env,
+    tokio::fs::create_dir_all(&queue_dir).await?;
-        "transformer",
+    let config_url = format!("{}/rustfs/admin/v3/set-config-kv", env.url);
-        vec![
+    let directive = format!(
-            ("endpoint", "http://127.0.0.1:9/transform".to_string()),
+        "notify_webhook:transformer enable=off endpoint=\"http://127.0.0.1:9/transform\" auth_token=\"secret-token\" queue_dir=\"{queue_dir}\""
-            ("auth_token", "secret-token".to_string()),
+    );
-            ("enable", "off".to_string()),
+    let disable_response = signed_request(
-        ],
+        http::Method::PUT,
+        &config_url,
+        &env.access_key,
+        &env.secret_key,
+        Some(directive.into_bytes()),
+        Some("text/plain"),
     )
     .await?;
-    wait_for_target_visibility(&env, "transformer").await?;
+    let disable_status = disable_response.status();
+    let disable_body = disable_response.text().await?;
+    assert_eq!(disable_status, StatusCode::OK, "failed to disable target: {disable_body}");
 
     let lambda_url = format!("{}/{}/{}?lambdaArn={}", env.url, bucket, key, urlencoding::encode(lambda_arn));
     let response = signed_request(http::Method::GET, &lambda_url, &env.access_key, &env.secret_key, None, None).await?;
@@ -1021,7 +1030,8 @@ async fn test_configure_object_lambda_target_rejects_invalid_endpoint() -> Resul
     init_logging();
 
     let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    env.start_rustfs_server_with_env(vec![], &[(ENV_NOTIFY_ENABLE, "true")])
+        .await?;
 
     let bucket = "object-lambda-e2e-invalid-endpoint";
 
@@ -1064,7 +1074,8 @@ async fn test_configure_object_lambda_notify_webhook_rejects_response_header_tim
     init_logging();
 
     let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    env.start_rustfs_server_with_env(vec![], &[(ENV_NOTIFY_ENABLE, "true")])
+        .await?;
 
     let response = send_configure_webhook_target_request(
         &env,
@@ -1173,6 +1184,8 @@ async fn test_listen_notification_fans_in_remote_node_events() -> Result<(), Box
     init_logging();
 
     let mut cluster = RustFSTestClusterEnvironment::new(2).await?;
+    cluster.set_env(ENV_NOTIFY_ENABLE, "true");
+    cluster.set_env(ENV_DRIVE_ACTIVE_CHECK_INTERVAL_SECS, "1");
     cluster.start().await?;
 
     let bucket = "listen-notification-cluster";

crates/e2e_test/src/stale_multipart_cleanup_cluster_test.rs

@@ -15,7 +15,6 @@
 use crate::common::{RustFSTestClusterEnvironment, init_logging};
 use aws_sdk_s3::error::SdkError;
 use aws_sdk_s3::primitives::ByteStream;
-use aws_sdk_s3::types::CompletedMultipartUpload;
 use tokio::time::{Duration, sleep};
 use tracing::info;
 use uuid::Uuid;
@@ -43,32 +42,18 @@ async fn list_parts_reports_missing_upload(
     }
 }
 
-async fn complete_reports_missing_upload(
+async fn multipart_listing_reports_missing_upload(
     client: &aws_sdk_s3::Client,
     bucket: &str,
     key: &str,
     upload_id: &str,
 ) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
-    let result = client
+    let result = client.list_multipart_uploads().bucket(bucket).prefix(key).send().await?;
-        .complete_multipart_upload()
+
-        .bucket(bucket)
+    Ok(!result
-        .key(key)
+        .uploads()
-        .upload_id(upload_id)
+        .iter()
-        .multipart_upload(CompletedMultipartUpload::builder().build())
+        .any(|upload| upload.key() == Some(key) && upload.upload_id() == Some(upload_id)))
-        .send()
-        .await;
-    match result {
-        Ok(_) => Ok(false),
-        Err(SdkError::ServiceError(err)) => {
-            let code = err.err().meta().code().unwrap_or("");
-            if code == "NoSuchUpload" {
-                Ok(true)
-            } else {
-                Err(format!("unexpected complete_multipart_upload service error: code={code}, err={err:?}").into())
-            }
-        }
-        Err(err) => Err(format!("unexpected complete_multipart_upload error: {err:?}").into()),
-    }
 }
 
 async fn wait_for_cleanup_on_all_nodes(
@@ -81,8 +66,8 @@ async fn wait_for_cleanup_on_all_nodes(
         let mut all_cleaned = true;
         for (idx, client) in clients.iter().enumerate() {
             let list_parts_missing = list_parts_reports_missing_upload(client, bucket, key, upload_id).await?;
-            let complete_missing = complete_reports_missing_upload(client, bucket, key, upload_id).await?;
+            let listing_missing = multipart_listing_reports_missing_upload(client, bucket, key, upload_id).await?;
-            if !(list_parts_missing && complete_missing) {
+            if !(list_parts_missing && listing_missing) {
                 info!("stale multipart still visible on node {} at attempt {}", idx, attempt + 1);
                 all_cleaned = false;
                 break;
@@ -146,6 +131,10 @@ async fn test_stale_multipart_cleanup_removes_incomplete_upload_across_cluster()
         1,
         "multipart upload should be visible before background cleanup"
     );
+    assert!(
+        !multipart_listing_reports_missing_upload(&clients[2], CLEANUP_BUCKET, &key, &upload_id).await?,
+        "multipart upload listing should contain the upload before background cleanup"
+    );
 
     wait_for_cleanup_on_all_nodes(&clients, CLEANUP_BUCKET, &key, &upload_id).await?;