Merge branch 'main' into overtrue/activate-group-e2e

test(kms): stabilize Vault failover validation (#6385 )
* test(kms): bound Vault failover progress wait * ci(nightly): honor manual dispatch ref * test(kms): preserve Vault worker failures * test(kms): validate Vault circuit recovery
2026-08-23 04:39:04 +00:00 · 2026-08-23 12:37:24 +08:00 · 2026-08-23 12:32:11 +08:00 · 2026-08-23 12:29:52 +08:00 · 2026-08-23 11:11:01 +08:00 · 2026-08-23 08:55:07 +08:00
9 changed files with 256 additions and 306 deletions
@@ -1,2 +1,2 @@
-sha256-darwin=9f767b37ed8b1c82da62ea441462d75487785c8086e56f08fb6f6cd89c6e2e52
-sha256-linux=fbdaf42b220958d4b1e8880e0f8b5a7992d38e21051bb60596dd4538424757d6
+sha256-darwin=f832043fcca8c0b616c5d820a3a652da7544298ef5812a8668a3a9a3e4607b8b
+sha256-linux=93b94adb110b86a41d0b7313909e0bf53cb1515e2d08e8f105652b29b249990f
@@ -39,11 +39,10 @@ jobs:
    env:
      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
    steps:
-      - name: Checkout main branch
+      - name: Checkout repository
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
          persist-credentials: false
-          ref: main

      - name: Setup Rust environment
        uses: ./.github/actions/setup
@@ -89,11 +88,10 @@ jobs:
      # either casing.
      NO_PROXY: 127.0.0.1,localhost
    steps:
-      - name: Checkout main branch
+      - name: Checkout repository
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
          persist-credentials: false
-          ref: main

      - name: Setup Rust environment
        uses: ./.github/actions/setup
@@ -178,11 +176,10 @@ jobs:
      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
      NO_PROXY: 127.0.0.1,localhost
    steps:
-      - name: Checkout main branch
+      - name: Checkout repository
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
          persist-credentials: false
-          ref: main

      - name: Setup Rust environment
        uses: ./.github/actions/setup
@@ -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(())
@@ -32,10 +32,6 @@ pub struct Credentials {
    pub access_key: String,
    #[serde(rename = "secretKey")]
    pub secret_key: String,
-    // The aliases accept madmin's JSON tags (MinIO-written bucket-targets
-    // metadata and mc request bodies) without changing the snake_case
-    // persisted/peer wire format this struct serializes to.
-    #[serde(alias = "sessionToken")]
    pub session_token: Option<String>,
    pub expiration: Option<Timestamp>,
 }
@@ -206,14 +202,12 @@ pub struct BucketTarget {
    #[serde(default)]
    pub region: String,

-    // madmin-go v3.0.109 tags this `bandwidthlimit`; `bandwidth` is a legacy
-    // alias kept for inputs written before the madmin tag was verified.
-    #[serde(alias = "bandwidthlimit", alias = "bandwidth", default)]
+    #[serde(alias = "bandwidth", default)]
    pub bandwidth_limit: i64,

    #[serde(rename = "replicationSync", default)]
    pub replication_sync: bool,
-    #[serde(alias = "storageclass", default)]
+    #[serde(default)]
    pub storage_class: String,
    #[serde(rename = "skipTlsVerify", default)]
    pub skip_tls_verify: bool,
@@ -226,7 +220,7 @@ pub struct BucketTarget {

    #[serde(rename = "resetBeforeDate", with = "time::serde::rfc3339::option", default)]
    pub reset_before_date: Option<OffsetDateTime>,
-    #[serde(alias = "resetID", default)]
+    #[serde(default)]
    pub reset_id: String,
    #[serde(rename = "totalDowntime", with = "duration_seconds", default)]
    pub total_downtime: Duration,
@@ -239,7 +233,7 @@ pub struct BucketTarget {
    #[serde(default)]
    pub latency: LatencyStat,

-    #[serde(alias = "deploymentID", default)]
+    #[serde(default)]
    pub deployment_id: String,

    #[serde(default)]
@@ -537,85 +531,6 @@ mod tests {
        assert_eq!(value["totalDowntime"], 90);
    }

-    #[test]
-    fn bucket_target_persisted_wire_keys_stay_snake_case() {
-        // bucket-targets.json (persisted via `serde_json::to_vec(&BucketTargets)`
-        // in the admin set/remove handlers) and the msgpack struct-map form
-        // (`BucketTargets::marshal_msg`) both come straight from this struct's
-        // serde field names. madmin naming is applied only in the admin
-        // response layer (`remote_target_admin_json`); renaming here would
-        // silently break every existing deployment's persisted metadata.
-        let targets = BucketTargets {
-            targets: vec![BucketTarget {
-                credentials: Some(Credentials {
-                    access_key: "ak".to_string(),
-                    secret_key: "sk".to_string(),
-                    session_token: Some("token".to_string()),
-                    expiration: None,
-                }),
-                bandwidth_limit: 5,
-                storage_class: "STANDARD".to_string(),
-                reset_id: "reset-1".to_string(),
-                deployment_id: "deploy-1".to_string(),
-                ..Default::default()
-            }],
-        };
-
-        let json = serde_json::to_value(&targets).expect("targets should serialize to JSON");
-        let msgpack: serde_json::Value =
-            rmp_serde::from_slice(&targets.marshal_msg().expect("targets should marshal to msgpack"))
-                .expect("msgpack struct map should decode into a JSON value");
-
-        for (wire, entry) in [("JSON", &json["targets"][0]), ("msgpack", &msgpack["targets"][0])] {
-            assert_eq!(entry["bandwidth_limit"], 5, "{wire} key `bandwidth_limit` must stay");
-            assert_eq!(entry["storage_class"], "STANDARD", "{wire} key `storage_class` must stay");
-            assert_eq!(entry["reset_id"], "reset-1", "{wire} key `reset_id` must stay");
-            assert_eq!(entry["deployment_id"], "deploy-1", "{wire} key `deployment_id` must stay");
-            assert_eq!(entry["credentials"]["session_token"], "token", "{wire} key `session_token` must stay");
-        }
-    }
-
-    #[test]
-    fn minio_written_bucket_targets_json_populates_madmin_named_fields() {
-        // A MinIO-written bucket-targets.json carries madmin's JSON tags
-        // (`bandwidthlimit`, `storageclass`, `resetID`, `deploymentID`,
-        // `credentials.sessionToken` — madmin-go v3.0.109 bucket-targets.go).
-        // On migration these must land in the matching fields instead of
-        // silently defaulting (backlog#1951).
-        let targets: BucketTargets = serde_json::from_value(serde_json::json!({
-            "targets": [{
-                "sourcebucket": "src",
-                "endpoint": "minio.example:9000",
-                "credentials": {
-                    "accessKey": "ak",
-                    "secretKey": "sk",
-                    "sessionToken": "minio-session-token"
-                },
-                "targetbucket": "dst",
-                "type": "replication",
-                "replicationSync": true,
-                "bandwidthlimit": 107374182400i64,
-                "storageclass": "STANDARD",
-                "resetID": "reset-789",
-                "deploymentID": "deploy-123"
-            }]
-        }))
-        .expect("MinIO-written bucket-targets.json must deserialize");
-
-        let target = &targets.targets[0];
-        assert_eq!(target.bandwidth_limit, 107374182400);
-        assert_eq!(target.storage_class, "STANDARD");
-        assert_eq!(target.reset_id, "reset-789");
-        assert_eq!(target.deployment_id, "deploy-123");
-        assert_eq!(
-            target
-                .credentials
-                .as_ref()
-                .and_then(|credentials| credentials.session_token.as_deref()),
-            Some("minio-session-token")
-        );
-    }
-
    #[test]
    fn test_bucket_target_debug_redacts_credentials() {
        let target = BucketTarget {
@@ -784,6 +784,24 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
 ///
 /// # Returns
 /// A Result containing the BitrotWriterWrapper or an error
+/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
+///
+/// A known length is grown by one checksum per shard so the on-disk file size
+/// matches what the bitrot writer emits. A negative length is the
+/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
+/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
+/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
+/// a fixed body length when locating the authenticated trailer. Clamping it
+/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
+/// because a genuinely empty object still means an empty body.
+fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
+    if length <= 0 {
+        return length;
+    }
+    let length = length as usize;
+    (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
+}
+
 pub async fn create_bitrot_writer(
    is_inline_buffer: bool,
    disk: Option<&DiskStore>,
@@ -796,12 +814,7 @@ pub async fn create_bitrot_writer(
    let writer = if is_inline_buffer {
        CustomWriter::new_inline_buffer()
    } else if let Some(disk) = disk {
-        let length = if length > 0 {
-            let length = length as usize;
-            (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
-        } else {
-            0
-        };
+        let length = bitrot_create_file_size(length, shard_size, &checksum_algo);

        let file = disk.create_file("", volume, path, length).await?;
        #[cfg(feature = "hotpath")]
@@ -820,6 +833,25 @@ mod tests {
    use rustfs_rio::ChunkReader;
    use std::collections::VecDeque;

+    #[test]
+    fn bitrot_create_file_size_grows_known_length_by_checksums() {
+        // 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
+    }
+
+    #[test]
+    fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
+        assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
+        // SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
+        // put_file_stream receiver relies on a non-positive size to parse the auth
+        // trailer from the stream tail, so the sentinel must survive untouched.
+        assert_eq!(
+            bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
+            rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
+        );
+    }
+
    struct TestChunkReader {
        chunks: VecDeque<Bytes>,
    }
@@ -16,14 +16,14 @@
 //!
 //! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
 //! kills the active node while this test continuously decrypts through a
-//! surviving standby. KV2 and Transit requests must remain successful, use a
-//! bounded number of attempts, and leave the circuit and in-flight gauges at
-//! zero after a new leader is elected.
+//! surviving standby. KV2 and Transit must recover after the bounded circuit
+//! interval, use a bounded number of attempts, and leave the circuit and
+//! in-flight gauges at zero after a new leader is elected.

 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;

 use metrics_util::MetricKind;
@@ -43,6 +43,11 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
 const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
 const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
 const MAX_ATTEMPTS: u32 = 10;
+const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
+const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
+// The circuit remains open for 30s after five failed attempts.
+const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
+const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);

 type MetricEntry = (
    metrics_util::CompositeKey,
@@ -64,7 +69,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
        backend,
        backend_config,
        allow_insecure_dev_defaults: true,
-        timeout: Duration::from_secs(2),
+        timeout: ATTEMPT_TIMEOUT,
        retry_attempts: MAX_ATTEMPTS,
        enable_cache: false,
        ..KmsConfig::default()
@@ -164,14 +169,31 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
        .sum()
 }

-async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
-    tokio::time::timeout(Duration::from_secs(20), async {
+async fn wait_for_count(
+    counter: &AtomicU64,
+    failure: &Mutex<Option<String>>,
+    minimum: u64,
+    description: &str,
+    timeout: Duration,
+) {
+    tokio::time::timeout(timeout, async {
        while counter.load(Ordering::SeqCst) < minimum {
+            if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
+                panic!(
+                    "{description} worker failed after {} successful decrypts: {error}",
+                    counter.load(Ordering::SeqCst)
+                );
+            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
    })
    .await
-    .unwrap_or_else(|_| panic!("timed out waiting for {description}"));
+    .unwrap_or_else(|_| {
+        panic!(
+            "timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
+            counter.load(Ordering::SeqCst)
+        )
+    });
 }

 async fn wait_for_file(path: &Path, description: &str) {
@@ -189,7 +211,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
    request: DecryptRequest,
    expected: Vec<u8>,
    completed: Arc<AtomicU64>,
-    failed: Arc<AtomicBool>,
+    allow_failover_errors: Arc<AtomicBool>,
+    failure: Arc<Mutex<Option<String>>>,
    stop: CancellationToken,
 ) {
    while !stop.is_cancelled() {
@@ -197,8 +220,18 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
            Ok(response) if response.plaintext == expected => {
                completed.fetch_add(1, Ordering::SeqCst);
            }
-            Ok(_) | Err(_) => {
-                failed.store(true, Ordering::SeqCst);
+            Ok(_) => {
+                *failure.lock().expect("decrypt failure lock poisoned") =
+                    Some("decrypt returned unexpected plaintext".to_string());
+                return;
+            }
+            Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
+                if allow_failover_errors.load(Ordering::SeqCst) =>
+            {
+                tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
+            }
+            Err(error) => {
+                *failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
                return;
            }
        }
@@ -296,7 +329,9 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
    );

    let stop = CancellationToken::new();
-    let failed = Arc::new(AtomicBool::new(false));
+    let allow_failover_errors = Arc::new(AtomicBool::new(false));
+    let kv2_failure = Arc::new(Mutex::new(None));
+    let transit_failure = Arc::new(Mutex::new(None));
    let kv2_completed = Arc::new(AtomicU64::new(0));
    let transit_completed = Arc::new(AtomicU64::new(0));
    let kv2_worker = tokio::spawn(decrypt_loop(
@@ -304,7 +339,8 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
        kv2_request,
        kv2_data_key.plaintext_key,
        Arc::clone(&kv2_completed),
-        Arc::clone(&failed),
+        Arc::clone(&allow_failover_errors),
+        Arc::clone(&kv2_failure),
        stop.clone(),
    ));
    let transit_worker = tokio::spawn(decrypt_loop(
@@ -312,12 +348,21 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
        transit_request,
        transit_data_key.plaintext_key,
        Arc::clone(&transit_completed),
-        Arc::clone(&failed),
+        Arc::clone(&allow_failover_errors),
+        Arc::clone(&transit_failure),
        stop.clone(),
    ));

-    wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
-    wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
+    wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
+    wait_for_count(
+        &transit_completed,
+        &transit_failure,
+        2,
+        "two healthy Transit decrypts",
+        HEALTHY_PROGRESS_TIMEOUT,
+    )
+    .await;
+    allow_failover_errors.store(true, Ordering::SeqCst);
    std::fs::write(&marker, b"ready").expect("publish failover readiness marker");

    wait_for_file(&elected, "the replacement Vault leader").await;
@@ -326,18 +371,39 @@ async fn exercise_failover(snapshotter: &Snapshotter) {

    let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
    let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
-    wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
-    wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
+    wait_for_count(
+        &kv2_completed,
+        &kv2_failure,
+        kv2_after_election,
+        "post-failover KV2 decrypts",
+        POST_FAILOVER_PROGRESS_TIMEOUT,
+    )
+    .await;
+    wait_for_count(
+        &transit_completed,
+        &transit_failure,
+        transit_after_election,
+        "post-failover Transit decrypts",
+        POST_FAILOVER_PROGRESS_TIMEOUT,
+    )
+    .await;

    stop.cancel();
    kv2_worker.await.expect("KV2 decrypt worker must join");
    transit_worker.await.expect("Transit decrypt worker must join");
-    assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
+    assert!(
+        kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
+        "no KV2 decrypt may fail or return different plaintext"
+    );
+    assert!(
+        transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
+        "no Transit decrypt may fail or return different plaintext"
+    );
 }

 #[test]
 #[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
-fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
+fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
    let recorder = DebuggingRecorder::new();
    let snapshotter = recorder.snapshotter();
    metrics::with_local_recorder(&recorder, || {
@@ -349,11 +415,6 @@ fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
    });
    let snapshot = snapshotter.snapshot().into_vec();

-    assert_eq!(
-        counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
-        0,
-        "a bounded leader election must not open the circuit"
-    );
    assert_eq!(
        counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
        0,
@@ -52,7 +52,7 @@
 | fault_proxy | 7 |  |
 | get_codec_streaming_compat_test | 1 |  |
 | get_stream_failure_observability_test | 1 |  |
-| group_delete_test | 1 |  |
+| group_delete_test | 4 |  |
 | head_object_consistency_test | 1 | ✅ |
 | head_object_range_test | 1 | ✅ |
 | heal_erasure_disk_rebuild_test | 4 | 🌙 |
@@ -99,4 +99,4 @@
 | tls_hot_reload_test | 1 | ✅ |
 | version_id_regression_test | 10 | ✅ |

-**Total listed: 575 tests across 82 modules · PR smoke: 163 tests / 36 modules · merge/main full: 453 tests / 73 modules · nightly replication: 55 tests · nightly cluster faults: 28 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-23.
+**Total listed: 578 tests across 82 modules · PR smoke: 163 tests / 36 modules · merge/main full: 456 tests / 73 modules · nightly replication: 55 tests · nightly cluster faults: 28 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-23.
@@ -376,10 +376,7 @@ impl RemoteTargetRequest {
 /// Admin-response encoding of a remote target: the persisted bucket-targets
 /// format keeps `healthCheckDuration`/`totalDowntime` in seconds and the
 /// `latency` stats in milliseconds, but madmin decodes all of them as Go
-/// `time.Duration` (nanoseconds) — and it looks the fields up under its own
-/// JSON tags (`bandwidthlimit`, `storageclass`, `resetID`, `deploymentID`,
-/// `credentials.sessionToken` — madmin-go v3.0.109 `bucket-targets.go`), not
-/// the persisted snake_case keys. Re-encode just those fields here without
+/// `time.Duration` (nanoseconds) — re-encode just those fields without
 /// touching the persistence wire format.
 fn remote_target_admin_json(target: &BucketTarget) -> Result<serde_json::Value, serde_json::Error> {
    fn go_duration_nanos(duration: Duration) -> serde_json::Value {
@@ -388,12 +385,6 @@ fn remote_target_admin_json(target: &BucketTarget) -> Result<serde_json::Value,
        u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX).into()
    }

-    fn rename_key(value: &mut serde_json::Value, from: &str, to: &str) {
-        if let Some(moved) = value.as_object_mut().and_then(|object| object.remove(from)) {
-            value[to] = moved;
-        }
-    }
-
    let mut value = serde_json::to_value(target)?;
    value["healthCheckDuration"] = go_duration_nanos(target.health_check_duration);
    value["totalDowntime"] = go_duration_nanos(target.total_downtime);
@@ -402,11 +393,6 @@ fn remote_target_admin_json(target: &BucketTarget) -> Result<serde_json::Value,
        "avg": go_duration_nanos(target.latency.avg),
        "max": go_duration_nanos(target.latency.max),
    });
-    rename_key(&mut value, "bandwidth_limit", "bandwidthlimit");
-    rename_key(&mut value, "storage_class", "storageclass");
-    rename_key(&mut value, "reset_id", "resetID");
-    rename_key(&mut value, "deployment_id", "deploymentID");
-    rename_key(&mut value["credentials"], "session_token", "sessionToken");
    Ok(value)
 }

@@ -1490,7 +1476,7 @@ mod tests {
        parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff, unique_replication_peers,
        validate_remote_target_tls_settings,
    };
-    use crate::admin::storage_api::bucket::target::{BucketTarget, Credentials as TargetCredentials, LatencyStat};
+    use crate::admin::storage_api::bucket::target::{BucketTarget, LatencyStat};
    use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry};
    use http::Uri;

@@ -2286,129 +2272,6 @@ mod tests {
        assert_eq!(persisted["latency"]["max"], 250);
    }

-    #[test]
-    fn list_remote_targets_response_uses_madmin_key_names() {
-        // madmin-go v3.0.109 BucketTarget JSON tags are `bandwidthlimit`,
-        // `storageclass`, `resetID`, `deploymentID`, and
-        // `credentials.sessionToken` (backlog#1951); the persisted snake_case
-        // keys decode to zero values in mc, blanking the bandwidth and
-        // reset-id columns of `mc replicate ls`.
-        let target = BucketTarget {
-            endpoint: "192.168.1.10:9000".to_string(),
-            target_bucket: "target".to_string(),
-            credentials: Some(TargetCredentials {
-                access_key: "access".to_string(),
-                secret_key: String::new(),
-                session_token: Some("session-token".to_string()),
-                expiration: None,
-            }),
-            bandwidth_limit: 107_374_182_400,
-            storage_class: "STANDARD".to_string(),
-            reset_id: "reset-123".to_string(),
-            deployment_id: "deploy-456".to_string(),
-            ..Default::default()
-        };
-
-        let value = super::remote_target_admin_json(&target).expect("admin response should serialize");
-
-        assert_eq!(value["bandwidthlimit"], 107_374_182_400i64);
-        assert_eq!(value["storageclass"], "STANDARD");
-        assert_eq!(value["resetID"], "reset-123");
-        assert_eq!(value["deploymentID"], "deploy-456");
-        assert_eq!(value["credentials"]["sessionToken"], "session-token");
-        // The madmin keys replace the snake_case ones rather than duplicating
-        // them next to each other.
-        for stale in ["bandwidth_limit", "bandwidth", "storage_class", "reset_id", "deployment_id"] {
-            assert!(value.get(stale).is_none(), "admin response must not carry `{stale}`");
-        }
-        assert!(value["credentials"].get("session_token").is_none());
-    }
-
-    /// Decode-side mirror of madmin-go v3.0.109 `BucketTarget`/`Credentials`
-    /// (`bucket-targets.go`): the exact `json:"..."` tags mc's `encoding/json`
-    /// looks fields up under. Unknown keys are ignored like Go does, and a
-    /// missing key leaves the Go zero value, which is exactly how a misnamed
-    /// key turns into a blank column in `mc replicate ls`.
-    #[derive(Debug, Default, serde::Deserialize)]
-    #[serde(default)]
-    struct MadminBucketTarget {
-        sourcebucket: String,
-        endpoint: String,
-        credentials: Option<MadminCredentials>,
-        targetbucket: String,
-        arn: String,
-        bandwidthlimit: i64,
-        #[serde(rename = "replicationSync")]
-        replication_sync: bool,
-        storageclass: String,
-        #[serde(rename = "healthCheckDuration")]
-        health_check_duration: i64,
-        #[serde(rename = "resetID")]
-        reset_id: String,
-        #[serde(rename = "totalDowntime")]
-        total_downtime: i64,
-        #[serde(rename = "deploymentID")]
-        deployment_id: String,
-    }
-
-    #[derive(Debug, Default, serde::Deserialize)]
-    #[serde(default)]
-    struct MadminCredentials {
-        #[serde(rename = "accessKey")]
-        access_key: String,
-        #[serde(rename = "secretKey")]
-        secret_key: String,
-        #[serde(rename = "sessionToken")]
-        session_token: String,
-    }
-
-    #[test]
-    fn list_remote_targets_response_decodes_through_madmin_tags() {
-        // Regression for the review on backlog#1951: the response must decode
-        // a nonzero bandwidth limit through madmin's `bandwidthlimit` tag (not
-        // `bandwidth`, which Go would silently drop as an unknown key).
-        let target = BucketTarget {
-            source_bucket: "src".to_string(),
-            endpoint: "192.168.1.10:9000".to_string(),
-            target_bucket: "target".to_string(),
-            arn: "arn:rustfs:replication:us-east-1:dep:target".to_string(),
-            credentials: Some(TargetCredentials {
-                access_key: "access".to_string(),
-                secret_key: String::new(),
-                session_token: Some("session-token".to_string()),
-                expiration: None,
-            }),
-            bandwidth_limit: 1_073_741_824,
-            replication_sync: true,
-            storage_class: "STANDARD".to_string(),
-            health_check_duration: std::time::Duration::from_secs(60),
-            reset_id: "reset-123".to_string(),
-            total_downtime: std::time::Duration::from_secs(90),
-            deployment_id: "deploy-456".to_string(),
-            ..Default::default()
-        };
-
-        let wire = serde_json::to_string(&super::remote_target_admin_json(&target).expect("admin response should serialize"))
-            .expect("admin response should encode");
-        let decoded: MadminBucketTarget = serde_json::from_str(&wire).expect("madmin-shaped decode must succeed");
-
-        assert_eq!(decoded.bandwidthlimit, 1_073_741_824, "mc must see the nonzero bandwidth limit");
-        assert_eq!(decoded.sourcebucket, "src");
-        assert_eq!(decoded.endpoint, "192.168.1.10:9000");
-        assert_eq!(decoded.targetbucket, "target");
-        assert_eq!(decoded.arn, "arn:rustfs:replication:us-east-1:dep:target");
-        assert!(decoded.replication_sync);
-        assert_eq!(decoded.storageclass, "STANDARD");
-        assert_eq!(decoded.health_check_duration, 60_000_000_000);
-        assert_eq!(decoded.reset_id, "reset-123");
-        assert_eq!(decoded.total_downtime, 90_000_000_000);
-        assert_eq!(decoded.deployment_id, "deploy-456");
-        let credentials = decoded.credentials.expect("credentials must decode");
-        assert_eq!(credentials.access_key, "access");
-        assert_eq!(credentials.secret_key, "");
-        assert_eq!(credentials.session_token, "session-token");
-    }
-
    #[test]
    fn remote_target_admin_json_latency_round_trips_through_go_duration() {
        // Round trip: a madmin reader decodes the latency values as Go
@@ -241,7 +241,7 @@ env \
  RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
  RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
  cargo test -p rustfs-kms --test vault_ha_failover_live \
-    vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
+    vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
    --ignored --nocapture --test-threads=1 &
 TEST_PID=$!
Author	SHA1	Message	Date
houseme	566877d3ba	Merge branch 'main' into overtrue/activate-group-e2e	2026-08-23 12:37:24 +08:00
Zhengchao An	23a2c7d776	test(kms): stabilize Vault failover validation (#6385 ) * test(kms): bound Vault failover progress wait * ci(nightly): honor manual dispatch ref * test(kms): preserve Vault worker failures * test(kms): validate Vault circuit recovery	2026-08-23 12:32:11 +08:00
唐小鸭	5f72209446	fix(ecstore): keep unknown-size sentinel in create_bitrot_writer (#6380 ) SSE and compression wrap the payload so its length is unknown and advertise HashReader::SIZE_PRESERVE_LAYER (-1). Every layer preserved that sentinel except create_bitrot_writer, which clamped it to 0 before calling DiskAPI::create_file. RemoteDisk forwards that size verbatim in the put_file_stream query, so remote peers were told the body was empty. Since the authenticated put-file trailer (#5868) the receiver used the declared size to split body from trailer, turning the clamp into a fatal "auth trailer has trailing data" failure for every SSE PUT on multi-node deployments (rc.2). #6320 relaxed the receiver to only trust size > 0; this change fixes the sender so the sentinel survives end to end and the wire no longer conflates empty objects with unknown-length streams. Refs #6331	2026-08-23 12:29:52 +08:00
overtrue	3294c64fcc	test(e2e): correct group regression fixtures	2026-08-23 11:11:01 +08:00
overtrue	a8be0f81f5	test(e2e): bind group selection to Linux listing	2026-08-23 08:55:07 +08:00
overtrue	3179b7acb8	test(e2e): pin group deletion errors	2026-08-23 06:42:55 +08:00
overtrue	d25b84a793	Merge remote-tracking branch 'origin/main' into overtrue/activate-group-e2e	2026-08-23 05:43:17 +08:00
overtrue	a79b806fb4	test(e2e): activate group management regressions	2026-08-23 02:37:15 +08:00