fix(site-replication): rotate-svc-acct no longer wedges replication (#6793)

.config/e2e-repl-nightly-selection.txt

@@ -1 +1 @@
-sha256=655a3f3c1d042e694339d15caba7580518320322d1bac0f09450b37e6c09e2e7
+sha256=8d5517f5f2fc32d561782dfccd51b7f746f5e25b2835e37e100c883f7f18777d

crates/e2e_test/src/replication_extension_test.rs

@@ -1912,6 +1912,21 @@ async fn site_replication_info(env: &RustFSTestEnvironment) -> Result<SiteReplic
     Ok(serde_json::from_slice(&response.bytes().await?)?)
 }
 
+async fn site_replication_rotate_svc_acct(
+    env: &RustFSTestEnvironment,
+) -> Result<ReplicateEditStatus, Box<dyn Error + Send + Sync>> {
+    let url = format!("{}/rustfs/admin/v3/site-replication/rotate-svc-acct", env.url);
+    let response = signed_request(http::Method::POST, &url, &env.access_key, &env.secret_key, None, None).await?;
+
+    if response.status() != StatusCode::OK {
+        let status = response.status();
+        let body = response.text().await.unwrap_or_default();
+        return Err(format!("site replication rotate-svc-acct failed: {status} {body}").into());
+    }
+
+    Ok(serde_json::from_slice(&response.bytes().await?)?)
+}
+
 async fn site_replication_resync_op(
     env: &RustFSTestEnvironment,
     operation: &str,
@@ -6310,6 +6325,112 @@ async fn test_site_replication_remove_all_real_dual_node() -> Result<(), Box<dyn
     Ok(())
 }
 
+#[tokio::test]
+async fn test_site_replication_rotate_svc_acct_completes_and_replication_survives_real_dual_node()
+-> Result<(), Box<dyn Error + Send + Sync>> {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    source_env
+        .start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
+        .await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env
+        .start_rustfs_server_without_cleanup_with_env(LOOPBACK_REPLICATION_TARGET_ENV)
+        .await?;
+
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+    let bucket = "site-repl-rotate-svc-acct";
+
+    let add_status = site_replication_add(
+        &source_env,
+        &[
+            PeerSite {
+                name: "source-site".to_string(),
+                endpoint: source_env.url.clone(),
+                access_key: source_env.access_key.clone(),
+                secret_key: source_env.secret_key.clone(),
+                ..Default::default()
+            },
+            PeerSite {
+                name: "target-site".to_string(),
+                endpoint: target_env.url.clone(),
+                access_key: target_env.access_key.clone(),
+                secret_key: target_env.secret_key.clone(),
+                ..Default::default()
+            },
+        ],
+    )
+    .await?;
+    assert!(add_status.success, "unexpected site add result: {add_status:?}");
+
+    let _source_info = wait_for_site_replication_enabled(&source_env, 2).await?;
+    let _target_info = wait_for_site_replication_enabled(&target_env, 2).await?;
+
+    source_client.create_bucket().bucket(bucket).send().await?;
+    enable_bucket_versioning(&source_env, bucket).await?;
+    wait_for_bucket_on_target(&target_client, bucket).await?;
+    let baseline_payload = b"before rotation".to_vec();
+    source_client
+        .put_object()
+        .bucket(bucket)
+        .key("before-rotate.txt")
+        .body(ByteStream::from(baseline_payload.clone()))
+        .send()
+        .await?;
+    let replicated_baseline = wait_for_object_on_target(&target_client, bucket, "before-rotate.txt").await?;
+    assert_eq!(replicated_baseline, baseline_payload);
+
+    // A single rotation call must finish the whole hand-over. Before the fix
+    // the join push could only sign with the freshly installed secret, every
+    // peer rejected it, the rotation stayed pending forever, and both
+    // replication directions were dead until an operator retried.
+    let rotate_status = site_replication_rotate_svc_acct(&source_env).await?;
+    assert!(rotate_status.success, "rotation did not complete in one call: {rotate_status:?}");
+
+    for env in [&source_env, &target_env] {
+        let deadline = std::time::Instant::now() + Duration::from_secs(30);
+        loop {
+            let info = site_replication_info(env).await?;
+            if info.enabled && info.pending_operation.is_none() {
+                break;
+            }
+            if std::time::Instant::now() > deadline {
+                return Err(format!("rotation left {} with a pending operation: {:?}", env.url, info.pending_operation).into());
+            }
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+    }
+
+    // Replication must actually flow again in both directions with the
+    // rotated service-account secret.
+    let forward_payload = b"after rotation from source".to_vec();
+    source_client
+        .put_object()
+        .bucket(bucket)
+        .key("after-rotate-forward.txt")
+        .body(ByteStream::from(forward_payload.clone()))
+        .send()
+        .await?;
+    let replicated_forward = wait_for_object_on_target(&target_client, bucket, "after-rotate-forward.txt").await?;
+    assert_eq!(replicated_forward, forward_payload);
+
+    let reverse_payload = b"after rotation from target".to_vec();
+    target_client
+        .put_object()
+        .bucket(bucket)
+        .key("after-rotate-reverse.txt")
+        .body(ByteStream::from(reverse_payload.clone()))
+        .send()
+        .await?;
+    let replicated_reverse = wait_for_object_on_target(&source_client, bucket, "after-rotate-reverse.txt").await?;
+    assert_eq!(replicated_reverse, reverse_payload);
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_site_replication_state_edit_fresh_and_stale_real_dual_node() -> Result<(), Box<dyn Error + Send + Sync>> {
     init_logging();

rustfs/src/admin/handlers/site_replication.rs

@@ -1637,7 +1637,17 @@ fn reconcile_site_replication_wiring() -> std::pin::Pin<Box<dyn std::future::Fut
 
         match load_site_replication_state().await {
             Ok(state) => {
-                if state.pending_endpoint_refresh.is_some() || state.pending_rotation.is_some() {
+                if state.pending_endpoint_refresh.is_some() {
+                    return;
+                }
+                // A wedged rotation is worse than a wedged removal: the local
+                // secret is already switched, so until the join lands every
+                // outbound peer push signs with a secret the peers reject and
+                // every inbound peer request carries a secret this site
+                // rejects — both replication directions are dead. Nothing
+                // else re-drives it; push it forward like the removal below.
+                if let Some(pending_rotation) = state.pending_rotation.clone() {
+                    resume_pending_rotation(&state, &pending_rotation).await;
                     return;
                 }
                 // A removal whose peers were unreachable is the one pending
@@ -3518,6 +3528,143 @@ async fn finalize_pending_rotation_if_complete(rotation_id: &str, local_peer: &P
     .await
 }
 
+/// The candidate secrets a rotation push may sign with, in trial order: the
+/// persisted candidates recorded by earlier attempts, then the currently
+/// installed secret (the pre-rotation one on a fresh drive — peers still hold
+/// it, so it must come before the new secret), then the new secret itself.
+fn rotation_secret_candidates(pending: &PendingRotation, current_secret: Option<String>) -> Vec<String> {
+    let mut candidates = pending.secret_candidates.clone();
+    if let Some(current_secret) = current_secret {
+        push_unique_secret_candidate(&mut candidates, current_secret);
+    }
+    push_unique_secret_candidate(&mut candidates, pending.new_secret_key.clone());
+    candidates
+}
+
+/// Push a half-finished service-account rotation one step forward: install the
+/// new secret locally (idempotent), send the rotation join to every peer that
+/// has not acked yet, then finalize if that completed the set. Returns the
+/// per-peer failures and whether the rotation is now finished.
+///
+/// Shared by the operator-driven `SRRotateServiceAccountHandler` and the
+/// reconcile tick. The tick is what makes this self-healing: a rotation whose
+/// join push failed used to sit in `pending_rotation` forever while the local
+/// secret was already switched — so both replication directions stayed dead
+/// (outbound signed with a secret the peers reject, inbound rejecting the
+/// secret the peers still sign with) until an operator re-ran the rotation.
+///
+/// Callers must hold the lifecycle guard.
+async fn drive_pending_rotation(pending: &PendingRotation, local_peer: &PeerInfo) -> S3Result<(Vec<String>, bool)> {
+    // Capture the still-installed secret BEFORE the overwrite below: on the
+    // first drive of a fresh rotation this is the secret the peers hold, and
+    // it must be able to sign the join push — with only the new secret as a
+    // candidate every peer rejects the push and the rotation wedges.
+    let current_secret = site_replicator_service_account_secret(&pending.access_key).await.ok();
+    if let Some(current_secret) = current_secret.clone() {
+        record_pending_rotation_secret_candidate(&pending.id, current_secret).await?;
+    }
+    let secret_candidates = rotation_secret_candidates(pending, current_secret);
+
+    set_site_replicator_service_account_secret(&pending.parent, pending.new_secret_key.clone()).await?;
+    refresh_bucket_targets_after_service_account_rotation().await;
+
+    let join_req = SRPeerJoinReq {
+        svc_acct_access_key: pending.access_key.clone(),
+        svc_acct_secret_key: pending.new_secret_key.clone(),
+        svc_acct_parent: pending.parent.clone(),
+        peers: pending.peers.clone(),
+        updated_at: pending.updated_at,
+    };
+
+    let mut peer_errors = Vec::new();
+    for peer in pending.peers.values() {
+        if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
+            || pending.acked_deployment_ids.contains(&peer.deployment_id)
+        {
+            continue;
+        }
+        // A superseded join returns BEFORE `apply_iam`, so a no-op answer
+        // means the peer never installed the new secret. Acking it would
+        // finalize a rotation half the mesh cannot authenticate against
+        // (rustfs/rustfs#5963).
+        let rotation_error =
+            match PeerAdminRequest::put(&runtime_peer_connection(peer)?, SITE_REPLICATION_PEER_JOIN_PATH, &pending.access_key)
+                .send_with_secret_candidates(&secret_candidates, &join_req)
+                .await
+            {
+                Err(err) => Some(summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint))),
+                Ok(body) => match parse_peer_join_response(&body, peer.clone()) {
+                    Ok(response) if response.applied == Some(false) => Some(summarize_peer_error_detail(&format!(
+                        "{}: peer did not apply the rotation join (its site replication state is newer than the snapshot it \
+                     was sent); the new service account secret was not installed",
+                        peer.endpoint
+                    ))),
+                    // Unparseable bodies keep the pre-existing behaviour: the
+                    // transport succeeded, and MinIO peers answer with an empty
+                    // body this helper already tolerates.
+                    Ok(_) | Err(_) => None,
+                },
+            };
+        if let Some(detail) = rotation_error {
+            warn!(
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                peer = %peer.endpoint,
+                result = "service_account_rotation_failed",
+                error = %detail,
+                "admin site replication state"
+            );
+            peer_errors.push(detail);
+        } else {
+            mark_pending_rotation_peer_acked(&pending.id, &peer.deployment_id).await?;
+        }
+    }
+
+    let complete = finalize_pending_rotation_if_complete(&pending.id, local_peer).await?;
+    Ok((peer_errors, complete))
+}
+
+/// The reconcile tick's half of [`drive_pending_rotation`]: resume the rotation
+/// this site could not finish, and report the outcome. Runs under the tick's
+/// lifecycle guard, which is what keeps it from racing an operator re-running
+/// the rotation (that handler takes the same guard).
+async fn resume_pending_rotation(state: &SiteReplicationState, pending: &PendingRotation) {
+    let local_peer = current_local_runtime_peer(state);
+    match drive_pending_rotation(pending, &local_peer).await {
+        Ok((peer_errors, complete)) => {
+            if complete && peer_errors.is_empty() {
+                info!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "pending_rotation_resumed",
+                    "admin site replication state"
+                );
+            } else {
+                warn!(
+                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                    result = "pending_rotation_still_pending",
+                    error_count = peer_errors.len(),
+                    "admin site replication state"
+                );
+            }
+        }
+        Err(err) => {
+            warn!(
+                event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+                component = LOG_COMPONENT_ADMIN,
+                subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+                result = "pending_rotation_resume_failed",
+                error = ?err,
+                "admin site replication state"
+            );
+        }
+    }
+}
+
 async fn pending_remove_ready_to_finalize(remove_id: &str, local_peer: &PeerInfo) -> S3Result<Option<PendingRemove>> {
     let state = load_site_replication_state().await?;
     let Some(pending) = state.pending_remove.as_ref() else {
@@ -7211,79 +7358,21 @@ impl Operation for SRRotateServiceAccountHandler {
         })
         .await?;
 
+        // Record the pre-rotation secret before drive_pending_rotation
+        // overwrites the local one: the join push must still be able to sign
+        // with the secret the peers hold, and the persisted candidate is also
+        // what lets a later resume recover a rotation this attempt could not
+        // finish. Push it into the local copy too — the persisted record alone
+        // is invisible to the snapshot this call already holds.
+        let mut pending_rotation = pending_rotation;
         if !previous_access_key.is_empty()
             && let Ok(previous_iam_secret) = site_replicator_service_account_secret(&previous_access_key).await
         {
-            record_pending_rotation_secret_candidate(&pending_rotation.id, previous_iam_secret).await?;
+            record_pending_rotation_secret_candidate(&pending_rotation.id, previous_iam_secret.clone()).await?;
+            push_unique_secret_candidate(&mut pending_rotation.secret_candidates, previous_iam_secret);
         }
 
-        set_site_replicator_service_account_secret(&pending_rotation.parent, pending_rotation.new_secret_key.clone()).await?;
-
-        refresh_bucket_targets_after_service_account_rotation().await;
-
-        let mut secret_candidates = pending_rotation.secret_candidates.clone();
-        if let Ok(current_secret) = site_replicator_service_account_secret(&pending_rotation.access_key).await {
-            push_unique_secret_candidate(&mut secret_candidates, current_secret);
-        }
-        push_unique_secret_candidate(&mut secret_candidates, pending_rotation.new_secret_key.clone());
-
-        let join_req = SRPeerJoinReq {
-            svc_acct_access_key: pending_rotation.access_key.clone(),
-            svc_acct_secret_key: pending_rotation.new_secret_key.clone(),
-            svc_acct_parent: pending_rotation.parent.clone(),
-            peers: pending_rotation.peers.clone(),
-            updated_at: pending_rotation.updated_at,
-        };
-
-        let mut peer_errors = Vec::new();
-        for peer in pending_rotation.peers.values() {
-            if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
-                || pending_rotation.acked_deployment_ids.contains(&peer.deployment_id)
-            {
-                continue;
-            }
-            // A superseded join returns BEFORE `apply_iam`, so a no-op answer
-            // means the peer never installed the new secret. Acking it would
-            // finalize a rotation half the mesh cannot authenticate against
-            // (rustfs/rustfs#5963).
-            let rotation_error = match PeerAdminRequest::put(
-                &runtime_peer_connection(peer)?,
-                SITE_REPLICATION_PEER_JOIN_PATH,
-                &pending_rotation.access_key,
-            )
-            .send_with_secret_candidates(&secret_candidates, &join_req)
-            .await
-            {
-                Err(err) => Some(summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint))),
-                Ok(body) => match parse_peer_join_response(&body, peer.clone()) {
-                    Ok(response) if response.applied == Some(false) => Some(summarize_peer_error_detail(&format!(
-                        "{}: peer did not apply the rotation join (its site replication state is newer than the snapshot it \
-                         was sent); the new service account secret was not installed",
-                        peer.endpoint
-                    ))),
-                    // Unparseable bodies keep the pre-existing behaviour: the
-                    // transport succeeded, and MinIO peers answer with an empty
-                    // body this helper already tolerates.
-                    Ok(_) | Err(_) => None,
-                },
-            };
-            if let Some(detail) = rotation_error {
-                warn!(
-                    event = EVENT_ADMIN_SITE_REPLICATION_STATE,
-                    component = LOG_COMPONENT_ADMIN,
-                    subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
-                    peer = %peer.endpoint,
-                    result = "service_account_rotation_failed",
-                    error = %detail,
-                    "admin site replication state"
-                );
-                peer_errors.push(detail);
-            } else {
-                mark_pending_rotation_peer_acked(&pending_rotation.id, &peer.deployment_id).await?;
-            }
-        }
-
-        let complete = finalize_pending_rotation_if_complete(&pending_rotation.id, &local_peer).await?;
+        let (mut peer_errors, complete) = drive_pending_rotation(&pending_rotation, &local_peer).await?;
         if !complete && peer_errors.is_empty() {
             peer_errors.push("service account rotation is still pending".to_string());
         }
@@ -7309,6 +7398,36 @@ impl Operation for SRRotateServiceAccountHandler {
 mod tests {
     use super::*;
     use crate::site_replication::identity::deployment_id_for_endpoint;
+
+    #[test]
+    fn test_rotation_secret_candidates_try_the_installed_secret_before_the_new_one() {
+        let mut pending = PendingRotation {
+            id: "rotation-id".to_string(),
+            access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+            parent: "root".to_string(),
+            new_secret_key: "new-secret".to_string(),
+            ..Default::default()
+        };
+
+        // Fresh rotation: nothing persisted yet, the installed secret is still
+        // the pre-rotation one the peers hold — it must be tried first.
+        assert_eq!(
+            rotation_secret_candidates(&pending, Some("old-secret".to_string())),
+            ["old-secret", "new-secret"]
+        );
+
+        // Resume after a failed first push: the old secret was persisted by
+        // that attempt, and the installed secret is already the new one.
+        pending.secret_candidates = vec!["old-secret".to_string()];
+        assert_eq!(
+            rotation_secret_candidates(&pending, Some("new-secret".to_string())),
+            ["old-secret", "new-secret"]
+        );
+
+        // An unreadable installed secret must still leave the push a candidate.
+        pending.secret_candidates = Vec::new();
+        assert_eq!(rotation_secret_candidates(&pending, None), ["new-secret"]);
+    }
     use axum::{Router, extract::State, routing::any};
     use base64_simd::STANDARD as BASE64_STANDARD;
     use http::Uri;

scripts/test/site_replication_smoke.py

@@ -27,6 +27,7 @@ Usage:
     ./scripts/test/site_replication_smoke.py            # up: start both + pair
     ./scripts/test/site_replication_smoke.py status     # process + pair status
     ./scripts/test/site_replication_smoke.py smoke      # bidirectional object check
+    ./scripts/test/site_replication_smoke.py rotate     # svc-account rotation regression
     ./scripts/test/site_replication_smoke.py diverge    # rustfs/rustfs#5963 regression
     ./scripts/test/site_replication_smoke.py logs       # tail both server logs
     ./scripts/test/site_replication_smoke.py down       # stop both processes
@@ -399,6 +400,42 @@ def smoke(site_a: Site, site_b: Site, timeout: float) -> None:
     print(f"[ok] bidirectional replication verified via bucket {bucket}")
 
 
+# ---------------------------------------------------------------------------
+# Service-account rotation regression
+# ---------------------------------------------------------------------------
+
+
+def rotate(site_a: Site, site_b: Site, timeout: float) -> None:
+    """Rotate the site-replicator service account and assert nothing wedges.
+
+    A wedged rotation is the nastiest failure mode this lab can produce: the
+    rotating site installs the new secret locally before the peers learn it,
+    so a failed join push kills BOTH replication directions at once. A single
+    rotate call must therefore complete the hand-over, clear pendingOperation
+    on every site, and leave replication flowing.
+    """
+    ensure_pair(site_a, site_b)
+
+    status, body = admin(site_a, "POST", "site-replication/rotate-svc-acct")
+    if status != 200:
+        raise SystemExit(f"[fail] rotate-svc-acct: HTTP {status} {body.decode(errors='replace')}")
+    result = json.loads(body)
+    print(f"[ok] rotate accepted: {json.dumps(result)}")
+    if not result.get("success", False):
+        raise SystemExit(f"[fail] rotation did not complete in one call: {json.dumps(result)}")
+
+    for site in (site_a, site_b):
+        wait_for(
+            f"{site.name} still reports a pending operation after the rotation",
+            lambda site=site: not pair_state(site).get("pendingOperation"),
+            timeout,
+        )
+    print("[ok] rotation settled on both sites")
+
+    smoke(site_a, site_b, timeout)
+    print("[ok] rotation regression passed")
+
+
 # ---------------------------------------------------------------------------
 # Divergence regression (rustfs/rustfs#5963)
 # ---------------------------------------------------------------------------
@@ -572,7 +609,7 @@ def main() -> None:
         "command",
         nargs="?",
         default="up",
-        choices=["up", "down", "restart", "status", "logs", "smoke", "diverge", "info", "remove", "clean"],
+        choices=["up", "down", "restart", "status", "logs", "smoke", "rotate", "diverge", "info", "remove", "clean"],
     )
     parser.add_argument("--port-a", type=int, default=9000, help="site A S3 port (default: %(default)s)")
     parser.add_argument("--port-b", type=int, default=9020, help="site B S3 port (default: %(default)s)")
@@ -603,6 +640,8 @@ def main() -> None:
         cmd_logs(sites, args.lines)
     elif args.command == "smoke":
         smoke(site_a, site_b, args.timeout)
+    elif args.command == "rotate":
+        rotate(site_a, site_b, args.timeout)
     elif args.command == "diverge":
         diverge(site_a, site_b, args.binary, args.console, args.timeout)
     elif args.command == "info":