From f9475e10dc85e5c11e158d44cb91d84f79774324 Mon Sep 17 00:00:00 2001 From: weisd Date: Sun, 24 May 2026 22:24:00 +0800 Subject: [PATCH] fix(replication): preserve multipart pending state (#3058) * Preserve multipart replication recovery state Multipart uploads previously only scheduled replication after completion, leaving no persisted pending state for scanner recovery if the initial async work was lost. Persist the same pending replication metadata during multipart initialization and let completion evaluate the object metadata that was actually stored. The scanner heal path also treated ordinary pending objects as delete-replication candidates. Restrict that path to delete markers and version purge state so pending objects remain eligible for object replication heal. Constraint: Bucket replication recovery depends on persisted object metadata after the async queue is unavailable. Rejected: Rely only on immediate completion-time scheduling | it cannot recover after process restart or worker loss. Confidence: high Scope-risk: moderate Directive: Keep multipart upload initialization aligned with single PUT replication metadata semantics. Tested: cargo fmt --all --check Tested: cargo clippy --workspace --all-features --all-targets -- -D warnings Tested: LC_ALL=en_US.UTF-8 LANG=en_US.UTF-8 make pre-commit Tested: Runtime replication outage check confirmed multipart xl.meta stores PENDING status and timestamp. * fix(replication): preserve version purge scanner state Role-derived replication configs need target-scoped status strings before scanner heal can build per-target purge status. The duplicated replication-status assignment left version purge status unset, so scanner recovery could lose the target-level purge state. Constraint: Scanner heal derives per-target purge decisions from version_purge_status_internal. Rejected: Leave the duplicate as a harmless cleanup | it changes recovery behavior for role-only configs with version purge state. Confidence: high Scope-risk: narrow Directive: Keep role-derived replication and version purge internal status mapping symmetric. Tested: cargo fmt --all --check Tested: cargo test -p rustfs-ecstore heal -- --nocapture --------- Co-authored-by: wly Co-authored-by: houseme --- .../replication/replication_resyncer.rs | 30 +++++++++++++++++++ crates/utils/src/os/fs_type.rs | 1 + rustfs/src/app/multipart_usecase.rs | 22 ++++++++++++-- 3 files changed, 50 insertions(+), 3 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 91308688f..305dcacbc 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -939,6 +939,10 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC if let Some(rc) = rcfg.config.as_ref() && !rc.role.is_empty() { + if !oi.version_purge_status.is_empty() { + oi.version_purge_status_internal = Some(format!("{}={};", rc.role, oi.version_purge_status.as_str())); + } + if !oi.replication_status.is_empty() { oi.replication_status_internal = Some(format!("{}={};", rc.role, oi.replication_status.as_str())); } @@ -4077,6 +4081,32 @@ mod tests { ); } + #[tokio::test] + async fn test_get_heal_replicate_object_info_maps_version_purge_status_for_role() { + let role = "arn:rustfs:replication::target:bucket"; + let oi = ObjectInfo { + bucket: "test-bucket".to_string(), + name: "key".to_string(), + delete_marker: false, + version_purge_status: VersionPurgeStatusType::Pending, + version_id: Some(Uuid::nil()), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + let rcfg = ReplicationConfig::new( + Some(ReplicationConfiguration { + role: role.to_string(), + rules: vec![], + }), + None, + ); + let roi = get_heal_replicate_object_info(&oi, &rcfg).await; + + assert_eq!(roi.replication_status_internal, None); + assert_eq!(roi.version_purge_status_internal.as_deref(), Some(format!("{role}=PENDING;").as_str())); + assert_eq!(roi.target_purge_statuses.get(role), Some(&VersionPurgeStatusType::Pending)); + } + #[tokio::test] async fn test_cancel_marks_only_matching_bucket_target_token() { let resyncer = ReplicationResyncer::new().await; diff --git a/crates/utils/src/os/fs_type.rs b/crates/utils/src/os/fs_type.rs index 2d162b130..9875f80c2 100644 --- a/crates/utils/src/os/fs_type.rs +++ b/crates/utils/src/os/fs_type.rs @@ -24,6 +24,7 @@ /// "ef51" => "EXT2OLD", /// "2fc12fc1" => "zfs", /// "53464846" => "wslfs", +#[allow(dead_code)] pub(crate) fn get_fs_type(fs_type: u64) -> &'static str { // Magic numbers for various filesystems. match fs_type { diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 82f8e03bf..2b8c064fc 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -56,13 +56,16 @@ use rustfs_s3_ops::S3Operation; use rustfs_targets::EventName; use rustfs_utils::CompressionAlgorithm; use rustfs_utils::http::{ - AMZ_CHECKSUM_TYPE, get_source_scheme, + AMZ_CHECKSUM_TYPE, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, get_source_scheme, headers::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING}, + insert_str, }; use s3s::dto::*; use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE}; use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; -use std::collections::{HashMap, HashSet}; +#[cfg(test)] +use std::collections::HashMap; +use std::collections::HashSet; use std::str::FromStr; use std::sync::Arc; use tokio::sync::RwLock; @@ -454,7 +457,7 @@ impl DefaultMultipartUsecase { version_id: mpu_version, ..Default::default() }; - let mt2 = HashMap::new(); + let mt2 = obj_info.user_defined.clone(); let replicate_options = get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone()); let dsc = must_replicate(&bucket, &key, replicate_options).await; @@ -574,10 +577,23 @@ impl DefaultMultipartUsecase { ); } + let mt2 = metadata.clone(); let mut opts: ObjectOptions = put_opts(&bucket, &key, version_id, &req.headers, metadata) .await .map_err(ApiError::from)?; + let replicate_options = + get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone()); + let dsc = must_replicate(&bucket, &key, replicate_options).await; + if dsc.replicate_any() { + insert_str(&mut opts.user_defined, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string()); + insert_str( + &mut opts.user_defined, + SUFFIX_REPLICATION_STATUS, + dsc.pending_status().unwrap_or_default(), + ); + } + let current_opts: ObjectOptions = get_opts(&bucket, &key, opts.version_id.clone(), None, &req.headers) .await .map_err(ApiError::from)?;