From 0e2e01d060b0a9f4e4f07f443e0eb5108ddb4564 Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 21 Jul 2026 00:12:08 +0800 Subject: [PATCH] fix(site-replication): harden add finalization (#5064) --- .../src/replication_extension_test.rs | 5 +- rustfs/src/admin/handlers/site_replication.rs | 284 +++++++++++++----- rustfs/src/admin/service/site_replication.rs | 95 +++++- rustfs/src/admin/site_replication_identity.rs | 10 +- 4 files changed, 298 insertions(+), 96 deletions(-) diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 8fa85d417..1e94e1688 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -1971,7 +1971,10 @@ async fn wait_for_site_replication_enabled( ) -> Result> { for _ in 0..40 { let info = site_replication_info(env).await?; - if info.enabled && info.sites.len() == expected_sites { + if info.enabled + && info.sites.len() == expected_sites + && info.sites.iter().all(|peer| peer.sync_state == SyncStatus::Enable) + { return Ok(info); } sleep(Duration::from_millis(250)).await; diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 7d607585f..70aa4d4d5 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -21,8 +21,8 @@ use crate::admin::runtime_sources::{ current_token_signing_key, object_store_from_req, }; use crate::admin::site_replication_identity::{ - canonical_endpoint, deployment_id_for_endpoint, normalize_peer_map_by_identity_with, same_identity_endpoint, - site_identity_key, + canonical_endpoint, deployment_id_for_endpoint, mark_unknown_peer_sync_enabled, normalize_peer_map_by_identity_with, + same_identity_endpoint, site_identity_key, }; use crate::admin::storage_api::bucket::metadata::{ BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_QUOTA_CONFIG_FILE, BUCKET_REPLICATION_CONFIG, @@ -119,6 +119,7 @@ const SITE_REPL_MAX_NETPERF_DURATION: Duration = Duration::from_secs(30); const SITE_REPLICATION_PEER_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); const SITE_REPLICATION_PEER_CONNECT_TIMEOUT: Duration = Duration::from_secs(3); 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; const ALLOW_LOOPBACK_REPLICATION_TARGET_ENV: &str = "RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET"; const SITE_REPLICATION_RETRY_QUEUE_LIMIT: usize = 256; @@ -427,6 +428,8 @@ struct SiteReplicationState { pending_endpoint_refresh: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] retry_queue: Vec, + #[serde(default)] + sync_state_initialized: bool, } #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -519,6 +522,57 @@ struct SiteReplicationBootstrapPlan { #[derive(Debug, Clone, Serialize, Deserialize, Default)] struct SRPeerJoinResponse { peer: PeerInfo, + #[serde(rename = "initialSyncErrorMessage", default, skip_serializing_if = "String::is_empty")] + initial_sync_error_message: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct SRPeerJoinEnvelope { + #[serde(flatten)] + request: SRPeerJoinReq, + #[serde(rename = "deferSyncStateEnable", default, skip_serializing_if = "std::ops::Not::not")] + defer_sync_state_enable: bool, +} + +#[derive(Debug, Default)] +struct SiteReplicationErrorSummary { + entries: Vec, + total: usize, +} + +impl SiteReplicationErrorSummary { + fn push(&mut self, error: impl AsRef) { + self.total = self.total.saturating_add(1); + if self.entries.len() < SITE_REPLICATION_INITIAL_SYNC_ERROR_LIMIT { + self.entries.push(summarize_peer_error_detail(error.as_ref())); + } + } + + fn extend(&mut self, other: Self) { + self.total = self.total.saturating_add(other.total); + let remaining = SITE_REPLICATION_INITIAL_SYNC_ERROR_LIMIT.saturating_sub(self.entries.len()); + self.entries.extend(other.entries.into_iter().take(remaining)); + } + + fn is_empty(&self) -> bool { + self.total == 0 + } + + fn reported(&self) -> usize { + self.entries.len() + } + + fn render(&self) -> String { + let mut message = self.entries.join("; "); + let omitted = self.total.saturating_sub(self.entries.len()); + if omitted > 0 { + if !message.is_empty() { + message.push_str("; "); + } + message.push_str(&format!("{omitted} additional error(s) omitted")); + } + message + } } const GO_GOB_SITE_NETPERF_SCHEMA: &[u8] = &[ @@ -838,6 +892,12 @@ async fn load_site_replication_state() -> S3Result { let mut state: SiteReplicationState = serde_json::from_slice(&data) .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid site replication state: {e}")))?; state.peers = normalize_peer_map_by_identity(state.peers); + if !state.sync_state_initialized { + if state.enabled() { + mark_unknown_peer_sync_enabled(&mut state.peers); + } + state.sync_state_initialized = true; + } Ok(state) } Err(StorageError::ConfigNotFound) => Ok(SiteReplicationState::default()), @@ -2138,21 +2198,6 @@ fn build_join_peers( normalize_peer_map_by_identity(peers) } -/// Promote each peer's persisted `sync_state` from `Unknown` to `Enable` once site -/// replication has been established for it. The persisted value is a static -/// "configured & enabled" signal so that the `replicate info` endpoint (and the -/// console, which read the persisted peer map verbatim) do not report `Unknown` for a -/// healthy peer. `build_status_info` still refines this to `Disable`/`Unknown` from -/// live health signals at query time for the `replicate status` endpoint. A peer that -/// has been explicitly marked `Disable` is left untouched. -fn mark_peers_sync_enabled(peers: &mut BTreeMap) { - for peer in peers.values_mut() { - if peer.sync_state == SyncStatus::Unknown { - peer.sync_state = SyncStatus::Enable; - } - } -} - fn normalize_join_peers_for_local(local_peer: &PeerInfo, peers: BTreeMap) -> BTreeMap { let mut normalized = BTreeMap::new(); @@ -2174,6 +2219,12 @@ fn normalize_join_peers_for_local(local_peer: &PeerInfo, peers: BTreeMap, defer_sync_state_enable: bool) { + if !defer_sync_state_enable { + mark_unknown_peer_sync_enabled(peers); + } +} + fn reconcile_peer_with_actual_identity(mut state: SiteReplicationState, actual_peer: PeerInfo) -> SiteReplicationState { let mut actual_peer = normalize_peer_info(actual_peer); if let Some(requested_peer) = state @@ -2724,27 +2775,25 @@ async fn bootstrap_existing_metadata_after_add( state: &SiteReplicationState, local_peer: &PeerInfo, service_account_secret_key: &str, -) -> Vec { +) -> SiteReplicationErrorSummary { let info = match build_sr_info(state, local_peer).await { Ok(info) => info, Err(err) => { - return vec![format!( - "local snapshot failed: {}", - summarize_peer_error_detail(&err.to_string()) - )]; + let mut errors = SiteReplicationErrorSummary::default(); + errors.push(format!("local snapshot failed: {err}")); + return errors; } }; let plan = match site_replication_bootstrap_plan(&info) { Ok(plan) => plan, Err(err) => { - return vec![format!( - "bootstrap plan failed: {}", - summarize_peer_error_detail(&err.to_string()) - )]; + let mut errors = SiteReplicationErrorSummary::default(); + errors.push(format!("bootstrap plan failed: {err}")); + return errors; } }; - let mut errors = Vec::new(); + let mut errors = SiteReplicationErrorSummary::default(); for peer in state.peers.values() { if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) { continue; @@ -2769,17 +2818,6 @@ async fn bootstrap_existing_metadata_after_add( errors } -/// Combine the metadata-bootstrap and object-back-fill failure lists into the single -/// `initial_sync_error_message` surfaced by `mc admin replicate add`, so an operator can -/// see exactly which buckets did not propagate instead of receiving an unqualified success. -fn compose_initial_sync_error_message(bootstrap_errors: Vec, backfill_errors: Vec) -> String { - bootstrap_errors - .into_iter() - .chain(backfill_errors) - .collect::>() - .join("; ") -} - pub async fn site_replication_make_bucket_hook(bucket: &str, lock_enabled: bool) -> S3Result<()> { let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.read().await; let runtime = { @@ -5431,10 +5469,10 @@ async fn backfill_existing_buckets_after_add( state: &SiteReplicationState, local_peer: &PeerInfo, bootstrap_token: Option<&str>, -) -> Vec { - let mut errors = Vec::new(); +) -> SiteReplicationErrorSummary { + let mut errors = SiteReplicationErrorSummary::default(); let Some(store) = current_object_store_handle() else { - errors.push("object store not initialized; pre-existing buckets were not backfilled".to_string()); + errors.push("object store not initialized; pre-existing buckets were not backfilled"); return errors; }; let buckets = match store.list_bucket(&BucketOptions::default()).await { @@ -5448,7 +5486,7 @@ async fn backfill_existing_buckets_after_add( error = ?err, "admin site replication state" ); - errors.push(format!("list buckets failed: {}", summarize_peer_error_detail(&err.to_string()))); + errors.push(format!("list buckets failed: {err}")); return errors; } }; @@ -5467,10 +5505,7 @@ async fn backfill_existing_buckets_after_add( error = ?err, "admin site replication state" ); - errors.push(format!( - "{name}: versioning setup failed: {}", - summarize_peer_error_detail(&err.to_string()) - )); + errors.push(format!("{name}: versioning setup failed: {err}")); continue; } match ensure_site_replication_bucket_setup(name).await { @@ -5500,7 +5535,7 @@ async fn backfill_existing_buckets_after_add( error = ?err, "admin site replication state" ); - errors.push(format!("{name}: bucket setup failed: {}", summarize_peer_error_detail(&err.to_string()))); + errors.push(format!("{name}: bucket setup failed: {err}")); } } // Broadcast the bucket to peers so they create it too (idempotent on the peer side). @@ -5532,10 +5567,7 @@ async fn backfill_existing_buckets_after_add( error = ?err, "admin site replication state" ); - errors.push(format!( - "{name}: make-bucket broadcast failed: {}", - summarize_peer_error_detail(&err.to_string()) - )); + errors.push(format!("{name}: make-bucket broadcast failed: {err}")); } // Kick a resync toward every remote peer so existing objects travel across. for peer in state.peers.values() { @@ -5554,11 +5586,7 @@ async fn backfill_existing_buckets_after_add( detail = %result.err_detail, "admin site replication state" ); - errors.push(format!( - "{name} -> {}: resync kick failed: {}", - peer.endpoint, - summarize_peer_error_detail(&result.err_detail) - )); + errors.push(format!("{name} -> {}: resync kick failed: {}", peer.endpoint, result.err_detail)); } } } @@ -6245,17 +6273,22 @@ impl Operation for SiteReplicationAddHandler { cred.access_key.clone(), replicate_ilm_expiry, ); - let join_req = SRPeerJoinReq { - svc_acct_access_key: service_account_access_key, - svc_acct_secret_key: service_account_secret_key.clone(), - svc_acct_parent: String::new(), - peers: state.peers.clone(), - updated_at: state.updated_at, + state.sync_state_initialized = true; + let join_req = SRPeerJoinEnvelope { + request: SRPeerJoinReq { + svc_acct_access_key: service_account_access_key, + svc_acct_secret_key: service_account_secret_key.clone(), + svc_acct_parent: String::new(), + peers: state.peers.clone(), + updated_at: state.updated_at, + }, + defer_sync_state_enable: true, }; let peer_join_path = with_site_replication_bootstrap_token(SITE_REPLICATION_PEER_JOIN_PATH, &add_in_progress_guard.token.to_string()); let mut joined_endpoints = HashSet::new(); + let mut initial_sync_errors = SiteReplicationErrorSummary::default(); for site in &sites { if same_identity_endpoint(&site.endpoint, &local_peer.endpoint) || !joined_endpoints.insert(site_identity_key(&site.endpoint)) @@ -6264,7 +6297,7 @@ impl Operation for SiteReplicationAddHandler { } let mut peer_join_req = join_req.clone(); - peer_join_req.svc_acct_parent = site.access_key.clone(); + peer_join_req.request.svc_acct_parent = site.access_key.clone(); let connection = PeerConnection::try_from(site)?; let body = send_peer_admin_request(&connection, &peer_join_path, &site.access_key, &site.secret_key, &peer_join_req).await?; @@ -6275,6 +6308,9 @@ impl Operation for SiteReplicationAddHandler { format!("parse peer join response from {} failed: {e}", site.endpoint), ) })?; + if !join_response.initial_sync_error_message.is_empty() { + initial_sync_errors.push(format!("{}: {}", site.endpoint, join_response.initial_sync_error_message)); + } state = reconcile_peer_with_actual_identity(state, join_response.peer); let reconciled_peer = existing_peer_for_endpoint(&state, &site.endpoint).ok_or_else(|| { S3Error::with_message( @@ -6290,22 +6326,49 @@ impl Operation for SiteReplicationAddHandler { })?; } - // BUG1: persist a real sync_state so `mc admin replicate info` and the console report - // Enable for healthy peers instead of Unknown. build_status_info still refines this from - // live health for the `replicate status` endpoint. - mark_peers_sync_enabled(&mut state.peers); + mark_unknown_peer_sync_enabled(&mut state.peers); persist_site_replication_state(&state).await?; - let bootstrap_errors = bootstrap_existing_metadata_after_add(&state, &local_peer, &service_account_secret_key).await; + + for target in state.peers.values() { + if target.deployment_id == local_peer.deployment_id || same_identity_endpoint(&target.endpoint, &local_peer.endpoint) + { + continue; + } + let transport = match PeerTransport::for_runtime_peer(target).await { + Ok(transport) => transport, + Err(err) => { + initial_sync_errors.push(format!("{}: finalize peer sync state failed: {err}", target.endpoint)); + continue; + } + }; + for peer in state.peers.values() { + if let Err(err) = send_peer_admin_request_with_client( + &transport.client, + &transport.connection, + SITE_REPLICATION_PEER_EDIT_PATH, + &state.service_account_access_key, + &service_account_secret_key, + peer, + ) + .await + { + initial_sync_errors + .push(format!("{}: finalize sync state for {} failed: {err}", target.endpoint, peer.endpoint)); + } + } + } + + initial_sync_errors.extend(bootstrap_existing_metadata_after_add(&state, &local_peer, &service_account_secret_key).await); // Fix 1: back-fill pre-existing buckets so objects created before `replicate add` // are not silently left out of replication. Per-bucket failures are surfaced in the add // response below (BUG2) rather than swallowed; they do not abort the overall add. - let backfill_errors = backfill_existing_buckets_after_add(&state, &local_peer, None).await; + initial_sync_errors.extend(backfill_existing_buckets_after_add(&state, &local_peer, None).await); json_response(&ReplicateAddStatus { success: true, status: SITE_REPL_ADD_SUCCESS.to_string(), - initial_sync_error_message: compose_initial_sync_error_message(bootstrap_errors, backfill_errors), + initial_sync_error_message: initial_sync_errors.render(), api_version: Some(SITE_REPL_API_VERSION.to_string()), ..Default::default() }) @@ -6533,18 +6596,22 @@ impl Operation for SRPeerJoinHandler { let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; let mut state = load_site_replication_state().await?; let local_peer = current_local_peer(&req, &state); - let join_req: SRPeerJoinReq = read_site_replication_json(req, &cred.secret_key, true).await?; + let join_envelope: SRPeerJoinEnvelope = read_site_replication_json(req, &cred.secret_key, true).await?; + let defer_sync_state_enable = join_envelope.defer_sync_state_enable; + let join_req = join_envelope.request; validate_join_peer_snapshot(&join_req.peers)?; if let Some(current_updated_at) = state.updated_at { let Some(incoming_updated_at) = join_req.updated_at else { return json_response(&SRPeerJoinResponse { peer: state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer), + ..Default::default() }); }; if incoming_updated_at <= current_updated_at { return json_response(&SRPeerJoinResponse { peer: state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer), + ..Default::default() }); } } @@ -6603,9 +6670,8 @@ impl Operation for SRPeerJoinHandler { state.service_account_parent = join_req.svc_acct_parent; state.updated_at = join_req.updated_at.or_else(|| Some(OffsetDateTime::now_utc())); state.peers = normalize_join_peers_for_local(&local_peer, join_req.peers); - // BUG1: persist Enable on the joining side too, so its `replicate info` endpoint reports a - // real sync state rather than Unknown. - mark_peers_sync_enabled(&mut state.peers); + initialize_join_peer_sync_state(&mut state.peers, defer_sync_state_enable); + state.sync_state_initialized = true; state.name = state .peers .get(&local_peer.deployment_id) @@ -6623,12 +6689,14 @@ impl Operation for SRPeerJoinHandler { component = LOG_COMPONENT_ADMIN, subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, result = "join_backfill_incomplete", - errors = %backfill_errors.join("; "), + error_count = backfill_errors.total, + reported_error_count = backfill_errors.reported(), "admin site replication state" ); } json_response(&SRPeerJoinResponse { peer: state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer), + initial_sync_error_message: backfill_errors.render(), }) } } @@ -7327,7 +7395,7 @@ impl Operation for SiteReplicationRepairHandler { } else { "Partial".to_string() }, - err_detail: repair_errors.join("; "), + err_detail: repair_errors.render(), api_version: Some(SITE_REPL_API_VERSION.to_string()), }) } @@ -11151,7 +11219,7 @@ mod tests { peers.values().all(|p| p.sync_state == SyncStatus::Unknown), "freshly constructed peers default to Unknown" ); - mark_peers_sync_enabled(&mut peers); + mark_unknown_peer_sync_enabled(&mut peers); assert!( !peers.is_empty() && peers.values().all(|p| p.sync_state == SyncStatus::Enable), "add/join must persist Enable so the info endpoint reports a real sync state" @@ -11178,11 +11246,35 @@ mod tests { ..peer("b", "https://b.example.com") }, ); - mark_peers_sync_enabled(&mut peers); + mark_unknown_peer_sync_enabled(&mut peers); assert_eq!(peers["a"].sync_state, SyncStatus::Enable, "Unknown must be promoted to Enable"); assert_eq!(peers["b"].sync_state, SyncStatus::Disable, "explicit Disable must be preserved"); } + #[test] + fn test_join_peer_sync_state_waits_for_deferred_commit() { + let mut peers = BTreeMap::from([("a".to_string(), peer("a", "https://a.example.com"))]); + + initialize_join_peer_sync_state(&mut peers, true); + assert_eq!(peers["a"].sync_state, SyncStatus::Unknown); + + initialize_join_peer_sync_state(&mut peers, false); + assert_eq!(peers["a"].sync_state, SyncStatus::Enable); + } + + #[test] + fn test_join_deferred_sync_state_flag_is_wire_compatible() { + let legacy: SRPeerJoinEnvelope = serde_json::from_value(serde_json::json!({})).expect("parse legacy peer join request"); + assert!(!legacy.defer_sync_state_enable); + + let value = serde_json::to_value(SRPeerJoinEnvelope { + defer_sync_state_enable: true, + ..Default::default() + }) + .expect("serialize peer join request"); + assert_eq!(value.get("deferSyncStateEnable"), Some(&Value::Bool(true))); + } + // BUG2: pre-existing-bucket back-fill failures must be surfaced in the add response's // initial_sync_error_message, not swallowed behind an unqualified success. #[test] @@ -11192,7 +11284,11 @@ mod tests { "test78787: replication setup skipped (site replication runtime unavailable)".to_string(), "test78787 -> https://peer.example.com: resync kick failed: timeout".to_string(), ]; - let msg = compose_initial_sync_error_message(bootstrap_errors, backfill_errors); + let mut errors = SiteReplicationErrorSummary::default(); + for error in bootstrap_errors.into_iter().chain(backfill_errors) { + errors.push(error); + } + let msg = errors.render(); assert!(msg.contains("peer-x: metadata sync failed"), "bootstrap errors must be surfaced"); assert!( msg.contains("test78787: replication setup skipped"), @@ -11202,8 +11298,34 @@ mod tests { } #[test] - fn test_compose_initial_sync_error_message_empty_on_success() { - assert_eq!(compose_initial_sync_error_message(Vec::new(), Vec::new()), ""); + fn test_initial_sync_error_summary_is_bounded() { + let mut errors = SiteReplicationErrorSummary::default(); + for index in 0..(SITE_REPLICATION_INITIAL_SYNC_ERROR_LIMIT + 5) { + errors.push(format!("bucket-{index}: {}", "x".repeat(SITE_REPLICATION_PEER_ERROR_DETAIL_LIMIT + 32))); + } + + let message = errors.render(); + + assert_eq!(errors.reported(), SITE_REPLICATION_INITIAL_SYNC_ERROR_LIMIT); + assert!(message.contains("5 additional error(s) omitted")); + assert!(message.chars().count() <= SITE_REPLICATION_INITIAL_SYNC_ERROR_LIMIT * 258 + 64); + } + + #[test] + fn test_peer_join_response_error_summary_is_wire_compatible() { + let response: SRPeerJoinResponse = serde_json::from_value(serde_json::json!({ + "peer": peer("remote", "https://remote.example.com") + })) + .expect("parse legacy peer join response"); + + assert!(response.initial_sync_error_message.is_empty()); + + let value = serde_json::to_value(SRPeerJoinResponse { + peer: peer("remote", "https://remote.example.com"), + initial_sync_error_message: "bucket setup failed".to_string(), + }) + .expect("serialize peer join response"); + assert_eq!(value.get("initialSyncErrorMessage").and_then(Value::as_str), Some("bucket setup failed")); } // Fix 5: remove --all must purge local state unconditionally even when peer errors occur diff --git a/rustfs/src/admin/service/site_replication.rs b/rustfs/src/admin/service/site_replication.rs index 74c5ff859..8af98d6a7 100644 --- a/rustfs/src/admin/service/site_replication.rs +++ b/rustfs/src/admin/service/site_replication.rs @@ -13,7 +13,9 @@ // limitations under the License. use crate::admin::runtime_sources::{AppContext, current_app_context, current_object_store_handle_for_context}; -use crate::admin::site_replication_identity::{deployment_id_for_endpoint, normalize_peer_map_by_identity_with}; +use crate::admin::site_replication_identity::{ + deployment_id_for_endpoint, mark_unknown_peer_sync_enabled, normalize_peer_map_by_identity_with, +}; use crate::admin::storage_api::config::{read_admin_config, save_admin_config}; use crate::admin::storage_api::error::Error as StorageError; use rustfs_madmin::PeerInfo; @@ -22,8 +24,9 @@ use serde_json::{Map, Value}; use tracing::info; const SITE_REPLICATION_STATE_PATH: &str = "config/site-replication/state.json"; +const SYNC_STATE_INITIALIZED_FIELD: &str = "sync_state_initialized"; -fn normalize_peers_map(peers: &Map) -> Map { +fn normalize_peers_map(peers: &Map, initialize_sync_state: bool) -> Map { let mut valid_peers = std::collections::BTreeMap::::new(); let mut passthrough_invalid = Vec::<(String, Value)>::new(); @@ -46,7 +49,10 @@ fn normalize_peers_map(peers: &Map) -> Map { } } - let deduped_by_deployment = normalize_peer_map_by_identity_with(valid_peers, |peer| peer); + let mut deduped_by_deployment = normalize_peer_map_by_identity_with(valid_peers, |peer| peer); + if initialize_sync_state && deduped_by_deployment.len() > 1 { + mark_unknown_peer_sync_enabled(&mut deduped_by_deployment); + } let mut normalized = Map::new(); for (_, peer) in deduped_by_deployment { @@ -69,20 +75,32 @@ fn normalize_site_replication_state_json(data: &[u8]) -> Result>, let before = obj.get("peers").and_then(|v| v.as_object()).map(|v| v.len()).unwrap_or(0); + let sync_state_initialized = obj + .get(SYNC_STATE_INITIALIZED_FIELD) + .and_then(Value::as_bool) + .unwrap_or(false); let Some(peers_obj) = obj.get("peers").and_then(|v| v.as_object()) else { return Ok(None); }; - - let normalized_peers = normalize_peers_map(peers_obj); - if normalized_peers == *peers_obj { + let normalized_peers = normalize_peers_map(peers_obj, !sync_state_initialized); + if normalized_peers == *peers_obj && sync_state_initialized { return Ok(None); } let after = normalized_peers.len(); obj.insert("peers".to_string(), Value::Object(normalized_peers)); + obj.insert(SYNC_STATE_INITIALIZED_FIELD.to_string(), Value::Bool(true)); let normalized = serde_json::to_vec(&state).map_err(|e| format!("serialize normalized site replication state failed: {e}"))?; - info!("normalized site-replication peers during reload: {before} -> {after}"); + info!( + event = "site_replication_state_normalized", + component = "admin", + subsystem = "site_replication", + peers_before = before, + peers_after = after, + sync_state_migrated = !sync_state_initialized, + "site replication state normalized" + ); Ok(Some(normalized)) } @@ -129,12 +147,12 @@ mod tests { serde_json::json!({ "name": name, "endpoint": endpoint, - "deployment_id": deployment_id, - "sync_state": "", - "default_bandwidth": {}, - "replicate_ilm_expiry": false, - "object_naming_mode": "", - "api_version": "1" + "deploymentID": deployment_id, + "sync": "unknown", + "defaultbandwidth": {}, + "replicate-ilm-expiry": false, + "objectNamingMode": "", + "apiVersion": "1" }) } @@ -220,4 +238,55 @@ mod tests { assert_eq!(peers.len(), 2); } + + #[test] + fn test_normalize_state_json_migrates_legacy_unknown_sync_state() { + let data = serde_json::to_vec(&serde_json::json!({ + "name": "local", + "peers": { + "local": peer_value("local", "https://local.example.com", "local"), + "remote": peer_value("remote", "https://remote.example.com", "remote") + } + })) + .expect("serialize legacy state"); + + let normalized = normalize_site_replication_state_json(&data) + .expect("normalize legacy state") + .expect("legacy state should be migrated"); + let value: Value = serde_json::from_slice(&normalized).expect("parse normalized state"); + let peers = value.get("peers").and_then(Value::as_object).expect("normalized peers"); + + assert_eq!(value.get(SYNC_STATE_INITIALIZED_FIELD), Some(&Value::Bool(true))); + assert!( + peers + .values() + .all(|peer| peer.get("sync").and_then(Value::as_str) == Some("enable")) + ); + } + + #[test] + fn test_normalize_state_json_preserves_initialized_pending_sync_state() { + let data = serde_json::to_vec(&serde_json::json!({ + "name": "local", + (SYNC_STATE_INITIALIZED_FIELD): true, + "peers": { + "local": peer_value("local", "https://local.example.com", "local"), + "remote": peer_value("remote", "https://remote.example.com", "remote") + } + })) + .expect("serialize pending state"); + + let normalized = normalize_site_replication_state_json(&data) + .expect("normalize pending state") + .expect("pending state should receive canonical peer fields"); + let value: Value = serde_json::from_slice(&normalized).expect("parse normalized pending state"); + let peers = value.get("peers").and_then(Value::as_object).expect("normalized peers"); + + assert_eq!(value.get(SYNC_STATE_INITIALIZED_FIELD), Some(&Value::Bool(true))); + assert!( + peers + .values() + .all(|peer| peer.get("sync").and_then(Value::as_str) == Some("unknown")) + ); + } } diff --git a/rustfs/src/admin/site_replication_identity.rs b/rustfs/src/admin/site_replication_identity.rs index b9bf31e73..67bf350fd 100644 --- a/rustfs/src/admin/site_replication_identity.rs +++ b/rustfs/src/admin/site_replication_identity.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use rustfs_madmin::PeerInfo; +use rustfs_madmin::{PeerInfo, SyncStatus}; use std::collections::{BTreeMap, hash_map::DefaultHasher}; use std::hash::{Hash, Hasher}; use url::Url; @@ -76,6 +76,14 @@ pub fn same_identity_endpoint(left: &str, right: &str) -> bool { site_identity_key(left) == site_identity_key(right) } +pub(crate) fn mark_unknown_peer_sync_enabled(peers: &mut BTreeMap) { + for peer in peers.values_mut() { + if peer.sync_state == SyncStatus::Unknown { + peer.sync_state = SyncStatus::Enable; + } + } +} + fn is_https_endpoint(endpoint: &str) -> bool { canonical_endpoint(endpoint).starts_with("https://") }