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 <wlywly0735@126.com>
Co-authored-by: houseme <housemecn@gmail.com>

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;

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 {

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)?;