From 80cc3b1fcfc5dec4e48da5470c86b182dc5a282e Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 8 Jul 2026 15:01:33 +0800 Subject: [PATCH] fix(replication): preserve SSE-C checksum state (#4410) * fix(replication): preserve ssec checksum state * fix(replication): route ssec checks through boundary --- .../bucket/replication/replication_pool.rs | 57 ++++++++++++++----- .../replication/replication_resyncer.rs | 27 ++++++++- .../replication_target_boundary.rs | 6 +- 3 files changed, 72 insertions(+), 18 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 1161952c5..68c22e0c2 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -42,7 +42,7 @@ use super::replication_state::ReplicationStats; use super::replication_storage_boundary::{ ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationDeletedObject, ReplicationObjectIO, ReplicationStorage, }; -use super::replication_target_boundary::ReplicationTargetStore; +use super::replication_target_boundary::{ReplicationTargetStore, replication_object_is_ssec_encrypted}; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; use lazy_static::lazy_static; @@ -1332,6 +1332,21 @@ pub(crate) async fn schedule_replication( dsc: ReplicateDecision, op_type: ReplicationType, ) { + let synchronous = dsc.is_synchronous(); + let ri = replicate_object_info_from_object_info(oi, dsc, op_type); + + if synchronous { + replicate_object(ri, o).await + } else if let Some(pool) = runtime_sources::replication_pool() { + let _ = pool.queue_replica_task(ri).await; + } +} + +fn replicate_object_info_from_object_info( + oi: ObjectInfo, + dsc: ReplicateDecision, + op_type: ReplicationType, +) -> ReplicateObjectInfo { let tgt_statuses = replication_statuses_map(&oi.replication_status_internal.clone().unwrap_or_default()); let purge_statuses = version_purge_statuses_map(&oi.version_purge_status_internal.clone().unwrap_or_default()); let tm = get_str(&oi.user_defined, SUFFIX_REPLICATION_TIMESTAMP) @@ -1339,8 +1354,10 @@ pub(crate) async fn schedule_replication( let mut rstate = oi.replication_state(); rstate.replicate_decision_str = dsc.to_string(); let asz = oi.get_actual_size().unwrap_or_default(); + let ssec = replication_object_is_ssec_encrypted(&oi.user_defined); + let checksum = if ssec { oi.checksum.clone() } else { None }; - let mut ri = ReplicateObjectInfo { + ReplicateObjectInfo { name: oi.name, size: oi.size, actual_size: asz, @@ -1356,25 +1373,16 @@ pub(crate) async fn schedule_replication( replication_state: Some(rstate), op_type, - dsc: dsc.clone(), + dsc, target_statuses: tgt_statuses, target_purge_statuses: purge_statuses, replication_timestamp: tm, user_tags: (*oi.user_tags).clone(), - checksum: None, + checksum, retry_count: 0, event_type: "".to_string(), existing_obj_resync: ResyncDecision::default(), - ssec: false, - }; - - if ri.ssec { - ri.checksum = oi.checksum - } - if dsc.is_synchronous() { - replicate_object(ri, o).await - } else if let Some(pool) = runtime_sources::replication_pool() { - let _ = pool.queue_replica_task(ri).await; + ssec, } } @@ -1529,6 +1537,7 @@ mod tests { StorageListObjectVersionsInfo, StorageListObjectsV2Info, StorageNamespaceLocking, StorageObjectInfoOrErr, WalkOptions, }; use super::*; + use std::collections::HashMap; use std::fmt::{Debug, Formatter}; use std::io::Cursor; use std::sync::atomic::AtomicUsize; @@ -1857,6 +1866,26 @@ mod tests { assert_eq!(admission, ReplicationQueueAdmission::Missed); } + #[test] + fn replicate_object_info_from_object_info_preserves_ssec_checksum() { + let checksum = bytes::Bytes::from_static(b"ssec-checksum"); + let oi = ObjectInfo { + bucket: "source".to_string(), + name: "object".to_string(), + user_defined: Arc::new(HashMap::from([( + rustfs_utils::http::SSEC_ALGORITHM_HEADER.to_string(), + "AES256".to_string(), + )])), + checksum: Some(checksum.clone()), + ..Default::default() + }; + + let ri = replicate_object_info_from_object_info(oi, ReplicateDecision::default(), ReplicationType::Object); + + assert!(ri.ssec); + assert_eq!(ri.checksum, Some(checksum)); + } + #[tokio::test] async fn mrf_save_admission_waits_for_capacity_instead_of_dropping() { let (tx, mut rx) = mpsc::channel(1); diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 74deade9e..5ed81a5f6 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -48,8 +48,8 @@ use super::replication_storage_boundary::{ use super::replication_target_boundary::{ PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, replication_action_for_target_head, replication_complete_multipart_options, replication_delete_marker_purge_remove_options, replication_delete_remove_options, - replication_force_delete_remove_options, replication_put_object_header_size, replication_put_object_options, - replication_target_head_is_newer_null_version, + replication_force_delete_remove_options, replication_object_is_ssec_encrypted, replication_put_object_header_size, + replication_put_object_options, replication_target_head_is_newer_null_version, }; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; @@ -1052,7 +1052,7 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC target_statuses, target_purge_statuses, replication_timestamp: None, - ssec: false, // TODO: add ssec support + ssec: replication_object_is_ssec_encrypted(&user_defined), user_tags: (*oi.user_tags).clone(), checksum: oi.checksum.clone(), retry_count: 0, @@ -3521,6 +3521,27 @@ mod tests { ); } + #[tokio::test] + async fn test_get_heal_replicate_object_info_preserves_ssec_checksum() { + let checksum = bytes::Bytes::from_static(b"ssec-checksum"); + let oi = ObjectInfo { + bucket: "test-bucket".to_string(), + name: "key".to_string(), + user_defined: Arc::new(HashMap::from([( + rustfs_utils::http::SSEC_ALGORITHM_HEADER.to_string(), + "AES256".to_string(), + )])), + checksum: Some(checksum.clone()), + ..Default::default() + }; + let rcfg = ReplicationConfig::new(None, None); + + let roi = get_heal_replicate_object_info(&oi, &rcfg).await; + + assert!(roi.ssec); + assert_eq!(roi.checksum, Some(checksum)); + } + #[tokio::test] async fn test_get_heal_replicate_object_info_maps_version_purge_status_for_role() { let role = "arn:rustfs:replication::target:bucket"; diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index cb6885504..94180607e 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -78,6 +78,10 @@ static VALID_SSE_REPLICATION_HEADERS: &[(&str, &str)] = &[ const ERR_REPLICATION_MANAGED_SSE_UNSUPPORTED: &str = "managed SSE replication requires target encryption support"; +pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap) -> bool { + rustfs_replication::is_ssec_encrypted(user_defined) +} + pub(crate) struct ReplicationTargetStore; impl ReplicationTargetStore { @@ -103,7 +107,7 @@ pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) use rustfs_utils::http::{AMZ_CHECKSUM_TYPE, AMZ_CHECKSUM_TYPE_FULL_OBJECT}; let mut meta = HashMap::new(); - let is_ssec = rustfs_replication::is_ssec_encrypted(&object_info.user_defined); + let is_ssec = replication_object_is_ssec_encrypted(&object_info.user_defined); for (key, value) in object_info.user_defined.iter() { let has_valid_sse_header = valid_sse_replication_header(key).is_some();