diff --git a/.config/e2e-repl-nightly-selection.txt b/.config/e2e-repl-nightly-selection.txt index d78fd611b..46bcb0862 100644 --- a/.config/e2e-repl-nightly-selection.txt +++ b/.config/e2e-repl-nightly-selection.txt @@ -1 +1 @@ -sha256=655a3f3c1d042e694339d15caba7580518320322d1bac0f09450b37e6c09e2e7 +sha256=8d5517f5f2fc32d561782dfccd51b7f746f5e25b2835e37e100c883f7f18777d diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 20927a55f..01f462c84 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -1912,6 +1912,21 @@ async fn site_replication_info(env: &RustFSTestEnvironment) -> Result Result> { + 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 Result<(), Box> { + 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> { init_logging(); diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 796e3804e..6af0b6cf1 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -1637,7 +1637,17 @@ fn reconcile_site_replication_wiring() -> std::pin::Pin { - 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) -> Vec { + 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, 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> { 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; diff --git a/scripts/test/site_replication_smoke.py b/scripts/test/site_replication_smoke.py index 259a98e79..bbfb83fd6 100755 --- a/scripts/test/site_replication_smoke.py +++ b/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":