fix(admin): bound site replication lifecycle lock and parallelize add preflight

The site replication add preflight probed peer sites serially while
holding the process-wide lifecycle lock, so k unreachable sites held the
lock for k peer-request timeouts, and every concurrent
add/remove/refresh waited on an unbounded lock acquire for the whole
time. Probe all sites concurrently (matching the file's other peer
fan-outs) so k unreachable sites cost roughly one timeout, and bound the
lifecycle lock acquire at 30s, returning a retryable 503 to waiters
instead of hanging indefinitely.

Regression tests pin the preflight fan-out concurrency, the bounded
acquire's 503, and the 10s/3s peer client timeout constants.

Refs rustfs/backlog#1952, rustfs/backlog#1946, rustfs/backlog#1889

crates/ecstore/src/bucket/target/bucket_target.rs

@@ -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 {

rustfs/src/admin/handlers/replication.rs

@@ -374,10 +374,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 {
@@ -386,12 +383,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);
@@ -400,11 +391,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)
 }
 
@@ -1487,7 +1473,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;
 
@@ -2282,129 +2268,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

rustfs/src/admin/handlers/site_replication.rs

@@ -137,6 +137,11 @@ const SITE_REPL_RESYNC_DEFAULT_PAGE_SIZE: usize = 100;
 const SITE_REPL_RESYNC_MAX_PAGE_SIZE: usize = 1000;
 const SITE_REPLICATION_PEER_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
 const SITE_REPLICATION_PEER_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);
+/// Bound on waiting for the lifecycle lock (below). 3x the peer request
+/// timeout: outlives one full peer round of a healthy concurrent lifecycle
+/// operation, while converting a holder wedged on unreachable peers into a
+/// retryable 503 for the waiter instead of an unbounded hang.
+const SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
 const SITE_REPLICATION_PEER_ERROR_DETAIL_LIMIT: usize = 256;
 const SITE_REPLICATION_INITIAL_SYNC_ERROR_LIMIT: usize = 32;
 const MAX_PEER_CA_CERT_PEM_SIZE: usize = 256 * 1024;
@@ -387,9 +392,17 @@ struct SiteReplicationLifecycleGuard {
 }
 
 impl SiteReplicationLifecycleGuard {
-    async fn acquire() -> Self {
-        Self {
-            _guard: SITE_REPLICATION_LIFECYCLE_LOCK.lock().await,
+    /// Bounded acquire: a holder wedged on unreachable peers (each probe
+    /// costs up to [`SITE_REPLICATION_PEER_REQUEST_TIMEOUT`]) must not hang
+    /// every other lifecycle operation indefinitely, so waiters get a
+    /// retryable 503 after [`SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT`].
+    async fn acquire() -> S3Result<Self> {
+        match tokio::time::timeout(SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT, SITE_REPLICATION_LIFECYCLE_LOCK.lock()).await {
+            Ok(guard) => Ok(Self { _guard: guard }),
+            Err(_) => Err(S3Error::with_message(
+                S3ErrorCode::ServiceUnavailable,
+                "another site replication lifecycle operation is in progress; retry later".to_string(),
+            )),
         }
     }
 
@@ -2115,6 +2128,27 @@ async fn remote_add_preflight_info(site: &PeerSite) -> S3Result<SiteReplicationA
     add_preflight_info_from_sr_info(site, info, idp_settings)
 }
 
+/// Preflight every site in an add request while the lifecycle lock is held.
+/// Probes run concurrently (matching the other peer fan-outs in this file):
+/// k unreachable sites cost roughly one peer request timeout, not k of them.
+/// Results (and the first error, if any) are reported in request order.
+async fn add_preflight_infos(
+    sites: &[PeerSite],
+    current_state: &SiteReplicationState,
+    local_peer: &PeerInfo,
+) -> S3Result<Vec<SiteReplicationAddPreflightInfo>> {
+    futures::future::join_all(sites.iter().map(|site| async move {
+        if same_identity_endpoint(&site.endpoint, &local_peer.endpoint) {
+            local_add_preflight_info(current_state, local_peer, site).await
+        } else {
+            remote_add_preflight_info(site).await
+        }
+    }))
+    .await
+    .into_iter()
+    .collect()
+}
+
 fn validate_add_preflight_topology(infos: &[SiteReplicationAddPreflightInfo], local_peer: &PeerInfo) -> S3Result<()> {
     let mut deployment_ids = HashSet::new();
     let mut local_seen = false;
@@ -9858,7 +9892,7 @@ impl Operation for SiteReplicationAddHandler {
         let cred = validate_site_replication_admin_request(&req, AdminAction::SiteReplicationAddAction).await?;
         reject_site_replicator_on_public_admin(&cred)?;
         let replicate_ilm_expiry = sr_add_replicate_ilm_expiry(&req.uri);
-        let lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
+        let lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
         // Everything up to the commit below is preflight: peer probes, IAM
         // work and the join fan-out all talk to the network, so none of it may
         // run inside the state transaction. The snapshot read here is what the
@@ -9873,14 +9907,7 @@ impl Operation for SiteReplicationAddHandler {
         // inject it so the add preflight (which requires the local deployment) succeeds. No-op for `mc`.
         ensure_local_site_present(&mut sites, &local_peer);
         validate_add_sites(&sites, &local_peer)?;
-        let mut preflight_infos = Vec::with_capacity(sites.len());
-        for site in &sites {
-            if same_identity_endpoint(&site.endpoint, &local_peer.endpoint) {
-                preflight_infos.push(local_add_preflight_info(&current_state, &local_peer, site).await?);
-            } else {
-                preflight_infos.push(remote_add_preflight_info(site).await?);
-            }
-        }
+        let preflight_infos = add_preflight_infos(&sites, &current_state, &local_peer).await?;
         validate_add_preflight_topology(&preflight_infos, &local_peer)?;
         let expected_updated_at = current_state.updated_at;
         require_add_peer_tls_capability(&sites, &local_peer).await?;
@@ -10088,7 +10115,7 @@ impl Operation for SiteReplicationRemoveHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let cred = validate_site_replication_admin_request(&req, AdminAction::SiteReplicationRemoveAction).await?;
         reject_site_replicator_on_public_admin(&cred)?;
-        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
+        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
         // The request body is read before the bucket-op guard and the state
         // transaction: a client that stalls mid-body must hold neither the
         // state-object lock nor the write half of the bucket-op RwLock (which
@@ -10286,7 +10313,7 @@ where
     F: FnOnce(SRPeerJoinReq) -> Fut + Send + 'static,
     Fut: std::future::Future<Output = S3Result<()>> + Send + 'static,
 {
-    let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
+    let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
     admit_peer_join_across_nodes(local_endpoint, join_req, defer_sync_state_enable, apply_iam).await
 }
 
@@ -11101,7 +11128,7 @@ impl Operation for SRPeerRemoveHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         validate_site_replication_admin_request(&req, AdminAction::SiteReplicationRemoveAction).await?;
         let remove_req: SRRemoveReq = read_site_replication_json(req, "", false).await?;
-        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
+        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
         let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.write().await;
         let removed_deployment_ids = update_site_replication_state(move |state| {
             if pending_endpoint_refresh(state).is_some() {
@@ -11145,7 +11172,7 @@ impl Operation for SiteReplicationResyncOpHandler {
         let operation = query.get("operation").cloned().unwrap_or_default();
         let resolved_store = object_store_from_req(&req);
         let requested_peer: PeerInfo = read_site_replication_json(req, "", false).await?;
-        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
+        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
         let (peer, existing_status) = {
             let state = load_site_replication_state().await?;
             let local_peer = current_local_runtime_peer(&state);
@@ -11437,7 +11464,7 @@ impl Operation for SRRotateServiceAccountHandler {
         // mid-repair and race its own IAM write against the reconciler's
         // stale one. (The removed process mutex used to provide this
         // exclusion as a side effect.)
-        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await;
+        let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await?;
         let local_endpoint = site_replication_local_endpoint(&req.uri, &req.headers);
         let rotation_parent = cred.access_key.clone();
         let (pending_rotation, local_peer, previous_access_key) = update_site_replication_state_when_changed(move |state| {
@@ -13930,7 +13957,9 @@ mod tests {
     async fn test_add_bootstrap_scope_only_allows_expected_bucket_setup_until_guard_drops() {
         let token;
         {
-            let lifecycle = SiteReplicationLifecycleGuard::acquire().await;
+            let lifecycle = SiteReplicationLifecycleGuard::acquire()
+                .await
+                .expect("acquire lifecycle guard");
             let guard = SiteReplicationAddInProgressGuard::start(lifecycle, HashSet::from(["legacy-bucket".to_string()]))
                 .expect("start site replication add guard");
             token = guard.token.to_string();
@@ -13986,14 +14015,18 @@ mod tests {
     #[tokio::test]
     #[serial]
     async fn test_add_lifecycle_allows_callback_before_remove_writer() {
-        let lifecycle = SiteReplicationLifecycleGuard::acquire().await;
+        let lifecycle = SiteReplicationLifecycleGuard::acquire()
+            .await
+            .expect("acquire lifecycle guard");
         let add_guard =
             SiteReplicationAddInProgressGuard::start(lifecycle, HashSet::new()).expect("start site replication add guard");
         let (started_tx, started_rx) = tokio::sync::oneshot::channel();
         let (entered_tx, mut entered_rx) = tokio::sync::oneshot::channel();
         let remove = tokio::spawn(async move {
             let _ = started_tx.send(());
-            let _lifecycle = SiteReplicationLifecycleGuard::acquire().await;
+            let _lifecycle = SiteReplicationLifecycleGuard::acquire()
+                .await
+                .expect("acquire lifecycle guard");
             let _bucket_op = SITE_REPLICATION_BUCKET_OP_LOCK.write().await;
             let _ = entered_tx.send(());
         });
@@ -14013,6 +14046,111 @@ mod tests {
         entered_rx.await.expect("remove entered lifecycle");
     }
 
+    /// Deleting either constant (or "simplifying" the client builders to
+    /// inline values) removes the only bound on how long a lifecycle
+    /// operation can be wedged per unreachable peer (#1889 C1 / #1952 C2).
+    #[test]
+    fn test_peer_timeout_constants_bound_unreachable_peer_probes() {
+        assert_eq!(SITE_REPLICATION_PEER_REQUEST_TIMEOUT, Duration::from_secs(10));
+        assert_eq!(SITE_REPLICATION_PEER_CONNECT_TIMEOUT, Duration::from_secs(3));
+        assert!(
+            SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT >= SITE_REPLICATION_PEER_REQUEST_TIMEOUT,
+            "a waiter must not give up before the holder's single wedged peer probe can finish"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[serial]
+    async fn test_lifecycle_guard_acquire_times_out_with_retryable_503() {
+        let holder = SiteReplicationLifecycleGuard::acquire().await.expect("first acquire");
+        let err =
+            match tokio::time::timeout(SITE_REPLICATION_LIFECYCLE_LOCK_TIMEOUT * 2, SiteReplicationLifecycleGuard::acquire())
+                .await
+                .expect("bounded acquire must not hang while the lock is held")
+            {
+                Ok(_) => panic!("acquire while the lock is held should time out"),
+                Err(err) => err,
+            };
+        assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
+
+        drop(holder);
+        tokio::time::timeout(Duration::from_secs(1), SiteReplicationLifecycleGuard::acquire())
+            .await
+            .expect("acquire after release must not wait")
+            .expect("acquire after release");
+    }
+
+    #[derive(Clone)]
+    struct PreflightFanoutTestState {
+        metainfo_barrier: Arc<tokio::sync::Barrier>,
+    }
+
+    async fn preflight_fanout_test_handler(State(state): State<PreflightFanoutTestState>, uri: Uri) -> (StatusCode, String) {
+        if uri.path().ends_with("/site-replication/metainfo") {
+            state.metainfo_barrier.wait().await;
+        }
+        (StatusCode::OK, "{}".to_string())
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_add_preflight_probes_sites_concurrently() {
+        temp_env::async_with_vars(
+            [(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))],
+            add_preflight_probes_sites_concurrently_inner(),
+        )
+        .await;
+    }
+
+    async fn add_preflight_probes_sites_concurrently_inner() {
+        const REMOTE_SITES: usize = 3;
+        let listener = match TcpListener::bind("127.0.0.1:0").await {
+            Ok(listener) => listener,
+            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
+            Err(err) => panic!("bind preflight test server: {err}"),
+        };
+        let endpoint = format!("http://{}", listener.local_addr().expect("preflight test address"));
+        let state = PreflightFanoutTestState {
+            metainfo_barrier: Arc::new(tokio::sync::Barrier::new(REMOTE_SITES)),
+        };
+        let server = tokio::spawn(async move {
+            axum::serve(listener, Router::new().fallback(any(preflight_fanout_test_handler)).with_state(state))
+                .await
+                .expect("serve preflight test requests");
+        });
+
+        let sites: Vec<PeerSite> = (0..REMOTE_SITES)
+            .map(|index| PeerSite {
+                name: format!("site-{index}"),
+                endpoint: endpoint.clone(),
+                access_key: "test-access".to_string(),
+                secret_key: "test-secret".to_string(),
+                ..Default::default()
+            })
+            .collect();
+        let local_peer = PeerInfo {
+            deployment_id: "local".to_string(),
+            endpoint: "http://192.0.2.1:9000".to_string(),
+            ..Default::default()
+        };
+        let current_state = SiteReplicationState::default();
+
+        // Each site's metainfo request parks on a barrier that only releases
+        // once every site's request has arrived: serial probing never sends
+        // the second request and dies on the peer request timeout, so
+        // finishing well inside that timeout proves the probes overlap —
+        // which is what caps k unreachable sites at one timeout, not k.
+        let infos = tokio::time::timeout(
+            SITE_REPLICATION_PEER_REQUEST_TIMEOUT / 2,
+            add_preflight_infos(&sites, &current_state, &local_peer),
+        )
+        .await
+        .expect("preflight probes must fan out concurrently, not serially")
+        .expect("preflight infos");
+        assert_eq!(infos.len(), REMOTE_SITES);
+        server.abort();
+    }
+
     #[test]
     fn test_merge_add_sites_propagates_replicate_ilm_expiry() {
         let state = merge_add_sites(