From 6e5c58ca3f100be1c55929ae645b89977375c40c Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 1 Jul 2026 09:21:30 +0800 Subject: [PATCH] fix(replication): harden bucket replication correctness (#4116) --- .../ecstore/src/bucket/bucket_target_sys.rs | 167 +++++++++++--- .../ecstore/src/bucket/replication/config.rs | 31 +-- .../replication/replication_resyncer.rs | 134 +++++++++-- crates/filemeta/src/replication.rs | 93 +++++++- rustfs/src/admin/handlers/replication.rs | 216 +++++++++++++++++- rustfs/src/admin/router.rs | 65 ++++-- rustfs/src/admin/storage_api.rs | 1 + rustfs/src/app/bucket_usecase.rs | 193 +++++++++++++--- 8 files changed, 763 insertions(+), 137 deletions(-) diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 83c39b188..bad9194c9 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -82,9 +82,14 @@ use uuid::Uuid; const DEFAULT_HEALTH_CHECK_DURATION: Duration = Duration::from_secs(5); const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60); +const REDACTED_CREDENTIAL: &str = ""; pub static GLOBAL_BUCKET_TARGET_SYS: OnceLock = OnceLock::new(); +fn replication_target_versioning_enabled(versioning: Option<&BucketVersioningStatus>) -> bool { + matches!(versioning, Some(BucketVersioningStatus::Enabled)) +} + #[derive(Debug, Clone)] pub struct ArnTarget { pub client: Option>, @@ -202,6 +207,7 @@ pub struct EpHealth { pub last_online: Option, pub last_hc_at: Option, pub offline_duration: Duration, + pub offline_count: u64, pub latency: LatencyStat, } @@ -214,11 +220,37 @@ impl Default for EpHealth { last_online: None, last_hc_at: None, offline_duration: Duration::from_secs(0), + offline_count: 0, latency: LatencyStat::new(), } } } +fn endpoint_health_key(url: &Url) -> String { + let host = url.host_str().unwrap_or_default(); + match url.port() { + Some(port) => format!("{host}:{port}"), + None => host.to_string(), + } +} + +fn update_endpoint_health(health: &mut EpHealth, online: bool, latency: Duration, now: OffsetDateTime) { + let prev_online = health.online; + health.online = online; + health.last_hc_at = Some(now); + health.latency.update(latency); + + if online { + health.last_online = Some(now); + return; + } + + if prev_online { + health.offline_count += 1; + } + health.offline_duration += latency; +} + #[derive(Debug, Default)] pub struct BucketTargetSys { pub arn_remotes_map: Arc>>, @@ -246,9 +278,10 @@ impl BucketTargetSys { } pub async fn is_offline(&self, url: &Url) -> bool { + let key = endpoint_health_key(url); { let health_map = self.h_mutex.read().await; - if let Some(health) = health_map.get(url.host_str().unwrap_or("")) { + if let Some(health) = health_map.get(&key) { return !health.online; } } @@ -258,15 +291,16 @@ impl BucketTargetSys { } pub async fn mark_offline(&self, url: &Url) { + let key = endpoint_health_key(url); let mut health_map = self.h_mutex.write().await; - if let Some(health) = health_map.get_mut(url.host_str().unwrap_or("")) { - health.online = false; + if let Some(health) = health_map.get_mut(&key) { + update_endpoint_health(health, false, Duration::from_secs(0), OffsetDateTime::now_utc()); } } pub async fn init_hc(&self, url: &Url) { let mut health_map = self.h_mutex.write().await; - let host = url.host_str().unwrap_or("").to_string(); + let host = endpoint_health_key(url); health_map.insert( host.clone(), EpHealth { @@ -285,51 +319,35 @@ impl BucketTargetSys { let endpoints = { let health_map = self.h_mutex.read().await; - health_map.keys().cloned().collect::>() + health_map + .iter() + .map(|(endpoint, health)| (endpoint.clone(), health.scheme.clone())) + .collect::>() }; - for endpoint in endpoints { + for (endpoint, scheme) in endpoints { // Perform health check let start = Instant::now(); - let online = self.check_endpoint_health(&endpoint).await; + let online = self.check_endpoint_health(&endpoint, &scheme).await; let duration = start.elapsed(); { let mut health_map = self.h_mutex.write().await; if let Some(health) = health_map.get_mut(&endpoint) { - let prev_online = health.online; - health.online = online; - health.last_hc_at = Some(OffsetDateTime::now_utc()); - health.latency.update(duration); - - if online { - health.last_online = Some(OffsetDateTime::now_utc()); - } else if prev_online { - // Just went offline - health.offline_duration += duration; - } + update_endpoint_health(health, online, duration, OffsetDateTime::now_utc()); } } } } } - async fn check_endpoint_health(&self, _endpoint: &str) -> bool { - true - // TODO: Health check - - // // Simple health check implementation - // // In a real implementation, you would make actual HTTP requests - // match self - // .hc_client - // .get(format!("https://{}/rustfs/health/ready", endpoint)) - // .timeout(Duration::from_secs(3)) - // .send() - // .await - // { - // Ok(response) => response.status().is_success(), - // Err(_) => false, - // } + async fn check_endpoint_health(&self, endpoint: &str, scheme: &str) -> bool { + let scheme = if scheme.is_empty() { "https" } else { scheme }; + let url = format!("{scheme}://{endpoint}/"); + match self.hc_client.head(url).timeout(Duration::from_secs(3)).send().await { + Ok(response) => response.status().as_u16() < 500, + Err(_) => false, + } } pub async fn health_stats(&self) -> HashMap { @@ -354,6 +372,7 @@ impl BucketTargetSys { avg: health.latency.avg, max: health.latency.peak, }; + target.offline_count = health.offline_count; } targets.push(target); } @@ -375,6 +394,7 @@ impl BucketTargetSys { avg: health.latency.avg, max: health.latency.peak, }; + target.offline_count = health.offline_count; } targets.push(target); } @@ -475,7 +495,7 @@ impl BucketTargetSys { error: e.to_string(), })?; - if versioning.is_none() { + if !replication_target_versioning_enabled(versioning.as_ref()) { return Err(BucketTargetError::BucketRemoteTargetNotVersioned { bucket: target.target_bucket.to_string(), }); @@ -1763,10 +1783,13 @@ impl fmt::Display for BucketTargetError { } BucketTargetError::RemoteTargetConnectionErr { bucket, - access_key, + access_key: _, error, } => { - write!(f, "Connection error for bucket: {bucket}, access key: {access_key}, error: {error}") + write!( + f, + "Connection error for bucket: {bucket}, access key: {REDACTED_CREDENTIAL}, error: {error}" + ) } BucketTargetError::BucketReplicationSourceNotVersioned { bucket } => { write!(f, "Replication source bucket not versioned: {bucket}") @@ -1795,6 +1818,76 @@ mod tests { use super::*; use rcgen::generate_simple_self_signed; + #[test] + fn replication_target_versioning_enabled_requires_enabled_status() { + let enabled = BucketVersioningStatus::Enabled; + let suspended = BucketVersioningStatus::Suspended; + + assert!(replication_target_versioning_enabled(Some(&enabled))); + assert!(!replication_target_versioning_enabled(Some(&suspended))); + assert!(!replication_target_versioning_enabled(None)); + } + + #[test] + fn remote_target_connection_error_display_redacts_access_key() { + let err = BucketTargetError::RemoteTargetConnectionErr { + bucket: "target".to_string(), + access_key: "sensitive-access-key".to_string(), + error: "connection refused".to_string(), + }; + let message = err.to_string(); + + assert!(message.contains(REDACTED_CREDENTIAL)); + assert!(!message.contains("sensitive-access-key")); + assert!(message.contains("connection refused")); + } + + #[test] + fn endpoint_health_key_preserves_explicit_port() { + let url = Url::parse("https://remote.example:9443").expect("url should parse"); + + assert_eq!(endpoint_health_key(&url), "remote.example:9443"); + } + + #[test] + fn update_endpoint_health_counts_offline_transitions() { + let mut health = EpHealth::default(); + let now = OffsetDateTime::now_utc(); + + update_endpoint_health(&mut health, false, Duration::from_millis(25), now); + update_endpoint_health(&mut health, false, Duration::from_millis(25), now); + update_endpoint_health(&mut health, true, Duration::from_millis(10), now); + update_endpoint_health(&mut health, false, Duration::from_millis(25), now); + + assert_eq!(health.offline_count, 2); + assert_eq!(health.offline_duration, Duration::from_millis(75)); + assert_eq!(health.last_online, Some(now)); + } + + #[tokio::test] + async fn list_targets_applies_health_stats_for_endpoint_with_port() { + let sys = BucketTargetSys::default(); + let url = Url::parse("https://remote.example:9443").expect("url should parse"); + sys.init_hc(&url).await; + sys.mark_offline(&url).await; + + sys.targets_map.write().await.insert( + "bucket".to_string(), + vec![BucketTarget { + endpoint: "remote.example:9443".to_string(), + arn: "arn:rustfs:replication:us-east-1:bucket:id".to_string(), + target_type: BucketTargetType::ReplicationService, + ..Default::default() + }], + ); + + let targets = sys.list_targets("", "").await; + + assert_eq!(targets.len(), 1); + assert!(!targets[0].online); + assert_eq!(targets[0].offline_count, 1); + } + #[test] fn build_remove_object_headers_includes_internal_version_id_for_replication_delete() { let version_id = Uuid::new_v4().to_string(); diff --git a/crates/ecstore/src/bucket/replication/config.rs b/crates/ecstore/src/bucket/replication/config.rs index ca78a89da..32e9fe373 100644 --- a/crates/ecstore/src/bucket/replication/config.rs +++ b/crates/ecstore/src/bucket/replication/config.rs @@ -49,9 +49,10 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { /// Check whether any object-replication rules exist fn has_existing_object_replication(&self, arn: &str) -> (bool, bool) { let mut has_arn = false; + let arn = arn.trim(); for rule in &self.rules { - if rule.destination.bucket == arn || self.role == arn { + if rule.destination.bucket.trim() == arn || self.role.trim() == arn { if !has_arn { has_arn = true; } @@ -77,7 +78,10 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { continue; } - if !obj.target_arn.is_empty() && rule.destination.bucket != obj.target_arn && self.role != obj.target_arn { + if !obj.target_arn.is_empty() + && rule.destination.bucket.trim() != obj.target_arn.trim() + && self.role.trim() != obj.target_arn.trim() + { continue; } @@ -215,6 +219,11 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { /// Filter target ARNs and return a slice of the distinct values in the config fn filter_target_arns(&self, obj: &ObjectOpts) -> Vec { + let role = self.role.trim(); + if !role.is_empty() { + return vec![role.to_string()]; + } + let mut arns = Vec::new(); let mut targets_map: HashSet = HashSet::new(); let rules = self.filter_actionable_rules(obj); @@ -224,16 +233,12 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { continue; } - if !rule.destination.bucket.is_empty() && !targets_map.contains(&rule.destination.bucket) { - targets_map.insert(rule.destination.bucket.clone()); + let arn = rule.destination.bucket.trim(); + if !arn.is_empty() && !targets_map.contains(arn) { + targets_map.insert(arn.to_string()); } } - if targets_map.is_empty() && !self.role.is_empty() { - arns.push(self.role.clone()); - return arns; - } - for arn in targets_map { arns.push(arn); } @@ -267,9 +272,9 @@ mod tests { } #[test] - fn filter_target_arns_keeps_multiple_destinations_when_role_is_present() { + fn filter_target_arns_uses_role_when_role_is_present() { let config = ReplicationConfiguration { - role: "arn:legacy:target".to_string(), + role: " arn:legacy:target ".to_string(), rules: vec![ replication_rule("rule-1", "arn:target:a"), replication_rule("rule-2", "arn:target:b"), @@ -282,9 +287,7 @@ mod tests { ..Default::default() }); - assert_eq!(arns.len(), 2); - assert!(arns.iter().any(|arn| arn == "arn:target:a")); - assert!(arns.iter().any(|arn| arn == "arn:target:b")); + assert_eq!(arns, vec!["arn:legacy:target".to_string()]); } #[test] diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 85a8d899e..0618d4fce 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -1424,7 +1424,7 @@ pub fn resync_target( if rs.is_none() { let reset_before_date = reset_before_date.unwrap_or(OffsetDateTime::UNIX_EPOCH); - if !reset_id.is_empty() && mod_time < reset_before_date { + if !reset_id.is_empty() && mod_time <= reset_before_date { dec.replicate = true; return dec; } @@ -1447,13 +1447,13 @@ pub fn resync_target( return dec; } - let new_reset = parts[0] == reset_id; + let new_reset = parts[1] != reset_id; if !new_reset && status == ReplicationStatusType::Completed { return dec; } - dec.replicate = new_reset && mod_time < reset_before_date; + dec.replicate = new_reset && mod_time <= reset_before_date; dec } @@ -3337,10 +3337,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { rinfo.replication_status = ReplicationStatusType::Completed; if replication_action == ReplicationAction::None { if self.op_type == ReplicationType::ExistingObject - && object_info.mod_time - > oi.last_modified - .map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH)) - && object_info.version_id.is_none() + && target_is_newer_than_source_null_version(&object_info, &oi) { warn!( event = EVENT_RESYNC_RUNTIME_SKIPPED, @@ -3350,7 +3347,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { object = %object, arn = %tgt_client.arn, endpoint = %tgt_client.to_url(), - reason = "newer_target_version_exists", + reason = "target_newer_than_source_null_version", "Skipping replication because newer target version exists" ); send_local_event(EventArgs { @@ -3558,7 +3555,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { replication_status: self.replication_status.clone(), version_purge_status_internal: self.version_purge_status_internal.clone(), version_purge_status: self.version_purge_status.clone(), - delete_marker: true, + delete_marker: self.delete_marker, checksum: self.checksum.clone(), ..Default::default() } @@ -3983,14 +3980,15 @@ async fn replicate_object_with_multipart(ctx: MultipartR Ok(()) } -fn get_replication_action(oi1: &ObjectInfo, oi2: &HeadObjectOutput, op_type: ReplicationType) -> ReplicationAction { - if op_type == ReplicationType::ExistingObject - && oi1.mod_time - > oi2 - .last_modified - .map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH)) +fn target_is_newer_than_source_null_version(oi1: &ObjectInfo, oi2: &HeadObjectOutput) -> bool { + oi2.last_modified + .map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH)) + .is_some_and(|target_mod_time| target_mod_time > oi1.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)) && oi1.version_id.is_none() - { +} + +fn get_replication_action(oi1: &ObjectInfo, oi2: &HeadObjectOutput, op_type: ReplicationType) -> ReplicationAction { + if op_type == ReplicationType::ExistingObject && target_is_newer_than_source_null_version(oi1, oi2) { return ReplicationAction::None; } @@ -4093,8 +4091,9 @@ fn get_replication_action(oi1: &ObjectInfo, oi2: &HeadObjectOutput, op_type: Rep #[cfg(test)] mod tests { use super::*; + use aws_smithy_types::DateTime; use std::collections::HashMap; - use time::OffsetDateTime; + use time::{Duration, OffsetDateTime}; use uuid::Uuid; #[test] @@ -4356,6 +4355,107 @@ mod tests { ); } + #[test] + fn test_get_replication_action_existing_object_source_newer_null_version_requires_replication() { + let source = ObjectInfo { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(20)), + version_id: None, + ..Default::default() + }; + let target = HeadObjectOutput::builder().last_modified(DateTime::from_secs(10)).build(); + + assert_eq!( + get_replication_action(&source, &target, ReplicationType::ExistingObject), + ReplicationAction::All, + "a newer source null version must not be skipped during existing-object replication" + ); + } + + #[test] + fn test_get_replication_action_existing_object_target_newer_null_version_skips() { + let source = ObjectInfo { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), + version_id: None, + ..Default::default() + }; + let target = HeadObjectOutput::builder().last_modified(DateTime::from_secs(20)).build(); + + assert_eq!( + get_replication_action(&source, &target, ReplicationType::ExistingObject), + ReplicationAction::None, + "a newer target null-version object should not be overwritten by existing-object replication" + ); + } + + #[test] + fn test_resync_target_includes_object_at_reset_before_boundary() { + let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); + let oi = ObjectInfo { + mod_time: Some(reset_before), + ..Default::default() + }; + + let decision = resync_target(&oi, "arn:target", "reset-1", Some(reset_before), ReplicationStatusType::Completed); + + assert!( + decision.replicate, + "objects whose mod_time equals reset_before must be included in the reset window" + ); + } + + #[test] + fn test_resync_target_replicates_when_reset_id_changes() { + let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); + let oi = ObjectInfo { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), + user_defined: Arc::new(HashMap::from([( + target_reset_header("arn:target"), + "1970-01-01T00:00:20Z;old-reset".to_string(), + )])), + ..Default::default() + }; + + let decision = resync_target(&oi, "arn:target", "new-reset", Some(reset_before), ReplicationStatusType::Completed); + + assert!(decision.replicate, "a new reset id must resync objects marked by an older reset"); + } + + #[test] + fn test_resync_target_skips_completed_object_for_same_reset_id() { + let reset_before = OffsetDateTime::UNIX_EPOCH + Duration::seconds(30); + let oi = ObjectInfo { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), + user_defined: Arc::new(HashMap::from([( + target_reset_header("arn:target"), + "1970-01-01T00:00:20Z;same-reset".to_string(), + )])), + ..Default::default() + }; + + let decision = resync_target(&oi, "arn:target", "same-reset", Some(reset_before), ReplicationStatusType::Completed); + + assert!(!decision.replicate, "the same completed reset id must not resync again"); + } + + #[test] + fn test_replicate_object_info_to_object_info_preserves_delete_marker_flag() { + let live = ReplicateObjectInfo { + bucket: "source".to_string(), + name: "object".to_string(), + delete_marker: false, + ..Default::default() + }; + let delete_marker = ReplicateObjectInfo { + bucket: "source".to_string(), + name: "object".to_string(), + delete_marker: true, + ..Default::default() + }; + + assert!(!live.to_object_info().delete_marker); + assert!(delete_marker.to_object_info().delete_marker); + } + #[test] fn test_delete_replication_object_opts_marks_replica_deletes() { let dobj = ObjectToDelete { diff --git a/crates/filemeta/src/replication.rs b/crates/filemeta/src/replication.rs index 32dd8a644..ff487858e 100644 --- a/crates/filemeta/src/replication.rs +++ b/crates/filemeta/src/replication.rs @@ -300,11 +300,18 @@ impl ReplicationState { /// Returns replicatedInfos struct initialized with the previous state of replication pub fn target_state(&self, arn: &str) -> ReplicatedTargetInfo { + let resync_timestamp = self + .reset_statuses_map + .get(&target_reset_header(arn)) + .or_else(|| self.reset_statuses_map.get(arn)) + .cloned() + .unwrap_or_default(); + ReplicatedTargetInfo { arn: arn.to_string(), prev_replication_status: self.targets.get(arn).cloned().unwrap_or_default(), version_purge_status: self.purge_targets.get(arn).cloned().unwrap_or_default(), - resync_timestamp: self.reset_statuses_map.get(arn).cloned().unwrap_or_default(), + resync_timestamp, ..Default::default() } } @@ -862,6 +869,28 @@ pub fn version_purge_statuses_map(s: &str) -> HashMap) -> Option { + let mut result = String::new(); + for (arn, status) in targets { + if arn.is_empty() || status.is_empty() { + continue; + } + result.push_str(&format!("{arn}={status};")); + } + if result.is_empty() { None } else { Some(result) } +} + +fn version_purge_statuses_string(targets: &HashMap) -> Option { + let mut result = String::new(); + for (arn, status) in targets { + if arn.is_empty() || status.is_empty() { + continue; + } + result.push_str(&format!("{arn}={status};")); + } + if result.is_empty() { None } else { Some(result) } +} + pub fn get_replication_state(rinfos: &ReplicatedInfos, prev_state: &ReplicationState, _vid: Option) -> ReplicationState { let reset_status_map: Vec<(String, String)> = rinfos .targets @@ -870,8 +899,18 @@ pub fn get_replication_state(rinfos: &ReplicatedInfos, prev_state: &ReplicationS .map(|t| (target_reset_header(t.arn.as_str()), t.resync_timestamp.clone())) .collect(); - let repl_statuses = rinfos.replication_status_internal(); - let vpurge_statuses = rinfos.version_purge_status_internal(); + let mut targets = prev_state.targets.clone(); + for (arn, status) in replication_statuses_map(&rinfos.replication_status_internal().unwrap_or_default()) { + targets.insert(arn, status); + } + + let mut purge_targets = prev_state.purge_targets.clone(); + for (arn, status) in version_purge_statuses_map(&rinfos.version_purge_status_internal().unwrap_or_default()) { + purge_targets.insert(arn, status); + } + + let repl_statuses = replication_statuses_string(&targets); + let vpurge_statuses = version_purge_statuses_string(&purge_targets); let mut reset_statuses_map = prev_state.reset_statuses_map.clone(); for (key, value) in reset_status_map { @@ -883,10 +922,10 @@ pub fn get_replication_state(rinfos: &ReplicatedInfos, prev_state: &ReplicationS reset_statuses_map, replica_timestamp: prev_state.replica_timestamp, replica_status: prev_state.replica_status.clone(), - targets: replication_statuses_map(&repl_statuses.clone().unwrap_or_default()), + targets, replication_status_internal: repl_statuses, replication_timestamp: rinfos.replication_timestamp, - purge_targets: version_purge_statuses_map(&vpurge_statuses.clone().unwrap_or_default()), + purge_targets, version_purge_status_internal: vpurge_statuses, ..Default::default() @@ -934,3 +973,47 @@ impl Default for ResyncDecision { Self::new() } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn target_state_reads_resync_timestamp_from_target_reset_header_key() { + let arn = "arn:rustfs:replication:us-east-1:target:bucket"; + let timestamp = "2026-06-30T00:00:00Z;reset-1".to_string(); + let mut state = ReplicationState::default(); + state.reset_statuses_map.insert(target_reset_header(arn), timestamp.clone()); + + let target_state = state.target_state(arn); + + assert_eq!(target_state.resync_timestamp, timestamp); + } + + #[test] + fn get_replication_state_preserves_untouched_target_statuses() { + let target_a = "arn:target:a".to_string(); + let target_b = "arn:target:b".to_string(); + let mut prev_state = ReplicationState::default(); + prev_state.targets.insert(target_a.clone(), ReplicationStatusType::Failed); + prev_state.targets.insert(target_b.clone(), ReplicationStatusType::Completed); + + let rinfos = ReplicatedInfos { + replication_timestamp: None, + targets: vec![ReplicatedTargetInfo { + arn: target_a.clone(), + replication_status: ReplicationStatusType::Completed, + ..Default::default() + }], + }; + + let state = get_replication_state(&rinfos, &prev_state, None); + + assert_eq!(state.targets.get(&target_a), Some(&ReplicationStatusType::Completed)); + assert_eq!(state.targets.get(&target_b), Some(&ReplicationStatusType::Completed)); + assert_eq!( + replication_statuses_map(&state.replication_status_internal.unwrap_or_default()).get(&target_b), + Some(&ReplicationStatusType::Completed) + ); + } +} diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index 792f52db5..f6f0d314a 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -20,7 +20,7 @@ use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE; use crate::admin::storage_api::bucket::metadata_sys; use crate::admin::storage_api::bucket::metadata_sys::get_replication_config; use crate::admin::storage_api::bucket::replication::BucketStats; -use crate::admin::storage_api::bucket::target::BucketTarget; +use crate::admin::storage_api::bucket::target::{BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat}; use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys}; use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions}; use crate::admin::storage_api::error::StorageError; @@ -36,7 +36,10 @@ use rustfs_credentials::Credentials; use rustfs_policy::policy::action::{Action, AdminAction}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; +use serde::Deserialize; use std::collections::HashMap; +use std::time::Duration; +use time::OffsetDateTime; use tracing::{debug, error, info, warn}; use url::Host; @@ -68,6 +71,139 @@ fn map_bucket_target_error(err: BucketTargetError) -> S3Error { } } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct RemoteTargetCredentialsRequest { + #[serde(rename = "accessKey")] + access_key: String, + #[serde(rename = "secretKey")] + secret_key: String, + session_token: Option, + expiration: Option>, +} + +impl From for TargetCredentials { + fn from(value: RemoteTargetCredentialsRequest) -> Self { + Self { + access_key: value.access_key, + secret_key: value.secret_key, + session_token: value.session_token, + expiration: value.expiration, + } + } +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct RemoteTargetRequest { + #[serde(rename = "sourcebucket", default)] + source_bucket: String, + endpoint: String, + credentials: RemoteTargetCredentialsRequest, + #[serde(rename = "targetbucket")] + target_bucket: String, + #[serde(default)] + secure: bool, + #[serde(default)] + path: String, + #[serde(default)] + api: String, + #[serde(default)] + arn: String, + #[serde(rename = "type")] + target_type: BucketTargetType, + #[serde(default)] + region: String, + #[serde(alias = "bandwidth", default)] + bandwidth_limit: i64, + #[serde(rename = "replicationSync", default)] + replication_sync: bool, + #[serde(default)] + storage_class: String, + #[serde(rename = "skipTlsVerify", default)] + skip_tls_verify: bool, + #[serde(rename = "caCertPem", default)] + ca_cert_pem: String, + #[serde(rename = "healthCheckDuration", default)] + health_check_duration: u64, + #[serde(rename = "disableProxy", default)] + disable_proxy: bool, + #[serde(rename = "resetBeforeDate", with = "time::serde::rfc3339::option", default)] + reset_before_date: Option, + #[serde(default)] + reset_id: String, + #[serde(rename = "totalDowntime", default)] + total_downtime: u64, + #[serde(rename = "lastOnline", with = "time::serde::rfc3339::option", default)] + last_online: Option, + #[serde(rename = "isOnline", default)] + online: bool, + #[serde(default)] + latency: LatencyStat, + #[serde(default)] + deployment_id: String, + #[serde(default)] + edge: bool, + #[serde(rename = "edgeSyncBeforeExpiry", default)] + edge_sync_before_expiry: bool, + #[serde(rename = "offlineCount", default)] + offline_count: u64, +} + +impl RemoteTargetRequest { + fn into_bucket_target(self) -> S3Result { + if self.endpoint.trim().is_empty() { + return Err(s3_error!(InvalidRequest, "endpoint is required")); + } + + if self.target_bucket.trim().is_empty() { + return Err(s3_error!(InvalidRequest, "targetbucket is required")); + } + + if !self.target_type.is_valid() { + return Err(s3_error!(InvalidRequest, "type is invalid")); + } + + if self.credentials.access_key.trim().is_empty() { + return Err(s3_error!(InvalidRequest, "credentials.accessKey is required")); + } + + if self.credentials.secret_key.trim().is_empty() { + return Err(s3_error!(InvalidRequest, "credentials.secretKey is required")); + } + + Ok(BucketTarget { + source_bucket: self.source_bucket, + endpoint: self.endpoint, + credentials: Some(self.credentials.into()), + target_bucket: self.target_bucket, + secure: self.secure, + path: self.path, + api: self.api, + arn: self.arn, + target_type: self.target_type, + region: self.region, + bandwidth_limit: self.bandwidth_limit, + replication_sync: self.replication_sync, + storage_class: self.storage_class, + skip_tls_verify: self.skip_tls_verify, + ca_cert_pem: self.ca_cert_pem, + health_check_duration: Duration::from_secs(self.health_check_duration), + disable_proxy: self.disable_proxy, + reset_before_date: self.reset_before_date, + reset_id: self.reset_id, + total_downtime: Duration::from_secs(self.total_downtime), + last_online: self.last_online, + online: self.online, + latency: self.latency, + deployment_id: self.deployment_id, + edge: self.edge, + edge_sync_before_expiry: self.edge_sync_before_expiry, + offline_count: self.offline_count, + }) + } +} + fn validate_remote_target_tls_settings(remote_target: &BucketTarget) -> S3Result<()> { let has_custom_ca = !remote_target.ca_cert_pem.trim().is_empty(); @@ -227,10 +363,12 @@ impl Operation for SetRemoteTargetHandler { } }; - let mut remote_target: BucketTarget = serde_json::from_slice(&body).map_err(|e| { - error!("Failed to parse BucketTarget from body: {}", e); - ApiError::other(e) - })?; + let mut remote_target = serde_json::from_slice::(&body) + .map_err(|e| { + error!("Failed to parse remote target request body: {}", e); + S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid remote target request: {e}")) + })? + .into_bucket_target()?; validate_remote_target_tls_settings(&remote_target)?; let Ok(target_url) = remote_target.url() else { @@ -448,10 +586,23 @@ impl Operation for RemoveRemoteTargetHandler { #[cfg(test)] mod tests { - use super::{extract_query_params, validate_remote_target_tls_settings}; + use super::{RemoteTargetRequest, extract_query_params, validate_remote_target_tls_settings}; use crate::admin::storage_api::bucket::target::BucketTarget; use http::Uri; + fn valid_remote_target_request() -> serde_json::Value { + serde_json::json!({ + "endpoint": "192.168.1.10:9000", + "credentials": { + "accessKey": "access", + "secretKey": "secret" + }, + "targetbucket": "target", + "secure": true, + "type": "replication" + }) + } + #[test] fn test_extract_query_params_decodes_percent_encoded_values() { let uri: Uri = "/rustfs/admin/v3/list-remote-targets?bucket=foo%2Fbar&flag=a+b" @@ -512,4 +663,57 @@ mod tests { }) .expect("HTTPS targets should allow skipTlsVerify when no custom CA is configured"); } + + #[test] + fn remote_target_request_rejects_unknown_fields() { + let mut request = valid_remote_target_request(); + request["unexpected"] = serde_json::json!(true); + + let err = serde_json::from_value::(request) + .expect_err("remote target request should reject unknown fields"); + + assert!(err.to_string().contains("unknown field")); + } + + #[test] + fn remote_target_request_rejects_missing_credentials() { + let mut request = valid_remote_target_request(); + request + .as_object_mut() + .expect("request should be an object") + .remove("credentials"); + + let err = + serde_json::from_value::(request).expect_err("remote target request should require credentials"); + + assert!(err.to_string().contains("missing field")); + } + + #[test] + fn remote_target_request_rejects_empty_secret_key() { + let mut request = valid_remote_target_request(); + request["credentials"]["secretKey"] = serde_json::json!(""); + let request: RemoteTargetRequest = + serde_json::from_value(request).expect("request should deserialize before semantic validation"); + + let err = match request.into_bucket_target() { + Ok(_) => panic!("empty secret key should fail semantic validation"), + Err(err) => err, + }; + + assert!(err.to_string().contains("credentials.secretKey is required")); + } + + #[test] + fn remote_target_request_converts_to_bucket_target() { + let target = serde_json::from_value::(valid_remote_target_request()) + .expect("request should deserialize") + .into_bucket_target() + .expect("request should pass semantic validation"); + + assert_eq!(target.endpoint, "192.168.1.10:9000"); + assert_eq!(target.target_bucket, "target"); + assert!(target.secure); + assert_eq!(target.credentials.expect("credentials should be present").access_key, "access"); + } } diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 493f3f1ec..bac70e1c3 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -1532,15 +1532,19 @@ async fn authorize_replication_extension_request(req: &mut S3Request, ext_ } })?; - let action = match ext_req.route { - ReplicationExtRoute::MetricsV1 | ReplicationExtRoute::MetricsV2 | ReplicationExtRoute::Check => { + authorize_request(req, replication_extension_policy_action(ext_req.route)).await +} + +fn replication_extension_policy_action(route: ReplicationExtRoute) -> Action { + match route { + ReplicationExtRoute::MetricsV1 | ReplicationExtRoute::MetricsV2 => { Action::S3Action(S3Action::GetReplicationConfigurationAction) } + ReplicationExtRoute::Check => Action::S3Action(S3Action::PutReplicationConfigurationAction), ReplicationExtRoute::ResetStart | ReplicationExtRoute::ResetStatus => { Action::S3Action(S3Action::ResetBucketReplicationStateAction) } - }; - authorize_request(req, action).await + } } fn parse_reset_start_target(uri: &Uri) -> S3Result { @@ -1586,17 +1590,17 @@ fn collect_resettable_replication_target_arns(config: &s3s::dto::ReplicationConf continue; } - let arn = if config.role.is_empty() { - rule.destination.bucket.clone() + let arn = if config.role.trim().is_empty() { + rule.destination.bucket.trim().to_string() } else { - config.role.clone() + config.role.trim().to_string() }; if seen.insert(arn.clone()) { arns.push(arn); } - if !config.role.is_empty() { + if !config.role.trim().is_empty() { break; } } @@ -1775,26 +1779,27 @@ fn validate_replication_check_config_targets( .map(|target| target.arn.as_str()) .collect::>(); + let role = config.role.trim(); + if !role.is_empty() { + if !configured_arns.contains(role) { + return Err(s3_error!(InvalidRequest, "replication config has stale target {role}")); + } + return Ok(()); + } + for rule in &config.rules { if rule.status == s3s::dto::ReplicationRuleStatus::from_static(s3s::dto::ReplicationRuleStatus::DISABLED) { continue; } - let configured_arn = if config.role.is_empty() { - rule.destination.bucket.as_str() - } else { - config.role.as_str() - }; - - if configured_arns.contains(configured_arn) { - continue; + let configured_arn = rule.destination.bucket.trim(); + if !configured_arn.is_empty() && !configured_arns.contains(configured_arn) { + let rule_id = rule.id.as_deref().unwrap_or(""); + return Err(s3_error!( + InvalidRequest, + "replication rule {rule_id} references stale target {configured_arn}" + )); } - - return Err(s3_error!( - InvalidRequest, - "replication config with rule ID {} has a stale target", - rule.id.clone().unwrap_or_default() - )); } Ok(()) @@ -2630,6 +2635,22 @@ mod tests { assert!(parse_replication_extension_request(&Method::PUT, &wrong_method_status).is_none()); } + #[test] + fn replication_extension_policy_action_uses_write_permission_for_active_check() { + assert_eq!( + replication_extension_policy_action(ReplicationExtRoute::MetricsV1), + Action::S3Action(S3Action::GetReplicationConfigurationAction) + ); + assert_eq!( + replication_extension_policy_action(ReplicationExtRoute::Check), + Action::S3Action(S3Action::PutReplicationConfigurationAction) + ); + assert_eq!( + replication_extension_policy_action(ReplicationExtRoute::ResetStart), + Action::S3Action(S3Action::ResetBucketReplicationStateAction) + ); + } + #[test] fn parse_reset_start_target_defaults_reset_before_and_supports_older_than() { let no_window: Uri = "/demo-bucket?replication-reset".parse().expect("uri should parse"); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index c24732226..b75a95272 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -319,6 +319,7 @@ pub(crate) mod target { pub(crate) type BucketTargetType = super::ecstore_bucket::target::BucketTargetType; pub(crate) type BucketTargets = super::ecstore_bucket::target::BucketTargets; pub(crate) type Credentials = super::ecstore_bucket::target::Credentials; + pub(crate) type LatencyStat = super::ecstore_bucket::target::LatencyStat; } pub(crate) mod ecstore_utils { diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 4754586b3..d1f331965 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -246,15 +246,14 @@ fn notify_bucket_metadata_reload( }); } -fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet { +fn active_replication_rule_destination_arns(config: &ReplicationConfiguration) -> HashSet { let mut arns = HashSet::new(); - if !config.role.trim().is_empty() { - arns.insert(config.role.clone()); - return arns; - } - for rule in &config.rules { + if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) { + continue; + } + let arn = rule.destination.bucket.trim(); if !arn.is_empty() { arns.insert(arn.to_string()); @@ -264,6 +263,17 @@ fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet arns } +fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet { + let role = config.role.trim(); + if !role.is_empty() { + let mut arns = HashSet::new(); + arns.insert(role.to_string()); + return arns; + } + + active_replication_rule_destination_arns(config) +} + fn validate_replication_config_targets(targets: &BucketTargets, config: &ReplicationConfiguration) -> S3Result<()> { let configured_arns = targets .targets @@ -272,28 +282,21 @@ fn validate_replication_config_targets(targets: &BucketTargets, config: &Replica .map(|target| target.arn.as_str()) .collect::>(); - for rule in &config.rules { - if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) { - continue; - } - - let configured_arn = if config.role.trim().is_empty() { - rule.destination.bucket.trim() - } else { - config.role.trim() - }; - - if !configured_arn.is_empty() && configured_arns.contains(configured_arn) { - continue; - } - + let role = config.role.trim(); + let destination_arns = active_replication_rule_destination_arns(config); + if !role.is_empty() && destination_arns.len() > 1 { return Err(s3_error!( InvalidRequest, - "replication config with rule ID {} has a stale target", - rule.id.clone().unwrap_or_default() + "replication config with Role cannot define multiple destination targets" )); } + for configured_arn in replication_target_arns(config) { + if !configured_arns.contains(configured_arn.as_str()) { + return Err(s3_error!(InvalidRequest, "replication config has a stale target")); + } + } + Ok(()) } @@ -317,41 +320,81 @@ async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationCo validate_replication_config_targets(&targets, config) } -async fn remove_replication_targets_for_config(bucket: &str, config: &ReplicationConfiguration) -> S3Result<()> { +async fn replication_targets_without_config_targets( + bucket: &str, + config: &ReplicationConfiguration, +) -> S3Result> { let target_arns = replication_target_arns(config); if target_arns.is_empty() { - return Ok(()); + return Ok(None); } let mut targets = match metadata_sys::get_bucket_targets_config(bucket).await { Ok(targets) => targets, Err(StorageError::ConfigNotFound) => { BucketTargetSys::get().update_all_targets(bucket, None).await; - return Ok(()); + return Ok(None); } Err(err) => return Err(ApiError::from(err).into()), }; + let removed = remove_replication_targets_from_config_targets(&mut targets, &target_arns); + if removed == 0 { + return Ok(None); + } + + Ok(Some((targets, removed))) +} + +fn remove_replication_targets_from_config_targets(targets: &mut BucketTargets, target_arns: &HashSet) -> usize { let original_len = targets.targets.len(); targets.targets.retain(|target| { target.target_type != BucketTargetType::ReplicationService || !target_arns.contains(target.arn.as_str()) }); - if targets.targets.len() == original_len { - return Ok(()); - } + original_len - targets.targets.len() +} - let removed = original_len - targets.targets.len(); +async fn write_replication_targets_after_config_delete(bucket: &str, targets: &BucketTargets, removed: usize) -> S3Result<()> { let json_targets = serde_json::to_vec(&targets).map_err(to_internal_error)?; metadata_sys::update(bucket, BUCKET_TARGETS_FILE, json_targets) .await .map_err(ApiError::from)?; - BucketTargetSys::get().update_all_targets(bucket, Some(&targets)).await; + BucketTargetSys::get().update_all_targets(bucket, Some(targets)).await; info!(bucket = %bucket, removed, "removed replication remote targets referenced by deleted bucket replication config"); Ok(()) } +async fn restore_replication_config_after_target_cleanup_failure( + bucket: &str, + config: &ReplicationConfiguration, + cleanup_err: S3Error, +) -> S3Error { + match serialize(config) { + Ok(data) => { + if let Err(restore_err) = metadata_sys::update(bucket, BUCKET_REPLICATION_CONFIG, data).await { + error!( + bucket = %bucket, + error = ?restore_err, + cleanup_error = ?cleanup_err, + "failed to restore bucket replication config after target cleanup failure" + ); + } + } + Err(restore_err) => { + error!( + bucket = %bucket, + error = ?restore_err, + cleanup_error = ?cleanup_err, + "failed to serialize bucket replication config for restore after target cleanup failure" + ); + } + } + + cleanup_err +} + fn versioning_configuration_has_object_lock_incompatible_settings(config: &VersioningConfiguration) -> bool { config.suspended() || config.exclude_folders.unwrap_or(false) @@ -1129,14 +1172,22 @@ impl DefaultBucketUsecase { Err(StorageError::ConfigNotFound) => None, Err(err) => return Err(ApiError::from(err).into()), }; + let updated_targets = if let Some(config) = replication_config.as_ref() { + replication_targets_without_config_targets(&bucket, config).await? + } else { + None + }; metadata_sys::delete(&bucket, BUCKET_REPLICATION_CONFIG) .await .map_err(ApiError::from)?; - if let Some(config) = replication_config.as_ref() - && let Err(err) = remove_replication_targets_for_config(&bucket, config).await + if let Some((targets, removed)) = updated_targets + && let Err(err) = write_replication_targets_after_config_delete(&bucket, &targets, removed).await { - warn!(bucket = %bucket, error = ?err, "failed to remove replication targets referenced by deleted bucket replication config"); + if let Some(config) = replication_config.as_ref() { + return Err(restore_replication_config_after_target_cleanup_failure(&bucket, config, err).await); + } + return Err(err); } let item = sr_bucket_meta_item(bucket.clone(), "replication-config"); @@ -2319,7 +2370,7 @@ mod tests { let role = "arn:rustfs:replication:us-east-1:source:bucket"; let destination = "arn:rustfs:replication:us-east-1:target:bucket"; let config = ReplicationConfiguration { - role: role.to_string(), + role: format!(" {role} "), rules: vec![replication_rule_for_target(destination)], }; @@ -2386,13 +2437,44 @@ mod tests { let arn = "arn:rustfs:replication:us-east-1:role-target:bucket"; let targets = replication_targets_with_arn(&[arn]); let config = ReplicationConfiguration { - role: arn.to_string(), + role: format!(" {arn} "), rules: vec![replication_rule_for_target("arn:rustfs:replication:us-east-1:ignored:bucket")], }; validate_replication_config_targets(&targets, &config).expect("matching role ARN should pass validation"); } + #[test] + fn validate_replication_config_targets_rejects_role_with_multiple_destinations() { + let role = "arn:rustfs:replication:us-east-1:role-target:bucket"; + let targets = replication_targets_with_arn(&[role]); + let config = ReplicationConfiguration { + role: role.to_string(), + rules: vec![ + replication_rule_for_target("arn:rustfs:replication:us-east-1:target-a:bucket"), + replication_rule_for_target("arn:rustfs:replication:us-east-1:target-b:bucket"), + ], + }; + + let err = validate_replication_config_targets(&targets, &config) + .expect_err("role plus multiple destinations should be rejected"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn validate_replication_config_targets_trims_destination_arns() { + let arn = "arn:rustfs:replication:us-east-1:target:bucket"; + let targets = replication_targets_with_arn(&[arn]); + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule_for_target( + " arn:rustfs:replication:us-east-1:target:bucket ", + )], + }; + + validate_replication_config_targets(&targets, &config).expect("trimmed destination ARN should match configured target"); + } + #[test] fn validate_replication_config_targets_ignores_disabled_rules() { let targets = replication_targets_with_arn(&[]); @@ -2406,6 +2488,45 @@ mod tests { validate_replication_config_targets(&targets, &config).expect("disabled rules should not require live targets"); } + #[test] + fn remove_replication_targets_from_config_targets_only_removes_referenced_replication_targets() { + let removed_arn = "arn:rustfs:replication:us-east-1:removed:bucket"; + let kept_replication_arn = "arn:rustfs:replication:us-east-1:kept:bucket"; + let kept_ilm_arn = "arn:rustfs:ilm:us-east-1:kept:bucket"; + let mut targets = BucketTargets { + targets: vec![ + BucketTarget { + arn: removed_arn.to_string(), + target_type: BucketTargetType::ReplicationService, + ..Default::default() + }, + BucketTarget { + arn: kept_replication_arn.to_string(), + target_type: BucketTargetType::ReplicationService, + ..Default::default() + }, + BucketTarget { + arn: kept_ilm_arn.to_string(), + target_type: BucketTargetType::IlmService, + ..Default::default() + }, + ], + }; + let target_arns = HashSet::from([removed_arn.to_string(), kept_ilm_arn.to_string()]); + + let removed = remove_replication_targets_from_config_targets(&mut targets, &target_arns); + + assert_eq!(removed, 1); + let remaining_arns = targets + .targets + .iter() + .map(|target| target.arn.as_str()) + .collect::>(); + assert!(!remaining_arns.contains(removed_arn)); + assert!(remaining_arns.contains(kept_replication_arn)); + assert!(remaining_arns.contains(kept_ilm_arn)); + } + #[test] fn versioning_configuration_has_object_lock_incompatible_settings_rejects_suspended() { let config = VersioningConfiguration {