Merge remote-tracking branch 'origin/main' into HEAD

# Conflicts:
#	.config/e2e-full-selection.txt

crates/e2e_test/README.md

@@ -123,7 +123,7 @@ via `create_s3_client(idx)` / `create_all_clients()`. See
 | `find_available_port` | Random free port (isolation primitive) |
 | `rustfs_binary_path` / `_with_features` | Locate/build the binary; honors `RUSTFS_BUILD_FEATURES` |
 | `requested_rustfs_build_features` / `rustfs_build_feature_enabled` | Feature-gate a test to what the binary was built with |
-| `awscurl_available` + `execute_awscurl` / `awscurl_post` / `_get` / `_put` / `_delete` / `awscurl_post_sts_form_urlencoded` | Admin/STS API calls via `awscurl` (skip gracefully when absent) |
+| `execute_awscurl` / `awscurl_post` / `_get` / `_put` / `_delete` / `awscurl_post_sts_form_urlencoded` | Admin/STS API calls via `awscurl`; missing binaries are test failures |
 | `replication_fast_env` | Env vars that shrink replication timers (from repl-4); pass to `start_rustfs_server_with_env` |
 | `local_http_client` / `init_logging` | Loopback HTTP client; idempotent tracing init |
 | `RustFSTestClusterEnvironment` (`new`/`start`/`start_node`/`stop_node`/`create_all_clients`) | Multi-node harness |
@@ -189,7 +189,7 @@ cargo nextest run --profile e2e-smoke -p e2e_test
 cargo nextest run --profile e2e-full -p e2e_test
 # Cluster fault nightly lane
 cargo nextest run --profile e2e-nightly -p e2e_test
-# Replication nightly lane; install awscurl so STS paths do not skip
+# Replication nightly lane; awscurl is required for STS paths
 cargo nextest run --profile e2e-repl-nightly -p e2e_test
 # Fixed-port protocol nightly lane
 RUSTFS_BUILD_FEATURES=ftps,webdav,sftp \
@@ -221,9 +221,8 @@ The `s3s-e2e` CI job selects a random `RUSTFS_TEST_PORT` (see the `e2e-tests`
 job) to dodge this; local single-node tests already use random ports, so a
 lingering orphan is usually the cause of a spurious bind failure.
 
-**`awscurl` not found.** `awscurl`-dependent tests skip gracefully with a
-visible log line (`awscurl_available()`); install `awscurl` to actually run
-them.
+**`awscurl` not found.** `awscurl`-dependent tests fail closed with a process
+spawn error. Install the pinned CI version before running their profiles.
 
 ## Related
 
@@ -258,10 +257,9 @@ A test module may join the smoke filter only if every test in it is:
 2. **Single-node** — spawns its own server via
    `RustFSTestEnvironment`/`start_rustfs_server` on a random port with an
    isolated temp dir. No `RustFSTestClusterEnvironment`, no fixed ports.
-3. **Dependency-free** — no pre-started server at `localhost:9000`, no Vault,
-   no fixed protocol ports. Tools that may be absent on the runner (e.g.
-   `awscurl`) are acceptable only when the test skips gracefully with a
-   visible log line (see `bucket_policy_check_test.rs`).
+3. **Hermetic dependencies** — no pre-started server at `localhost:9000`, no
+   Vault, and no fixed protocol ports. Any required CLI must be pinned and
+   installed by the workflow; a missing CLI must fail the test.
 4. **Not `#[ignore]`** — ignored tests are activation work (backlog#1149
    ci-13 / backlog#1148 ilm-3), not smoke candidates.
 
@@ -278,4 +276,9 @@ listed by `cargo nextest list -p e2e_test`. Regenerate it when adding or
 moving e2e tests so acceptance numbers in the test-strategy issues
 (backlog#1147–#1155) stay auditable. When a profile membership change is
 intentional, review its JSON listing before updating the matching
-`.config/e2e-*-selection.txt` test-ID digest.
+`.config/e2e-*-selection.txt` test-ID digest. Update only the platform that
+produced the listing:
+
+```bash
+python3 scripts/check_test_wiring.py --update-profile e2e-full /path/to/listing.json linux
+```

crates/e2e_test/src/bucket_policy_check_test.rs

@@ -52,10 +52,6 @@ fn create_user_client(env: &RustFSTestEnvironment, access_key: &str, secret_key:
 #[tokio::test]
 async fn test_bucket_policy_authenticated_user() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if !crate::common::awscurl_available() {
-        info!("Skipping test_bucket_policy_authenticated_user because awscurl is not available");
-        return Ok(());
-    }
     info!("Starting test_bucket_policy_authenticated_user...");
 
     let mut env = RustFSTestEnvironment::new().await?;

crates/e2e_test/src/cluster_concurrency_test.rs

@@ -15,16 +15,27 @@
 use crate::common::RustFSTestClusterEnvironment;
 use aws_sdk_s3::Client;
 use aws_sdk_s3::error::SdkError;
+use aws_sdk_s3::types::{CorsConfiguration, CorsRule};
 use bytes::Bytes;
 use std::sync::Arc;
 use tokio::sync::Barrier;
 use tracing::{info, warn};
 
 const BUCKET: &str = "conditional-put-race-bucket";
+const BUCKET_METADATA_RELOAD_BUCKET: &str = "bucket-metadata-reload-barrier";
 
-async fn cleanup_object(client: &Client, key: &str) {
-    if let Err(e) = client.delete_object().bucket(BUCKET).key(key).send().await {
-        warn!("Failed to delete object '{}' from bucket '{}' during cleanup: {:?}", key, BUCKET, e);
+async fn cleanup_object(client: &Client, key: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    client.delete_object().bucket(BUCKET).key(key).send().await?;
+    Ok(())
+}
+
+async fn assert_bucket_cors_missing(client: &Client) {
+    let result = client.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await;
+    match result {
+        Err(SdkError::ServiceError(error)) => {
+            assert_eq!(error.err().meta().code(), Some("NoSuchCORSConfiguration"));
+        }
+        result => panic!("expected the peer to report a missing CORS configuration: {result:?}"),
     }
 }
 
@@ -71,14 +82,13 @@ async fn run_race_iteration(
     test_key: &str,
     iteration: usize,
 ) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
-    cleanup_object(&clients[0], test_key).await;
+    cleanup_object(&clients[0], test_key).await?;
     tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
 
-    let head_result = clients[0].head_object().bucket(BUCKET).key(test_key).send().await;
-
-    if head_result.is_ok() {
-        warn!("Warning: Object still exists after cleanup, skipping iteration {}", iteration);
-        return Ok(0);
+    match clients[0].head_object().bucket(BUCKET).key(test_key).send().await {
+        Ok(_) => return Err(format!("object still exists after cleanup in iteration {iteration}").into()),
+        Err(error) if error.as_service_error().is_some_and(|error| error.is_not_found()) => {}
+        Err(error) => return Err(format!("failed to verify cleanup in iteration {iteration}: {error:?}").into()),
     }
 
     info!("\n=== Iteration {} ===", iteration);
@@ -120,14 +130,16 @@ async fn run_race_iteration(
 
     info!("Result: {} out of {} succeeded", success_count, clients.len());
 
+    if had_error {
+        return Err("one or more conditional PUTs failed unexpectedly".into());
+    }
+
     if success_count > 1 {
         info!(">>> RACE CONDITION DETECTED!");
     } else if success_count == 1 {
         info!(">>> Correct behavior: exactly 1 writer succeeded.");
-    } else if had_error {
-        return Err("all conditional PUTs failed (e.g. cluster/bucket not ready)".into());
     } else {
-        info!(">>> Unexpected: no writers succeeded.");
+        return Err("no conditional PUT succeeded".into());
     }
 
     Ok(success_count)
@@ -167,7 +179,7 @@ async fn test_conditional_put_race_cluster() -> Result<(), Box<dyn std::error::E
             }
         }
 
-        cleanup_object(&clients[0], &test_key).await;
+        cleanup_object(&clients[0], &test_key).await?;
         tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
     }
 
@@ -177,7 +189,7 @@ async fn test_conditional_put_race_cluster() -> Result<(), Box<dyn std::error::E
     info!("Total iterations:   {}", iterations);
     info!("Correct (1 winner): {}", correct_count);
     info!("Race conditions:    {}", races_detected);
-    info!("Errors (skipped):   {}", error_count);
+    info!("Failed iterations:  {}", error_count);
 
     assert_eq!(races_detected, 0, "Race conditions detected: {}/{}", races_detected, iterations);
     assert_eq!(
@@ -185,6 +197,10 @@ async fn test_conditional_put_race_cluster() -> Result<(), Box<dyn std::error::E
         "{} iteration(s) failed due to errors (e.g. cluster not ready)",
         error_count
     );
+    assert_eq!(
+        correct_count, iterations,
+        "only {correct_count}/{iterations} iterations observed exactly one winner"
+    );
 
     Ok(())
 }
@@ -201,7 +217,7 @@ async fn test_conditional_put_basic_cluster() -> Result<(), Box<dyn std::error::
 
     let client = cluster.create_s3_client(0)?;
     let test_key = "basic-conditional-put";
-    cleanup_object(&client, test_key).await;
+    cleanup_object(&client, test_key).await?;
 
     let result = client
         .put_object()
@@ -233,6 +249,51 @@ async fn test_conditional_put_basic_cluster() -> Result<(), Box<dyn std::error::
         assert_eq!(code, "PreconditionFailed");
     }
 
-    cleanup_object(&client, test_key).await;
+    cleanup_object(&client, test_key).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_bucket_cors_write_is_visible_on_peer_before_response() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    crate::common::init_logging();
+
+    let mut cluster = RustFSTestClusterEnvironment::new(2).await?;
+    cluster.start().await?;
+    cluster.create_test_bucket(BUCKET_METADATA_RELOAD_BUCKET).await?;
+
+    let writer = cluster.create_s3_client(0)?;
+    let reader = cluster.create_s3_client(1)?;
+    assert_bucket_cors_missing(&reader).await;
+
+    let rule = CorsRule::builder()
+        .allowed_methods("GET")
+        .allowed_origins("https://example.com")
+        .build()?;
+    let configuration = CorsConfiguration::builder().cors_rules(rule).build()?;
+
+    writer
+        .put_bucket_cors()
+        .bucket(BUCKET_METADATA_RELOAD_BUCKET)
+        .cors_configuration(configuration)
+        .send()
+        .await?;
+
+    let response = reader.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
+    let rules = response.cors_rules();
+    assert_eq!(
+        rules.len(),
+        1,
+        "peer should observe the committed CORS rule before the write response returns"
+    );
+    assert_eq!(rules[0].allowed_methods(), ["GET"]);
+    assert_eq!(rules[0].allowed_origins(), ["https://example.com"]);
+
+    writer
+        .delete_bucket_cors()
+        .bucket(BUCKET_METADATA_RELOAD_BUCKET)
+        .send()
+        .await?;
+    assert_bucket_cors_missing(&reader).await;
+    writer.delete_bucket().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
     Ok(())
 }

crates/e2e_test/src/common.rs

@@ -499,15 +499,20 @@ fn awscurl_binary_path() -> PathBuf {
         .unwrap_or_else(|| PathBuf::from("awscurl"))
 }
 
-pub fn awscurl_available() -> bool {
-    let path = awscurl_binary_path();
-    if path.components().count() > 1 || path.is_absolute() {
-        return path.is_file();
+fn verify_awscurl_path(path: &Path) -> std::io::Result<()> {
+    let output = Command::new(path).arg("--help").output()?;
+    if output.status.success() {
+        return Ok(());
     }
 
-    std::env::var_os("PATH")
-        .map(|paths| std::env::split_paths(&paths).any(|dir| dir.join(&path).is_file()))
-        .unwrap_or(false)
+    Err(std::io::Error::other(format!(
+        "awscurl prerequisite check failed: {}",
+        String::from_utf8_lossy(&output.stderr).trim()
+    )))
+}
+
+pub fn require_awscurl() -> std::io::Result<()> {
+    verify_awscurl_path(&awscurl_binary_path())
 }
 
 // Global initialization
@@ -1752,6 +1757,22 @@ mod tests {
         assert_eq!(normalize_rustfs_build_features(" , "), None);
     }
 
+    #[test]
+    fn missing_awscurl_is_a_prerequisite_failure() {
+        let missing = std::env::temp_dir().join(format!("missing-awscurl-{}", Uuid::new_v4()));
+
+        let error = verify_awscurl_path(&missing).expect_err("a missing awscurl binary must fail the test prerequisite");
+
+        assert_eq!(error.kind(), ErrorKind::NotFound);
+    }
+
+    #[test]
+    fn available_awscurl_client_passes_prerequisite_check() {
+        let executable = std::env::current_exe().expect("the test executable should have a path");
+
+        verify_awscurl_path(&executable).expect("an available client with a working help command should pass");
+    }
+
     #[test]
     fn capture_log_path_uses_temp_directory_basename() {
         assert_eq!(

crates/e2e_test/src/existing_object_tag_policy_test.rs

@@ -16,9 +16,7 @@
 //! session policy** (`Policy` parameter) via `awscurl --service sts` with explicit
 //! `Content-Type: application/x-www-form-urlencoded` on `POST /`.
 
-use crate::common::{
-    RustFSTestEnvironment, awscurl_available, awscurl_delete, awscurl_post_sts_form_urlencoded, awscurl_put, init_logging,
-};
+use crate::common::{RustFSTestEnvironment, awscurl_delete, awscurl_post_sts_form_urlencoded, awscurl_put, init_logging};
 use aws_sdk_s3::config::{Credentials, Region};
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{Delete, ObjectIdentifier, Tag, Tagging};
@@ -175,11 +173,6 @@ async fn cleanup_bucket_and_object(admin: &Client, bucket: &str, key: &str) {
 #[tokio::test]
 async fn test_e2e_iam_policy_existing_object_tag_get_object() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if !awscurl_available() {
-        info!("Skipping test_e2e_iam_policy_existing_object_tag_get_object: awscurl not available");
-        return Ok(());
-    }
-
     let suffix = Uuid::new_v4();
     let user = format!("e2eiamtag-{suffix}");
     let user_secret = "longSecretKeyForTest123!";
@@ -233,11 +226,6 @@ async fn test_e2e_iam_policy_existing_object_tag_get_object() -> Result<(), Box<
 #[tokio::test]
 async fn test_e2e_bucket_policy_existing_object_tag_get_object() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if !awscurl_available() {
-        info!("Skipping test_e2e_bucket_policy_existing_object_tag_get_object: awscurl not available");
-        return Ok(());
-    }
-
     let suffix = Uuid::new_v4();
     let user = format!("e2ebptag-{suffix}");
     let user_secret = "longSecretKeyForTest456!";
@@ -294,11 +282,6 @@ async fn test_e2e_bucket_policy_existing_object_tag_get_object() -> Result<(), B
 #[tokio::test]
 async fn test_e2e_sts_assume_role_session_policy_existing_object_tag() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if !awscurl_available() {
-        info!("Skipping test_e2e_sts_assume_role_session_policy_existing_object_tag: awscurl not available");
-        return Ok(());
-    }
-
     let suffix = Uuid::new_v4();
     let parent = format!("e2e-sts-par-{suffix}");
     let parent_secret = "longSecretKeyForParentSts99!";
@@ -370,11 +353,6 @@ async fn test_e2e_sts_assume_role_session_policy_existing_object_tag() -> Result
 #[tokio::test]
 async fn test_e2e_sts_session_policy_delete_objects_object_prefix_only() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if !awscurl_available() {
-        info!("Skipping test_e2e_sts_session_policy_delete_objects_object_prefix_only: awscurl not available");
-        return Ok(());
-    }
-
     let suffix = Uuid::new_v4();
     let parent = format!("e2e-sts-del-par-{suffix}");
     let parent_secret = "longSecretKeyForParentDelete99!";

crates/e2e_test/src/group_delete_test.rs

@@ -14,7 +14,7 @@
 
 //! E2E tests for group management (fixes #2028).
 
-use crate::common::{RustFSTestEnvironment, admin_request, awscurl_delete, awscurl_get, awscurl_put, init_logging};
+use crate::common::{RustFSTestEnvironment, admin_ok, admin_request, init_logging};
 use aws_sdk_s3::config::{Credentials, Region};
 use aws_sdk_s3::{Client, Config};
 use tracing::info;
@@ -83,7 +83,6 @@ async fn update_group_members_rejects_invalid_new_group_names() -> Result<(), Bo
 
 /// Test that deleting a group with members fails, and deleting an empty group succeeds.
 #[tokio::test(flavor = "multi_thread")]
-#[ignore = "requires awscurl and spawns a real RustFS server"]
 async fn test_delete_group_requires_empty_membership() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
 
@@ -91,29 +90,58 @@ async fn test_delete_group_requires_empty_membership() -> Result<(), Box<dyn std
     env.start_rustfs_server(vec![]).await?;
 
     // 1. Create a user
-    let add_user_url = format!("{}/rustfs/admin/v3/add-user?accessKey=testuser1", env.url);
     let user_body = serde_json::json!({
         "secretKey": "testuser1secret",
         "status": "enabled"
     });
-    awscurl_put(&add_user_url, &user_body.to_string(), &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::PUT,
+        "/rustfs/admin/v3/add-user?accessKey=testuser1",
+        Some(user_body.to_string()),
+    )
+    .await?;
     info!("Created testuser1");
 
     // 2. Create a group with testuser1 as a member
-    let update_members_url = format!("{}/rustfs/admin/v3/update-group-members", env.url);
     let add_member_body = serde_json::json!({
         "group": "testgroup",
         "members": ["testuser1"],
         "isRemove": false,
         "groupStatus": "enabled"
     });
-    awscurl_put(&update_members_url, &add_member_body.to_string(), &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::PUT,
+        "/rustfs/admin/v3/update-group-members",
+        Some(add_member_body.to_string()),
+    )
+    .await?;
     info!("Added testuser1 to testgroup");
 
     // 3. Attempt to delete the group while it still has members — should fail
-    let delete_group_url = format!("{}/rustfs/admin/v3/group/testgroup", env.url);
-    let delete_result = awscurl_delete(&delete_group_url, &env.access_key, &env.secret_key).await;
-    assert!(delete_result.is_err(), "deleting a non-empty group should fail");
+    let (delete_status, delete_body) = admin_request(
+        &env.url,
+        http::Method::DELETE,
+        "/rustfs/admin/v3/group/testgroup",
+        None,
+        &env.access_key,
+        &env.secret_key,
+    )
+    .await?;
+    assert_eq!(
+        delete_status,
+        reqwest::StatusCode::BAD_REQUEST,
+        "deleting a non-empty group must return HTTP 400, body: {delete_body}"
+    );
+    assert!(
+        delete_body.contains("<Code>InvalidRequest</Code>"),
+        "deleting a non-empty group must return InvalidRequest, body: {delete_body}"
+    );
+    assert!(
+        delete_body.contains("<Message>group is not empty</Message>"),
+        "deleting a non-empty group returned an unexpected message: {delete_body}"
+    );
     info!("Delete of non-empty group correctly rejected");
 
     // 4. Remove the member from the group
@@ -123,17 +151,42 @@ async fn test_delete_group_requires_empty_membership() -> Result<(), Box<dyn std
         "isRemove": true,
         "groupStatus": "enabled"
     });
-    awscurl_put(&update_members_url, &remove_member_body.to_string(), &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::PUT,
+        "/rustfs/admin/v3/update-group-members",
+        Some(remove_member_body.to_string()),
+    )
+    .await?;
     info!("Removed testuser1 from testgroup");
 
     // 5. Delete the now-empty group — should succeed
-    awscurl_delete(&delete_group_url, &env.access_key, &env.secret_key).await?;
+    admin_ok(&env, http::Method::DELETE, "/rustfs/admin/v3/group/testgroup", None).await?;
     info!("Deleted empty testgroup successfully");
 
     // 6. Verify the group no longer exists
-    let get_group_url = format!("{}/rustfs/admin/v3/group?group=testgroup", env.url);
-    let get_result = awscurl_get(&get_group_url, &env.access_key, &env.secret_key).await;
-    assert!(get_result.is_err(), "group should no longer exist after deletion");
+    let (get_status, get_body) = admin_request(
+        &env.url,
+        http::Method::GET,
+        "/rustfs/admin/v3/group?group=testgroup",
+        None,
+        &env.access_key,
+        &env.secret_key,
+    )
+    .await?;
+    assert_eq!(
+        get_status,
+        reqwest::StatusCode::NOT_FOUND,
+        "a deleted group must return HTTP 404, body: {get_body}"
+    );
+    assert!(
+        get_body.contains("<Code>NoSuchResource</Code>"),
+        "a deleted group must return NoSuchResource, body: {get_body}"
+    );
+    assert!(
+        get_body.contains("<Message>group &apos;testgroup&apos; does not exist</Message>"),
+        "a deleted group returned an unexpected message: {get_body}"
+    );
     info!("Confirmed testgroup no longer exists");
 
     Ok(())
@@ -142,7 +195,6 @@ async fn test_delete_group_requires_empty_membership() -> Result<(), Box<dyn std
 /// Test that a user with only group membership (no explicit user policy) gets group policies
 /// and can perform actions allowed by the group (regression test for #2028.1).
 #[tokio::test(flavor = "multi_thread")]
-#[ignore = "requires awscurl and spawns a real RustFS server"]
 async fn test_user_with_only_group_gets_group_policies() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
 
@@ -160,39 +212,56 @@ async fn test_user_with_only_group_gets_group_policies() -> Result<(), Box<dyn s
         "Statement": [{
             "Effect": "Allow",
             "Action": ["s3:ListAllMyBuckets"],
-            "Resource": ["*"]
+            "Resource": ["arn:aws:s3:::*"]
         }]
     });
-    let add_policy_url = format!("{}/rustfs/admin/v3/add-canned-policy?name={}", env.url, policy_name);
-    awscurl_put(&add_policy_url, &policy_doc.to_string(), &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::PUT,
+        &format!("/rustfs/admin/v3/add-canned-policy?name={policy_name}"),
+        Some(policy_doc.to_string()),
+    )
+    .await?;
     info!("Created canned policy {}", policy_name);
 
     // 2. Create user with no explicit policy
-    let add_user_url = format!("{}/rustfs/admin/v3/add-user?accessKey={}", env.url, user_name);
     let user_body = serde_json::json!({
         "secretKey": user_secret,
         "status": "enabled"
     });
-    awscurl_put(&add_user_url, &user_body.to_string(), &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::PUT,
+        &format!("/rustfs/admin/v3/add-user?accessKey={user_name}"),
+        Some(user_body.to_string()),
+    )
+    .await?;
     info!("Created user {} with no explicit policy", user_name);
 
     // 3. Add user to group (creates group with this member; user_group_memberships must be updated)
-    let update_members_url = format!("{}/rustfs/admin/v3/update-group-members", env.url);
     let add_member_body = serde_json::json!({
         "group": group_name,
         "members": [user_name],
         "isRemove": false,
         "groupStatus": "enabled"
     });
-    awscurl_put(&update_members_url, &add_member_body.to_string(), &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::PUT,
+        "/rustfs/admin/v3/update-group-members",
+        Some(add_member_body.to_string()),
+    )
+    .await?;
     info!("Added {} to group {}", user_name, group_name);
 
     // 4. Attach policy to group
-    let set_policy_url = format!(
-        "{}/rustfs/admin/v3/set-user-or-group-policy?policyName={}&userOrGroup={}&isGroup=true",
-        env.url, policy_name, group_name
-    );
-    awscurl_put(&set_policy_url, "", &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::PUT,
+        &format!("/rustfs/admin/v3/set-user-or-group-policy?policyName={policy_name}&userOrGroup={group_name}&isGroup=true"),
+        Some(String::new()),
+    )
+    .await?;
     info!("Attached policy {} to group {}", policy_name, group_name);
 
     // 5. User with only group (no user policy) should be able to list buckets
@@ -209,7 +278,6 @@ async fn test_user_with_only_group_gets_group_policies() -> Result<(), Box<dyn s
 /// Test that after deleting a user who was the only member of a group, the group can be deleted
 /// (regression test for #2028.2: delete group uses backend membership, not stale cache).
 #[tokio::test(flavor = "multi_thread")]
-#[ignore = "requires awscurl and spawns a real RustFS server"]
 async fn test_delete_group_after_deleting_user() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
 
@@ -221,33 +289,47 @@ async fn test_delete_group_after_deleting_user() -> Result<(), Box<dyn std::erro
     let group_name = "soledeletegroup";
 
     // 1. Create user
-    let add_user_url = format!("{}/rustfs/admin/v3/add-user?accessKey={}", env.url, user_name);
     let user_body = serde_json::json!({
         "secretKey": user_secret,
         "status": "enabled"
     });
-    awscurl_put(&add_user_url, &user_body.to_string(), &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::PUT,
+        &format!("/rustfs/admin/v3/add-user?accessKey={user_name}"),
+        Some(user_body.to_string()),
+    )
+    .await?;
     info!("Created user {}", user_name);
 
     // 2. Add user to group
-    let update_members_url = format!("{}/rustfs/admin/v3/update-group-members", env.url);
     let add_member_body = serde_json::json!({
         "group": group_name,
         "members": [user_name],
         "isRemove": false,
         "groupStatus": "enabled"
     });
-    awscurl_put(&update_members_url, &add_member_body.to_string(), &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::PUT,
+        "/rustfs/admin/v3/update-group-members",
+        Some(add_member_body.to_string()),
+    )
+    .await?;
     info!("Added {} to group {}", user_name, group_name);
 
     // 3. Delete the user (backend and cache update so group membership becomes empty)
-    let remove_user_url = format!("{}/rustfs/admin/v3/remove-user?accessKey={}", env.url, user_name);
-    awscurl_delete(&remove_user_url, &env.access_key, &env.secret_key).await?;
+    admin_ok(
+        &env,
+        http::Method::DELETE,
+        &format!("/rustfs/admin/v3/remove-user?accessKey={user_name}"),
+        None,
+    )
+    .await?;
     info!("Deleted user {}", user_name);
 
     // 4. Deleting the group should succeed (backend has empty members; no stale cache)
-    let delete_group_url = format!("{}/rustfs/admin/v3/group/{}", env.url, group_name);
-    awscurl_delete(&delete_group_url, &env.access_key, &env.secret_key).await?;
+    admin_ok(&env, http::Method::DELETE, &format!("/rustfs/admin/v3/group/{group_name}"), None).await?;
     info!("Deleted group {} after user was removed", group_name);
 
     Ok(())

crates/e2e_test/src/kms/common.rs

@@ -22,9 +22,7 @@
 //! - KMS backend configuration (Local and Vault)
 //! - SSE encryption testing utilities
 
-use crate::common::{
-    RustFSTestEnvironment, awscurl_available, awscurl_get, awscurl_post, init_logging as common_init_logging, local_http_client,
-};
+use crate::common::{RustFSTestEnvironment, awscurl_get, awscurl_post, init_logging as common_init_logging, local_http_client};
 use aws_sdk_s3::Client;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::ServerSideEncryption;
@@ -59,15 +57,6 @@ pub fn init_logging() {
     // Additional KMS-specific logging configuration can be added here if needed
 }
 
-pub fn skip_if_kms_admin_tool_unavailable(test_name: &str) -> bool {
-    if awscurl_available() {
-        return false;
-    }
-
-    info!("Skipping {} because awscurl is not available in PATH", test_name);
-    true
-}
-
 pub fn sse_customer_key_md5_base64(key: &str) -> String {
     let mut hasher = Md5::new();
     hasher.update(key.as_bytes());
@@ -490,10 +479,6 @@ pub async fn test_kms_key_management(
     access_key: &str,
     secret_key: &str,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    if skip_if_kms_admin_tool_unavailable("test_kms_key_management") {
-        return Ok(());
-    }
-
     info!("Testing KMS key management APIs");
 
     // Test CreateKey

crates/e2e_test/src/kms/kms_local_test.rs

@@ -20,8 +20,7 @@
 //! - Complete encryption/decryption lifecycle
 
 use super::common::{
-    LocalKMSTestEnvironment, get_kms_status, skip_if_kms_admin_tool_unavailable, sse_customer_key_md5_base64,
-    test_kms_key_management, test_sse_c_encryption,
+    LocalKMSTestEnvironment, get_kms_status, sse_customer_key_md5_base64, test_kms_key_management, test_sse_c_encryption,
 };
 use crate::common::{TEST_BUCKET, init_logging};
 use tracing::{error, info};
@@ -29,9 +28,6 @@ use tracing::{error, info};
 #[tokio::test]
 async fn test_local_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if skip_if_kms_admin_tool_unavailable("test_local_kms_end_to_end") {
-        return Ok(());
-    }
     info!("Starting Local KMS End-to-End Test");
 
     // Create LocalKMS test environment

crates/e2e_test/src/kms/kms_vault_test.rs

@@ -22,8 +22,8 @@ use crate::common::{TEST_BUCKET, init_logging};
 use tracing::{error, info};
 
 use super::common::{
-    VAULT_KEY_NAME, VaultTestEnvironment, get_kms_status, skip_if_kms_admin_tool_unavailable, sse_customer_key_md5_base64,
-    start_kms, test_all_multipart_encryption_types, test_error_scenarios, test_kms_key_management, test_sse_c_encryption,
+    VAULT_KEY_NAME, VaultTestEnvironment, get_kms_status, sse_customer_key_md5_base64, start_kms,
+    test_all_multipart_encryption_types, test_error_scenarios, test_kms_key_management, test_sse_c_encryption,
     test_sse_kms_encryption, test_sse_s3_encryption,
 };
 
@@ -62,9 +62,6 @@ impl VaultKmsTestContext {
 #[tokio::test]
 async fn test_vault_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if skip_if_kms_admin_tool_unavailable("test_vault_kms_end_to_end") {
-        return Ok(());
-    }
     info!("Starting Vault KMS End-to-End Test with default key {}", VAULT_KEY_NAME);
 
     let context = VaultKmsTestContext::new().await?;
@@ -117,9 +114,6 @@ async fn test_vault_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + S
 #[tokio::test]
 async fn test_vault_kms_key_isolation() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if skip_if_kms_admin_tool_unavailable("test_vault_kms_key_isolation") {
-        return Ok(());
-    }
     info!("Starting Vault KMS SSE-C key isolation test");
 
     let context = VaultKmsTestContext::new().await?;
@@ -203,9 +197,6 @@ async fn test_vault_kms_key_isolation() -> Result<(), Box<dyn std::error::Error
 #[tokio::test]
 async fn test_vault_kms_large_file() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if skip_if_kms_admin_tool_unavailable("test_vault_kms_large_file") {
-        return Ok(());
-    }
     info!("Starting Vault KMS large file SSE-S3 test");
 
     let context = VaultKmsTestContext::new().await?;
@@ -267,9 +258,6 @@ async fn test_vault_kms_large_file() -> Result<(), Box<dyn std::error::Error + S
 #[tokio::test]
 async fn test_vault_kms_multipart_upload() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if skip_if_kms_admin_tool_unavailable("test_vault_kms_multipart_upload") {
-        return Ok(());
-    }
     info!("Starting Vault KMS multipart upload encryption suite");
 
     let context = VaultKmsTestContext::new().await?;
@@ -297,9 +285,6 @@ async fn test_vault_kms_multipart_upload() -> Result<(), Box<dyn std::error::Err
 #[tokio::test]
 async fn test_vault_kms_key_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if skip_if_kms_admin_tool_unavailable("test_vault_kms_key_operations") {
-        return Ok(());
-    }
     info!("Starting Vault KMS key operations test (CRUD)");
 
     let context = VaultKmsTestContext::new().await?;

crates/e2e_test/src/lib.rs

@@ -61,6 +61,11 @@ mod get_codec_streaming_compat_test;
 #[cfg(test)]
 mod version_id_regression_test;
 
+// Receiver-side replication LWW (rustfs/backlog#1953): stale inbound
+// replication metadata must not overwrite a newer local category state.
+#[cfg(test)]
+mod replication_lww_receiver_test;
+
 // Data usage regression tests
 #[cfg(test)]
 mod data_usage_test;

crates/e2e_test/src/mc_mirror_small_bucket_test.rs

@@ -41,13 +41,6 @@ async fn create_issue_3107_fixture(root: &Path) -> TestResult {
     Ok(())
 }
 
-fn mc_available() -> bool {
-    Command::new("mc")
-        .arg("--version")
-        .output()
-        .is_ok_and(|output| output.status.success())
-}
-
 fn run_mc(args: &[&str]) -> TestResult {
     let output = Command::new("mc").args(args).output()?;
     if !output.status.success() {
@@ -75,10 +68,7 @@ fn count_files(root: &Path) -> usize {
 async fn test_mc_mirror_small_bucket_completes_without_list_timeout() -> TestResult {
     crate::common::init_logging();
     info!("Starting issue #3107 mc mirror regression test");
-    if !mc_available() {
-        info!("Skipping issue #3107 mc mirror regression test because mc is not installed");
-        return Ok(());
-    }
+    run_mc(&["--version"])?;
 
     let mut env = RustFSTestEnvironment::new().await?;
     env.start_rustfs_server(vec![]).await?;

crates/e2e_test/src/multipart_auth_test.rs

@@ -4278,10 +4278,6 @@ async fn test_signed_put_object_extract_preserves_pax_metadata_and_version_id()
 async fn test_signed_put_object_extract_authorizes_each_pax_privilege_and_retention_conditions()
 -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     init_logging();
-    if !crate::common::awscurl_available() {
-        return Ok(());
-    }
-
     let mut env = RustFSTestEnvironment::new().await?;
     env.start_rustfs_server(vec![]).await?;
 

crates/e2e_test/src/protocols/sftp_helpers.rs

@@ -26,7 +26,7 @@ use aws_sdk_s3::config::{Credentials, Region};
 use aws_smithy_http_client::Builder as SmithyHttpClientBuilder;
 use russh::client::{self, Handle};
 use russh::keys::ssh_key::LineEnding;
-use russh::keys::{Algorithm, PrivateKey, PublicKey};
+use russh::keys::{Algorithm, PrivateKey, PublicKeyOrCertificate};
 use russh_sftp::client::SftpSession;
 use russh_sftp::protocol::OpenFlags;
 use std::path::Path;
@@ -46,7 +46,7 @@ pub struct AcceptAnyServerKey;
 impl client::Handler for AcceptAnyServerKey {
     type Error = anyhow::Error;
 
-    async fn check_server_key(&mut self, _server_public_key: &PublicKey) -> Result<bool, Self::Error> {
+    async fn check_server_key(&mut self, _server_public_key: &PublicKeyOrCertificate) -> Result<bool, Self::Error> {
         Ok(true)
     }
 }

crates/e2e_test/src/quota_test.rs

@@ -18,15 +18,6 @@ use http::{Method, StatusCode};
 use tokio::time::{Duration, sleep, timeout};
 use tracing::{debug, info};
 
-fn skip_without_awscurl() -> bool {
-    if crate::common::awscurl_available() {
-        return false;
-    }
-
-    info!("Skipping quota test because awscurl is not available");
-    true
-}
-
 /// Test environment setup for quota tests
 pub struct QuotaTestEnv {
     pub env: RustFSTestEnvironment,
@@ -276,9 +267,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_basic_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         // Create test bucket
@@ -320,9 +308,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_admission_aws_chunked_declared_encoding() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
         env.create_bucket().await?;
 
@@ -371,9 +356,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_update_and_clear() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;
@@ -406,9 +388,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_delete_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;
@@ -442,9 +421,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_usage_tracking() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;
@@ -480,9 +456,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_statistics() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;
@@ -513,9 +486,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_check_api() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;
@@ -553,9 +523,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_multiple_buckets() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         // Create two buckets in the same environment
@@ -593,9 +560,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_error_handling() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;
@@ -628,9 +592,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_http_endpoints() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;
@@ -689,9 +650,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_normal_user_permissions() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
         env.create_bucket().await?;
 
@@ -744,9 +702,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_copy_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;
@@ -789,9 +744,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_batch_delete() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;
@@ -847,9 +799,6 @@ mod integration_tests {
     #[tokio::test]
     async fn test_quota_multipart_upload() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         init_logging();
-        if skip_without_awscurl() {
-            return Ok(());
-        }
         let env = QuotaTestEnv::new().await?;
 
         env.create_bucket().await?;

crates/e2e_test/src/reliant/node_interact_test.rs

@@ -13,207 +13,228 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::common::workspace_root;
+use crate::common::RustFSTestEnvironment;
 use crate::storage_api::node_interact::{
     TonicInterceptor, VolumeInfo, WalkDirOptions, gen_tonic_signature_interceptor, node_service_time_out_client,
 };
-use futures::future::join_all;
+use aws_sdk_s3::primitives::ByteStream;
 use rmp_serde::{Deserializer, Serializer};
-use rustfs_filemeta::{MetaCacheEntry, MetacacheReader, MetacacheWriter};
+use rustfs_filemeta::MetaCacheEntry;
 use rustfs_protos::proto_gen::node_service::WalkDirRequest;
 use rustfs_protos::{
     models::{PingBody, PingBodyBuilder},
-    proto_gen::node_service::{
-        ListVolumesRequest, LocalStorageInfoRequest, MakeVolumeRequest, PingRequest, PingResponse, ReadAllRequest,
-    },
+    proto_gen::node_service::{ListVolumesRequest, LocalStorageInfoRequest, MakeVolumeRequest, PingRequest, ReadAllRequest},
 };
 use serde::{Deserialize, Serialize};
 use std::error::Error;
 use std::io::Cursor;
-use std::path::PathBuf;
-use tokio::spawn;
 use tonic::Request;
 use tonic::codegen::tokio_stream::StreamExt;
 
-const CLUSTER_ADDR: &str = "http://localhost:9000";
+type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
+
+const TEST_RPC_SECRET: &str = "rustfs-internode-signature-e2e-secret";
 
 fn signature_interceptor() -> TonicInterceptor {
     TonicInterceptor::Signature(gen_tonic_signature_interceptor())
 }
 
+fn rpc_client_error(error: Box<dyn Error>) -> std::io::Error {
+    std::io::Error::other(error.to_string())
+}
+
+async fn start_server() -> Result<RustFSTestEnvironment, Box<dyn Error + Send + Sync>> {
+    let _ = rustfs_credentials::set_global_rpc_secret(TEST_RPC_SECRET.to_string());
+    let effective = rustfs_credentials::try_get_rpc_token().expect("RPC secret must resolve in the test process");
+    assert_eq!(effective, TEST_RPC_SECRET, "the test process uses an unexpected RPC secret");
+
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server_without_cleanup_with_env(&[
+        ("RUSTFS_RPC_SECRET", TEST_RPC_SECRET),
+        ("RUSTFS_INTERNODE_RPC_SIGNATURE_STRICT", "false"),
+        ("RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT", "false"),
+        ("RUSTFS_INTERNODE_RPC_REPLAY_SCOPE_STRICT", "false"),
+        ("RUST_LOG", "error"),
+    ])
+    .await?;
+    Ok(env)
+}
+
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn ping() -> Result<(), Box<dyn Error>> {
+async fn ping() -> TestResult {
+    let env = start_server().await?;
     let mut fbb = flatbuffers::FlatBufferBuilder::new();
     let payload = fbb.create_vector(b"hello world");
-
     let mut builder = PingBodyBuilder::new(&mut fbb);
     builder.add_payload(payload);
     let root = builder.finish();
     fbb.finish(root, None);
 
-    let finished_data = fbb.finished_data();
-
-    let decoded_payload = flatbuffers::root::<PingBody>(finished_data);
-    assert!(decoded_payload.is_ok());
-
-    // Create client
-    let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string(), signature_interceptor()).await?;
-
-    // Construct PingRequest
-    let request = Request::new(PingRequest {
-        version: 1,
-        body: bytes::Bytes::copy_from_slice(finished_data),
-    });
-
-    // Send request and get response
-    let response: PingResponse = client.ping(request).await?.into_inner();
-
-    // Print response
-    let ping_response_body = flatbuffers::root::<PingBody>(&response.body);
-    if let Err(e) = ping_response_body {
-        eprintln!("{e}");
-    } else {
-        println!("ping_resp:body(flatbuffer): {ping_response_body:?}");
-    }
+    let mut client = node_service_time_out_client(&env.url, signature_interceptor())
+        .await
+        .map_err(rpc_client_error)?;
+    let response = client
+        .ping(Request::new(PingRequest {
+            version: 1,
+            body: bytes::Bytes::copy_from_slice(fbb.finished_data()),
+        }))
+        .await?
+        .into_inner();
 
+    assert_eq!(response.version, 1);
+    let body = flatbuffers::root::<PingBody>(&response.body)?;
+    assert_eq!(body.payload().expect("ping response must contain a payload").bytes(), b"hello, caller");
     Ok(())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn make_volume() -> Result<(), Box<dyn Error>> {
-    let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string(), signature_interceptor()).await?;
-    let request = Request::new(MakeVolumeRequest {
-        disk: "data".to_string(),
-        volume: "dandan".to_string(),
-    });
+async fn make_volume() -> TestResult {
+    let env = start_server().await?;
+    let mut client = node_service_time_out_client(&env.url, signature_interceptor())
+        .await
+        .map_err(rpc_client_error)?;
+    let response = client
+        .make_volume(Request::new(MakeVolumeRequest {
+            disk: env.temp_dir.clone(),
+            volume: "node-rpc-volume".to_string(),
+        }))
+        .await?
+        .into_inner();
 
-    let response = client.make_volume(request).await?.into_inner();
-    if response.success {
-        println!("success");
-    } else {
-        println!("failed: {:?}", response.error);
-    }
+    assert!(response.success, "make_volume failed: {:?}", response.error);
+    assert!(std::path::Path::new(&env.temp_dir).join("node-rpc-volume").is_dir());
     Ok(())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn list_volumes() -> Result<(), Box<dyn Error>> {
-    let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string(), signature_interceptor()).await?;
-    let request = Request::new(ListVolumesRequest {
-        disk: "data".to_string(),
-    });
+async fn list_volumes() -> TestResult {
+    let env = start_server().await?;
+    let mut client = node_service_time_out_client(&env.url, signature_interceptor())
+        .await
+        .map_err(rpc_client_error)?;
+    let created = client
+        .make_volume(Request::new(MakeVolumeRequest {
+            disk: env.temp_dir.clone(),
+            volume: "node-rpc-listed-volume".to_string(),
+        }))
+        .await?
+        .into_inner();
+    assert!(created.success, "make_volume failed: {:?}", created.error);
 
-    let response = client.list_volumes(request).await?.into_inner();
-    let volume_infos: Vec<VolumeInfo> = response
+    let response = client
+        .list_volumes(Request::new(ListVolumesRequest {
+            disk: env.temp_dir.clone(),
+        }))
+        .await?
+        .into_inner();
+    assert!(response.success, "list_volumes failed: {:?}", response.error);
+    let volumes = response
         .volume_infos
-        .into_iter()
-        .filter_map(|json_str| serde_json::from_str::<VolumeInfo>(&json_str).ok())
-        .collect();
-
-    println!("{volume_infos:?}");
+        .iter()
+        .map(|json| serde_json::from_str::<VolumeInfo>(json))
+        .collect::<Result<Vec<_>, _>>()?;
+    assert!(volumes.iter().any(|volume| volume.name == "node-rpc-listed-volume"));
     Ok(())
 }
 
 #[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn walk_dir() -> Result<(), Box<dyn Error>> {
-    println!("walk_dir");
-    // TODO: use writer
+async fn walk_dir() -> TestResult {
+    let env = start_server().await?;
+    let s3 = env.create_s3_client();
+    let bucket = "node-rpc-walk-bucket";
+    let key = "prefix/object.txt";
+    env.create_test_bucket(bucket).await?;
+    s3.put_object()
+        .bucket(bucket)
+        .key(key)
+        .body(ByteStream::from_static(b"walk payload"))
+        .send()
+        .await?;
+
     let opts = WalkDirOptions {
-        bucket: "dandan".to_owned(),
-        base_dir: "".to_owned(),
+        bucket: bucket.to_string(),
         recursive: true,
         ..Default::default()
     };
-    let (rd, mut wr) = tokio::io::duplex(1024);
-    let mut buf = Vec::new();
-    opts.serialize(&mut Serializer::new(&mut buf))?;
-    let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string(), signature_interceptor()).await?;
-    let disk_path = std::env::var_os("RUSTFS_DISK_PATH").map(PathBuf::from).unwrap_or_else(|| {
-        let mut path = workspace_root();
-        path.push("target");
-        path.push(if cfg!(debug_assertions) { "debug" } else { "release" });
-        path.push("data");
-        path
-    });
-    let request = Request::new(WalkDirRequest {
-        disk: disk_path.to_string_lossy().into_owned(),
-        walk_dir_options: buf.into(),
-    });
-    let mut response = client.walk_dir(request).await?.into_inner();
+    let mut encoded = Vec::new();
+    opts.serialize(&mut Serializer::new(&mut encoded))?;
+    let mut client = node_service_time_out_client(&env.url, signature_interceptor())
+        .await
+        .map_err(rpc_client_error)?;
+    let mut stream = client
+        .walk_dir(Request::new(WalkDirRequest {
+            disk: env.temp_dir.clone(),
+            walk_dir_options: encoded.into(),
+        }))
+        .await?
+        .into_inner();
 
-    let job1 = spawn(async move {
-        let mut out = MetacacheWriter::new(&mut wr);
-        loop {
-            match response.next().await {
-                Some(Ok(resp)) => {
-                    if !resp.success {
-                        println!("{}", resp.error_info.unwrap_or_else(|| "".to_string()));
-                    }
-                    let entry = serde_json::from_str::<MetaCacheEntry>(&resp.meta_cache_entry)
-                        .map_err(|_e| std::io::Error::other(format!("Unexpected response: {response:?}")))
-                        .unwrap();
-                    out.write_obj(&entry).await.unwrap();
-                }
-                None => {
-                    let _ = out.close().await;
-                    break;
-                }
-                _ => {
-                    println!("Unexpected response: {response:?}");
-                    let _ = out.close().await;
-                    break;
-                }
-            }
-        }
-    });
-    let job2 = spawn(async move {
-        let mut reader = MetacacheReader::new(rd);
-        while let Ok(Some(entry)) = reader.peek().await {
-            println!("{entry:?}");
-        }
-    });
-
-    join_all(vec![job1, job2]).await;
-    Ok(())
-}
-
-#[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn read_all() -> Result<(), Box<dyn Error>> {
-    let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string(), signature_interceptor()).await?;
-    let request = Request::new(ReadAllRequest {
-        disk: "data".to_string(),
-        volume: "ff".to_string(),
-        path: "format.json".to_string(),
-    });
-
-    let response = client.read_all(request).await?.into_inner();
-    let volume_infos = response.data;
-
-    println!("{}", response.success);
-    println!("{volume_infos:?}");
-    Ok(())
-}
-
-#[tokio::test]
-#[ignore = "requires running RustFS server at localhost:9000"]
-async fn storage_info() -> Result<(), Box<dyn Error>> {
-    let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string(), signature_interceptor()).await?;
-    let request = Request::new(LocalStorageInfoRequest { metrics: true });
-
-    let response = client.local_storage_info(request).await?.into_inner();
-    if !response.success {
-        println!("{:?}", response.error_info);
-        return Ok(());
+    let mut entries = Vec::new();
+    while let Some(response) = stream.next().await {
+        let response = response?;
+        assert!(response.success, "walk_dir failed: {:?}", response.error_info);
+        entries.push(serde_json::from_str::<MetaCacheEntry>(&response.meta_cache_entry)?);
     }
-    let info = response.storage_info;
-
-    let mut buf = Deserializer::new(Cursor::new(info));
-    let storage_info: rustfs_madmin::StorageInfo = Deserialize::deserialize(&mut buf).unwrap();
-    println!("{storage_info:?}");
+    assert!(
+        entries.iter().any(|entry| entry.name == key),
+        "walk_dir did not return {key}: {entries:?}"
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn read_all() -> TestResult {
+    let env = start_server().await?;
+    let mut client = node_service_time_out_client(&env.url, signature_interceptor())
+        .await
+        .map_err(rpc_client_error)?;
+    let volume = "node-rpc-read-volume";
+    let created = client
+        .make_volume(Request::new(MakeVolumeRequest {
+            disk: env.temp_dir.clone(),
+            volume: volume.to_string(),
+        }))
+        .await?
+        .into_inner();
+    assert!(created.success, "make_volume failed: {:?}", created.error);
+    tokio::fs::write(std::path::Path::new(&env.temp_dir).join(volume).join("payload.bin"), b"read payload").await?;
+
+    let response = client
+        .read_all(Request::new(ReadAllRequest {
+            disk: env.temp_dir.clone(),
+            volume: volume.to_string(),
+            path: "payload.bin".to_string(),
+        }))
+        .await?
+        .into_inner();
+    assert!(response.success, "read_all failed: {:?}", response.error);
+    assert_eq!(response.data.as_ref(), b"read payload");
+    Ok(())
+}
+
+#[tokio::test]
+async fn storage_info() -> TestResult {
+    let env = start_server().await?;
+    let mut client = node_service_time_out_client(&env.url, signature_interceptor())
+        .await
+        .map_err(rpc_client_error)?;
+    let response = client
+        .local_storage_info(Request::new(LocalStorageInfoRequest { metrics: true }))
+        .await?
+        .into_inner();
+    assert!(response.success, "local_storage_info failed: {:?}", response.error_info);
+
+    let mut decoder = Deserializer::new(Cursor::new(response.storage_info));
+    let storage_info: rustfs_madmin::StorageInfo = Deserialize::deserialize(&mut decoder)?;
+    let expected_disk = std::fs::canonicalize(&env.temp_dir)?;
+    assert!(!storage_info.disks.is_empty(), "local_storage_info returned no disks");
+    assert!(
+        storage_info
+            .disks
+            .iter()
+            .any(|disk| std::path::Path::new(&disk.drive_path) == expected_disk),
+        "local_storage_info did not include the configured disk: {:?}",
+        storage_info.disks
+    );
     Ok(())
 }

crates/e2e_test/src/replication_extension_test.rs

@@ -13,9 +13,8 @@
 // limitations under the License.
 
 use crate::common::{
-    RustFSTestEnvironment, admin_create_user, awscurl_available, awscurl_post_sts_form_urlencoded, init_logging,
-    local_http_client, replication_fast_env, rustfs_binary_path, signed_request, signed_request_with_client,
-    signed_request_with_session_token,
+    RustFSTestEnvironment, admin_create_user, awscurl_post_sts_form_urlencoded, init_logging, local_http_client,
+    replication_fast_env, rustfs_binary_path, signed_request, signed_request_with_client, signed_request_with_session_token,
 };
 use crate::fake_s3_target::{
     FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation,
@@ -57,7 +56,7 @@ use rustfs_madmin::{
     AddServiceAccountReq, ListServiceAccountsResp, PeerInfo, PeerSite, ReplicateAddStatus, ReplicateEditStatus,
     ReplicateRemoveStatus, SRRemoveReq, SRResyncOpStatus, SRStatusInfo, SiteReplicationInfo, SyncStatus,
 };
-use s3s::header::X_AMZ_REPLICATION_STATUS;
+use s3s::header::{X_AMZ_REPLICATION_STATUS, X_AMZ_TAGGING};
 use sha2::{Digest, Sha256};
 use std::collections::BTreeMap;
 use std::convert::Infallible;
@@ -2023,6 +2022,7 @@ async fn forward_replication_proxy_request(
     client: &reqwest::Client,
     request_count: &AtomicU64,
     mut replication_enabled: watch::Receiver<bool>,
+    mut held_tagging: watch::Receiver<Option<String>>,
 ) -> Response<Full<bytes::Bytes>> {
     let (parts, body) = request.into_parts();
     let is_replication = parts
@@ -2036,6 +2036,17 @@ async fn forward_replication_proxy_request(
                 return proxy_error_response("replication gate closed");
             }
         }
+        // Content-keyed hold: park only the replication request whose
+        // `x-amz-tagging` matches the held value, letting every other delivery
+        // through, so a test can make one specific (stale) delivery the last
+        // write the backend sees.
+        if let Some(tagging) = parts.headers.get(X_AMZ_TAGGING).and_then(|value| value.to_str().ok()) {
+            while held_tagging.borrow().as_deref() == Some(tagging) {
+                if held_tagging.changed().await.is_err() {
+                    return proxy_error_response("replication tag hold closed");
+                }
+            }
+        }
     }
 
     let Some(path_and_query) = parts.uri.path_and_query() else {
@@ -2070,12 +2081,26 @@ async fn start_replication_counting_proxy(
     backend_url: &str,
     tasks: &mut JoinSet<()>,
 ) -> Result<(String, Arc<AtomicU64>, watch::Sender<bool>), Box<dyn Error + Send + Sync>> {
+    let (proxy_url, request_count, replication_enabled, _held_tagging) =
+        start_replication_counting_proxy_with_tag_hold(backend_url, tasks).await?;
+    Ok((proxy_url, request_count, replication_enabled))
+}
+
+/// [`start_replication_counting_proxy`] plus a content-keyed hold: while the
+/// returned `watch::Sender<Option<String>>` holds `Some(tagging)`, replication
+/// requests whose `x-amz-tagging` equals `tagging` are parked (and still
+/// counted); all other traffic flows. Send `None` to release them.
+async fn start_replication_counting_proxy_with_tag_hold(
+    backend_url: &str,
+    tasks: &mut JoinSet<()>,
+) -> Result<(String, Arc<AtomicU64>, watch::Sender<bool>, watch::Sender<Option<String>>), Box<dyn Error + Send + Sync>> {
     let listener = TcpListener::bind("127.0.0.1:0").await?;
     let proxy_url = format!("http://{}", listener.local_addr()?);
     let backend_url = backend_url.to_string();
     let request_count = Arc::new(AtomicU64::new(0));
     let task_request_count = request_count.clone();
     let (replication_enabled, task_replication_enabled) = watch::channel(true);
+    let (held_tagging, task_held_tagging) = watch::channel(None);
     tasks.spawn(async move {
         let client = local_http_client();
         let mut connections = JoinSet::new();
@@ -2087,12 +2112,14 @@ async fn start_replication_counting_proxy(
                     let client = client.clone();
                     let request_count = task_request_count.clone();
                     let replication_enabled = task_replication_enabled.clone();
+                    let held_tagging = task_held_tagging.clone();
                     connections.spawn(async move {
                         let service = service_fn(move |request| {
                             let backend_url = backend_url.clone();
                             let client = client.clone();
                             let request_count = request_count.clone();
                             let replication_enabled = replication_enabled.clone();
+                            let held_tagging = held_tagging.clone();
                             async move {
                                 Ok::<_, Infallible>(
                                     forward_replication_proxy_request(
@@ -2101,6 +2128,7 @@ async fn start_replication_counting_proxy(
                                         &client,
                                         &request_count,
                                         replication_enabled,
+                                        held_tagging,
                                     )
                                     .await,
                                 )
@@ -2113,7 +2141,7 @@ async fn start_replication_counting_proxy(
             }
         }
     });
-    Ok((proxy_url, request_count, replication_enabled))
+    Ok((proxy_url, request_count, replication_enabled, held_tagging))
 }
 
 async fn site_replication_remove(
@@ -6949,6 +6977,395 @@ async fn test_site_replication_active_active_converges_without_loops_real_dual_n
     }
 }
 
+/// Replication status a site reports for one object version via HEAD
+/// (`x-amz-replication-status`), or `None` when the header is absent.
+async fn head_replication_status(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+) -> Result<Option<String>, Box<dyn Error + Send + Sync>> {
+    let head = client
+        .head_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(version_id)
+        .send()
+        .await?;
+    Ok(head.replication_status().map(|status| status.as_str().to_string()))
+}
+
+/// Poll one site until the version's replication status is one of `expected`.
+async fn wait_for_version_replication_status(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    expected: &[&str],
+    site: &str,
+) -> Result<String, Box<dyn Error + Send + Sync>> {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+    loop {
+        let last = head_replication_status(client, bucket, key, version_id).await?;
+        if let Some(status) = last.as_deref()
+            && expected.contains(&status)
+        {
+            return Ok(status.to_string());
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return Err(format!(
+                "{site}: {bucket}/{key}?versionId={version_id} replication status {last:?} never reached {expected:?}"
+            )
+            .into());
+        }
+        sleep(Duration::from_millis(200)).await;
+    }
+}
+
+async fn put_single_tag(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    tag_key: &str,
+    tag_value: &str,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    client
+        .put_object_tagging()
+        .bucket(bucket)
+        .key(key)
+        .version_id(version_id)
+        .tagging(
+            aws_sdk_s3::types::Tagging::builder()
+                .tag_set(aws_sdk_s3::types::Tag::builder().key(tag_key).value(tag_value).build()?)
+                .build()?,
+        )
+        .send()
+        .await?;
+    Ok(())
+}
+
+async fn get_single_tag(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    tag_key: &str,
+) -> Result<Option<String>, Box<dyn Error + Send + Sync>> {
+    let tagging = client
+        .get_object_tagging()
+        .bucket(bucket)
+        .key(key)
+        .version_id(version_id)
+        .send()
+        .await?;
+    Ok(tagging
+        .tag_set()
+        .iter()
+        .find(|tag| tag.key() == tag_key)
+        .map(|tag| tag.value().to_string()))
+}
+
+/// Poll one site until the version's `tag_key` equals `expected`.
+async fn wait_for_single_tag(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    tag_key: &str,
+    expected: &str,
+    site: &str,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+    loop {
+        let observed = get_single_tag(client, bucket, key, version_id, tag_key).await?;
+        if observed.as_deref() == Some(expected) {
+            return Ok(());
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return Err(format!(
+                "{site}: {bucket}/{key}?versionId={version_id} tag {tag_key}={observed:?} never became {expected}"
+            )
+            .into());
+        }
+        sleep(Duration::from_millis(200)).await;
+    }
+}
+
+/// Tag key the dual-node LWW scenario edits on both sites.
+const LWW_TAG_KEY: &str = "owner";
+
+/// Assert the version's [`LWW_TAG_KEY`] stays `expected` on both sites for a
+/// full quiet window (no late stale delivery flips it back).
+async fn assert_tag_stable_on_both_sites(
+    site_a_client: &Client,
+    site_b_client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    expected: &str,
+    quiet: Duration,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    let deadline = tokio::time::Instant::now() + quiet;
+    loop {
+        let on_a = get_single_tag(site_a_client, bucket, key, version_id, LWW_TAG_KEY).await?;
+        let on_b = get_single_tag(site_b_client, bucket, key, version_id, LWW_TAG_KEY).await?;
+        assert_eq!(on_a.as_deref(), Some(expected), "site A tag {LWW_TAG_KEY} regressed from the LWW winner");
+        assert_eq!(on_b.as_deref(), Some(expected), "site B tag {LWW_TAG_KEY} regressed from the LWW winner");
+        if tokio::time::Instant::now() >= deadline {
+            return Ok(());
+        }
+        sleep(Duration::from_millis(250)).await;
+    }
+}
+
+/// Wait until the counting proxy in front of a site has admitted `expected`
+/// replication requests in total (requests held by a closed gate still count).
+async fn wait_for_proxy_replication_requests(
+    counter: &AtomicU64,
+    expected: u64,
+    site: &str,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+    loop {
+        let observed = counter.load(Ordering::Relaxed);
+        if observed >= expected {
+            return Ok(());
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return Err(format!("{site} proxy saw {observed} replication requests, expected at least {expected}").into());
+        }
+        sleep(Duration::from_millis(25)).await;
+    }
+}
+
+/// rustfs/backlog#1953 (audit A4/P1-6): receiver-side LWW for replicated
+/// metadata categories, exercised end to end over the real dual-node
+/// active-active site-replication control plane — sender, worker, status
+/// bookkeeping and persisted failure recovery all participate (the single-server
+/// `replication_lww_receiver_test` only injects authorized replication PUTs).
+///
+/// Scenario on one versioned object:
+/// 1. reciprocal tag edits in real order (A then B) converge both sites on the
+///    newer tag and leave the author COMPLETED / the receiver REPLICA;
+/// 2. out-of-order delivery: A's edit is held at B's inbound proxy while B
+///    authors a newer edit that reaches A first; releasing the stale delivery
+///    must NOT roll B back — both sites settle on B's value and stay there
+///    through a quiet window, with no FAILED/PENDING status left behind;
+/// 3. persisted retry: B is stopped, A's delivery reaches FAILED, A restarts,
+///    then B returns and the scanner-replayed edit converges both sites forward.
+/// Durable metadata-MRF serialization/reconstruction is covered separately by
+/// `metadata_mrf_roundtrip_preserves_tags_and_admitted_targets`.
+#[tokio::test]
+async fn test_site_replication_tagging_lww_converges_active_active_real_dual_node() -> TestResult {
+    init_logging();
+
+    match tokio::time::timeout(Duration::from_secs(420), async {
+        // The scanner is fast for the final persisted-failure recovery phase.
+        // Step 2 finishes and proves a quiet stable winner before that phase,
+        // so a later scanner pass cannot mask its stale-delivery assertion.
+        let mut site_env = replication_fast_env();
+        site_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+        site_env.extend_from_slice(FAST_SCANNER_ENV);
+
+        let mut site_a_env = RustFSTestEnvironment::new().await?;
+        site_a_env.start_rustfs_server_with_env(vec![], &site_env).await?;
+
+        let mut site_b_env = RustFSTestEnvironment::new().await?;
+        site_b_env.start_rustfs_server_with_env(vec![], &site_env).await?;
+
+        let mut proxy_tasks = JoinSet::new();
+        let (site_a_proxy, site_a_replication_requests, _site_a_replication_enabled, site_a_held_tagging) =
+            start_replication_counting_proxy_with_tag_hold(&site_a_env.url, &mut proxy_tasks).await?;
+        let (site_b_proxy, site_b_replication_requests, _site_b_replication_enabled, site_b_held_tagging) =
+            start_replication_counting_proxy_with_tag_hold(&site_b_env.url, &mut proxy_tasks).await?;
+
+        let site_a_client = site_a_env.create_s3_client();
+        let site_b_client = site_b_env.create_s3_client();
+        let bucket = "site-repl-tag-lww";
+        let key = "lww.txt";
+
+        let add_status = site_replication_add(
+            &site_a_env,
+            &[
+                PeerSite {
+                    name: "lww-site-a".to_string(),
+                    endpoint: site_a_env.url.clone(),
+                    access_key: site_a_env.access_key.clone(),
+                    secret_key: site_a_env.secret_key.clone(),
+                    ..Default::default()
+                },
+                PeerSite {
+                    name: "lww-site-b".to_string(),
+                    endpoint: site_b_env.url.clone(),
+                    access_key: site_b_env.access_key.clone(),
+                    secret_key: site_b_env.secret_key.clone(),
+                    ..Default::default()
+                },
+            ],
+        )
+        .await?;
+        assert!(add_status.success, "unexpected site add result: {add_status:?}");
+
+        let site_info = wait_for_site_replication_enabled(&site_a_env, 2).await?;
+        wait_for_site_replication_enabled(&site_b_env, 2).await?;
+
+        // Route both directions through the counting proxies so inbound
+        // replication to B can be held (out-of-order delivery) and observed.
+        for (env_url, proxy_url, label) in [(&site_a_env.url, &site_a_proxy, "A"), (&site_b_env.url, &site_b_proxy, "B")] {
+            let mut peer = site_info
+                .sites
+                .iter()
+                .find(|peer| peer.endpoint == *env_url)
+                .ok_or_else(|| format!("site {label} peer missing from replication info"))?
+                .clone();
+            peer.endpoint = proxy_url.clone();
+            peer.sync_state = SyncStatus::Enable;
+            let edit = site_replication_edit(&site_a_env, "", &peer).await?;
+            assert!(edit.success, "unexpected site {label} endpoint edit: {edit:?}");
+        }
+        for env in [&site_a_env, &site_b_env] {
+            wait_for_site_replication_info(env, |info| {
+                info.sites.iter().any(|peer| peer.endpoint == site_a_proxy)
+                    && info.sites.iter().any(|peer| peer.endpoint == site_b_proxy)
+            })
+            .await?;
+        }
+
+        site_a_client.create_bucket().bucket(bucket).send().await?;
+        wait_for_bucket_on_target(&site_b_client, bucket).await?;
+
+        let version_id = site_a_client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(ByteStream::from_static(b"tag lww payload"))
+            .send()
+            .await?
+            .version_id()
+            .ok_or("site A PUT omitted version ID")?
+            .to_string();
+        wait_for_replicated_object(&site_b_client, bucket, key, "tag lww payload").await?;
+        wait_for_version_replication_status(&site_a_client, bucket, key, &version_id, &["COMPLETED"], "site A").await?;
+        wait_for_version_replication_status(&site_b_client, bucket, key, &version_id, &["REPLICA"], "site B").await?;
+        wait_for_proxy_replication_requests(&site_b_replication_requests, 1, "site B").await?;
+
+        // --- 1. reciprocal edits in real order: A then B ----------------------
+        put_single_tag(&site_a_client, bucket, key, &version_id, LWW_TAG_KEY, "a1").await?;
+        wait_for_single_tag(&site_b_client, bucket, key, &version_id, LWW_TAG_KEY, "a1", "site B").await?;
+        wait_for_version_replication_status(&site_a_client, bucket, key, &version_id, &["COMPLETED"], "site A").await?;
+        wait_for_version_replication_status(&site_b_client, bucket, key, &version_id, &["REPLICA"], "site B").await?;
+        wait_for_proxy_replication_requests(&site_b_replication_requests, 2, "site B").await?;
+
+        put_single_tag(&site_b_client, bucket, key, &version_id, LWW_TAG_KEY, "b1").await?;
+        wait_for_single_tag(&site_a_client, bucket, key, &version_id, LWW_TAG_KEY, "b1", "site A").await?;
+        wait_for_version_replication_status(&site_b_client, bucket, key, &version_id, &["COMPLETED"], "site B").await?;
+        wait_for_version_replication_status(&site_a_client, bucket, key, &version_id, &["REPLICA"], "site A").await?;
+        assert_tag_stable_on_both_sites(&site_a_client, &site_b_client, bucket, key, &version_id, "b1", Duration::from_secs(3))
+            .await?;
+
+        // --- 2. concurrent edits, stale delivery last ------------------------
+        // Both sites edit the same version while each other's delivery is
+        // parked at the peer's inbound proxy (content-keyed: only the
+        // `owner=a2` / `owner=b2` replication PUTs wait, everything else
+        // flows). B's edit is the newer one. Releasing A's stale `a2` first
+        // makes it the last write B sees while A itself still holds `a2`, so
+        // nothing A could re-deliver carries the winner: only receiver-side
+        // LWW on B can keep `b2`. Releasing `b2` afterwards converges A.
+        site_b_held_tagging.send(Some("owner=a2".to_string()))?;
+        site_a_held_tagging.send(Some("owner=b2".to_string()))?;
+        let a2_parked_at = site_b_replication_requests.load(Ordering::Relaxed) + 1;
+        let b2_parked_at = site_a_replication_requests.load(Ordering::Relaxed) + 1;
+        put_single_tag(&site_a_client, bucket, key, &version_id, LWW_TAG_KEY, "a2").await?;
+        wait_for_proxy_replication_requests(&site_b_replication_requests, a2_parked_at, "site B").await?;
+        sleep(Duration::from_millis(50)).await;
+        put_single_tag(&site_b_client, bucket, key, &version_id, LWW_TAG_KEY, "b2").await?;
+        wait_for_proxy_replication_requests(&site_a_replication_requests, b2_parked_at, "site A").await?;
+        assert_eq!(
+            get_single_tag(&site_a_client, bucket, key, &version_id, LWW_TAG_KEY)
+                .await?
+                .as_deref(),
+            Some("a2")
+        );
+        assert_eq!(
+            get_single_tag(&site_b_client, bucket, key, &version_id, LWW_TAG_KEY)
+                .await?
+                .as_deref(),
+            Some("b2")
+        );
+
+        // Release the stale a2 delivery onto B: the newer local b2 must
+        // survive, and the delivery itself must still succeed (A reaches
+        // COMPLETED instead of looping through MRF with the stale value).
+        site_b_held_tagging.send(None)?;
+        wait_for_version_replication_status(&site_a_client, bucket, key, &version_id, &["COMPLETED"], "site A").await?;
+        let stale_deadline = tokio::time::Instant::now() + Duration::from_secs(3);
+        loop {
+            assert_eq!(
+                get_single_tag(&site_b_client, bucket, key, &version_id, LWW_TAG_KEY)
+                    .await?
+                    .as_deref(),
+                Some("b2"),
+                "a stale inbound delivery rolled back site B's newer tag (receiver-side LWW regression)"
+            );
+            if tokio::time::Instant::now() >= stale_deadline {
+                break;
+            }
+            sleep(Duration::from_millis(250)).await;
+        }
+
+        // Release b2 onto A: the newer edit wins there and both sites settle.
+        // B's own version may legitimately read REPLICA here: the stale inbound
+        // a2 write re-labelled it as a replica write (keeping B's tags); what
+        // must not remain is PENDING/FAILED.
+        site_a_held_tagging.send(None)?;
+        wait_for_single_tag(&site_a_client, bucket, key, &version_id, LWW_TAG_KEY, "b2", "site A").await?;
+        wait_for_version_replication_status(&site_b_client, bucket, key, &version_id, &["COMPLETED", "REPLICA"], "site B")
+            .await?;
+        assert_tag_stable_on_both_sites(&site_a_client, &site_b_client, bucket, key, &version_id, "b2", Duration::from_secs(4))
+            .await?;
+        for (client, site) in [(&site_a_client, "site A"), (&site_b_client, "site B")] {
+            let status = head_replication_status(client, bucket, key, &version_id).await?;
+            assert!(
+                matches!(status.as_deref(), Some("COMPLETED" | "REPLICA")),
+                "{site} must not be left PENDING/FAILED after the concurrent edits: {status:?}"
+            );
+        }
+
+        // --- 3. persisted FAILED state survives a source restart ------------
+        site_b_env.stop_server();
+        put_single_tag(&site_a_client, bucket, key, &version_id, LWW_TAG_KEY, "a3").await?;
+        wait_for_version_replication_status(&site_a_client, bucket, key, &version_id, &["FAILED"], "site A").await?;
+        site_a_env.restart_server_preserving_data(vec![], &site_env).await?;
+        wait_for_site_replication_enabled(&site_a_env, 2).await?;
+        site_b_env.restart_server_preserving_data(vec![], &site_env).await?;
+        wait_for_site_replication_enabled(&site_b_env, 2).await?;
+
+        wait_for_single_tag(&site_b_client, bucket, key, &version_id, LWW_TAG_KEY, "a3", "site B").await?;
+        wait_for_version_replication_status(&site_a_client, bucket, key, &version_id, &["COMPLETED"], "site A").await?;
+        wait_for_version_replication_status(&site_b_client, bucket, key, &version_id, &["REPLICA"], "site B").await?;
+        assert_tag_stable_on_both_sites(&site_a_client, &site_b_client, bucket, key, &version_id, "a3", Duration::from_secs(3))
+            .await?;
+
+        // The object itself never forked: one version on each side.
+        tokio::time::timeout(
+            Duration::from_secs(70),
+            assert_replication_converged(&site_a_client, bucket, &site_b_client, bucket),
+        )
+        .await??;
+        let state = list_replication_state(&site_a_client, bucket).await?;
+        assert_eq!(state.len(), 1, "tag edits must not create new object versions: {state:?}");
+        assert_eq!(state[0].version_id, version_id);
+
+        proxy_tasks.abort_all();
+        Ok(())
+    })
+    .await
+    {
+        Ok(result) => result,
+        Err(_) => Err("site replication tagging LWW test timed out".into()),
+    }
+}
 #[tokio::test]
 async fn test_site_replication_replicates_policy_backed_user_access_real_dual_node() -> Result<(), Box<dyn Error + Send + Sync>> {
     init_logging();
@@ -7281,11 +7698,6 @@ async fn test_site_replication_replicates_multiple_service_accounts_real_dual_no
 async fn test_site_replication_replicates_service_accounts_created_from_sts_session_real_dual_node() -> TestResult {
     init_logging();
 
-    if !awscurl_available() {
-        eprintln!("Skipping STS site replication service-account test because awscurl is unavailable");
-        return Ok(());
-    }
-
     let mut source_env = RustFSTestEnvironment::new().await?;
     source_env
         .start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)

crates/e2e_test/src/replication_lww_receiver_test.rs

@@ -0,0 +1,155 @@
+#![cfg(test)]
+// 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.
+
+//! Receiver-side replication LWW over the wire (rustfs/backlog#1953, audit
+//! A4/P1-6).
+//!
+//! In an active-active topology both sites' metadata states arrive at the
+//! peer as authorized replication PUTs carrying per-category source
+//! timestamps (`x-rustfs-source-replication-tagging-timestamp` header
+//! family). Before the fix the receiver applied them unconditionally, so a
+//! stale delivery overwrote a newer local state and the two sites diverged
+//! permanently while both reported COMPLETED. This test drives one live
+//! `rustfs` server with simulated inbound replication PUTs for the same
+//! object version and asserts the newer tagging state wins regardless of
+//! delivery order, while a stale delivery still succeeds at the object level
+//! (a failure would loop through MRF re-delivering the stale value).
+//!
+//! The real dual-site path (sender, worker, status bookkeeping, MRF replay)
+//! is covered by
+//! `replication_extension_test::test_site_replication_tagging_lww_converges_active_active_real_dual_node`;
+//! this file stays as the fast, single-process receiver check.
+
+use crate::common::{RustFSTestEnvironment, init_logging};
+use aws_sdk_s3::Client;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
+
+type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
+
+const HDR_SOURCE_REPLICATION_REQUEST: &str = "x-rustfs-source-replication-request";
+const HDR_SOURCE_VERSION_ID: &str = "x-rustfs-source-version-id";
+const HDR_SOURCE_MTIME: &str = "x-rustfs-source-mtime";
+const HDR_SOURCE_TAGGING_TIMESTAMP: &str = "x-rustfs-source-replication-tagging-timestamp";
+
+const SOURCE_MTIME: &str = "2026-01-01T00:00:00Z";
+const T_STALE: &str = "2026-01-01T00:00:01Z";
+const T_LOCAL: &str = "2026-02-01T00:00:00Z";
+const T_NEWER: &str = "2026-03-01T00:00:00Z";
+
+/// Simulated inbound authorized replication PUT: same object version, tags and
+/// the source-authored tagging timestamp carried in transport headers.
+async fn inbound_replication_put(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    tags: &str,
+    tagging_timestamp: &str,
+) -> TestResult {
+    let version_id = version_id.to_string();
+    let tagging_timestamp = tagging_timestamp.to_string();
+    client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .body(ByteStream::from_static(b"lww-e2e-body"))
+        .tagging(tags)
+        .customize()
+        .mutate_request(move |req| {
+            req.headers_mut().insert(HDR_SOURCE_REPLICATION_REQUEST, "true");
+            req.headers_mut().insert(HDR_SOURCE_VERSION_ID, version_id.clone());
+            req.headers_mut().insert(HDR_SOURCE_MTIME, SOURCE_MTIME);
+            req.headers_mut()
+                .insert(HDR_SOURCE_TAGGING_TIMESTAMP, tagging_timestamp.clone());
+        })
+        .send()
+        .await?;
+    Ok(())
+}
+
+async fn tag_value(client: &Client, bucket: &str, key: &str, version_id: &str, tag_key: &str) -> Option<String> {
+    let tagging = client
+        .get_object_tagging()
+        .bucket(bucket)
+        .key(key)
+        .version_id(version_id)
+        .send()
+        .await
+        .expect("object tagging should be readable");
+    tagging
+        .tag_set()
+        .iter()
+        .find(|tag| tag.key() == tag_key)
+        .map(|tag| tag.value().to_string())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn receiver_lww_keeps_newer_tags_across_delivery_orders() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    let client = env.create_s3_client();
+
+    let bucket = "replication-lww-receiver";
+    let key = "object";
+    client.create_bucket().bucket(bucket).send().await?;
+    client
+        .put_bucket_versioning()
+        .bucket(bucket)
+        .versioning_configuration(
+            VersioningConfiguration::builder()
+                .status(BucketVersioningStatus::Enabled)
+                .build(),
+        )
+        .send()
+        .await?;
+
+    // First delivery establishes version V with tags stamped T_LOCAL.
+    let version_id = uuid::Uuid::new_v4().to_string();
+    inbound_replication_put(&client, bucket, key, &version_id, "site=local", T_LOCAL).await?;
+    assert_eq!(
+        tag_value(&client, bucket, key, &version_id, "site").await.as_deref(),
+        Some("local"),
+        "the first delivery must establish the tagged version"
+    );
+
+    // A stale delivery (older source timestamp) must succeed at the object
+    // level but must NOT overwrite the newer tags.
+    inbound_replication_put(&client, bucket, key, &version_id, "site=stale", T_STALE).await?;
+    assert_eq!(
+        tag_value(&client, bucket, key, &version_id, "site").await.as_deref(),
+        Some("local"),
+        "a stale inbound delivery must not overwrite newer tags (rustfs/backlog#1953)"
+    );
+
+    // A newer delivery still converges the version onto the newest state.
+    inbound_replication_put(&client, bucket, key, &version_id, "site=newer", T_NEWER).await?;
+    assert_eq!(
+        tag_value(&client, bucket, key, &version_id, "site").await.as_deref(),
+        Some("newer"),
+        "a newer inbound delivery must overwrite older tags"
+    );
+
+    client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(&version_id)
+        .send()
+        .await?;
+    env.delete_test_bucket(bucket).await.ok();
+    Ok(())
+}

crates/e2e_test/src/security_boundary_test.rs

@@ -21,12 +21,11 @@
 //! - SSRF prevention (internal/private endpoints rejected for tiering)
 //! - Race condition handling (concurrent writes converge without corruption)
 
-use crate::common::{RustFSTestEnvironment, awscurl_available, awscurl_put, init_logging};
+use crate::common::{RustFSTestEnvironment, awscurl_put, init_logging, require_awscurl};
 use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart, Tag, Tagging};
 use std::error::Error;
-use tracing::info;
 
 /// Oversized tagging payloads must be rejected by the per-object tag limit.
 ///
@@ -225,16 +224,12 @@ async fn test_concurrent_object_operations() -> Result<(), Box<dyn Error + Send
 /// outcome — the internal endpoint is not accepted — is asserted here.
 ///
 /// The admin API is exercised via signed `awscurl` requests, matching the
-/// pattern used by the other admin-API E2E tests in this crate; the test is
-/// skipped when `awscurl` is not installed.
+/// pattern used by the other admin-API E2E tests in this crate. The full E2E
+/// lane installs and verifies the pinned `awscurl` prerequisite.
 #[tokio::test]
 async fn test_tiering_url_validation() -> Result<(), Box<dyn Error + Send + Sync>> {
     init_logging();
-    if !awscurl_available() {
-        info!("Skipping tiering URL validation test because awscurl is not available");
-        return Ok(());
-    }
-
+    require_awscurl()?;
     let mut env = RustFSTestEnvironment::new().await?;
     env.start_rustfs_server(vec![]).await?;