fix(replication): fence stale metadata status writeback (#7083)

Co-authored-by: houseme <housemecn@gmail.com>

crates/e2e_test/src/replication_extension_test.rs

@@ -69,7 +69,7 @@ use std::net::IpAddr;
 use std::path::Path;
 use std::process::Command;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use time::{Duration as TimeDuration, OffsetDateTime};
 use tokio::fs;
 use tokio::net::TcpListener;
@@ -2008,6 +2008,87 @@ fn proxy_error_response(error: impl std::fmt::Display) -> Response<Full<bytes::B
         .expect("static proxy response must be valid")
 }
 
+#[derive(Clone)]
+struct ReplicationResponseHoldRuntime {
+    armed: Arc<AtomicBool>,
+    backend_committed: watch::Sender<bool>,
+    release: watch::Receiver<bool>,
+}
+
+impl ReplicationResponseHoldRuntime {
+    fn try_claim(&self) -> bool {
+        self.armed
+            .compare_exchange(true, false, Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+    }
+}
+
+struct ReplicationResponseHold {
+    armed: Arc<AtomicBool>,
+    backend_committed_signal: watch::Sender<bool>,
+    backend_committed: watch::Receiver<bool>,
+    release: watch::Sender<bool>,
+}
+
+impl ReplicationResponseHold {
+    fn arm(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
+        if self.armed.load(Ordering::Acquire) {
+            return Err("replication response hold was already armed".into());
+        }
+        self.backend_committed_signal
+            .send(false)
+            .map_err(|_| "replication response hold closed before rearming")?;
+        self.release
+            .send(false)
+            .map_err(|_| "replication response hold closed before rearming")?;
+        self.armed
+            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+            .map_err(|_| "replication response hold was already armed")?;
+        Ok(())
+    }
+
+    async fn wait_for_backend_commit(&mut self) -> Result<(), Box<dyn Error + Send + Sync>> {
+        let wait = async {
+            while !*self.backend_committed.borrow() {
+                self.backend_committed
+                    .changed()
+                    .await
+                    .map_err(|_| "replication response hold closed before the backend committed")?;
+            }
+            Ok::<(), Box<dyn Error + Send + Sync>>(())
+        };
+        timeout(Duration::from_secs(60), wait)
+            .await
+            .map_err(|_| "timed out waiting for the replication backend to commit")?
+    }
+
+    fn release(&self) -> Result<(), Box<dyn Error + Send + Sync>> {
+        self.release
+            .send(true)
+            .map_err(|_| "replication response hold closed before release")?;
+        Ok(())
+    }
+}
+
+fn replication_response_hold() -> (ReplicationResponseHoldRuntime, ReplicationResponseHold) {
+    let armed = Arc::new(AtomicBool::new(false));
+    let (backend_committed, backend_committed_rx) = watch::channel(false);
+    let (release, release_rx) = watch::channel(false);
+    (
+        ReplicationResponseHoldRuntime {
+            armed: armed.clone(),
+            backend_committed: backend_committed.clone(),
+            release: release_rx,
+        },
+        ReplicationResponseHold {
+            armed,
+            backend_committed_signal: backend_committed,
+            backend_committed: backend_committed_rx,
+            release,
+        },
+    )
+}
+
 async fn forward_replication_proxy_request(
     request: Request<Incoming>,
     backend_url: &str,
@@ -2015,6 +2096,7 @@ async fn forward_replication_proxy_request(
     request_count: &AtomicU64,
     mut replication_enabled: watch::Receiver<bool>,
     mut held_tagging: watch::Receiver<Option<String>>,
+    mut response_hold: ReplicationResponseHoldRuntime,
 ) -> Response<Full<bytes::Bytes>> {
     let (parts, body) = request.into_parts();
     let is_replication = parts
@@ -2040,6 +2122,7 @@ async fn forward_replication_proxy_request(
             }
         }
     }
+    let hold_response = is_replication && parts.method == http::Method::PUT && response_hold.try_claim();
 
     let Some(path_and_query) = parts.uri.path_and_query() else {
         return proxy_error_response("request URI omitted path");
@@ -2062,6 +2145,16 @@ async fn forward_replication_proxy_request(
         Ok(body) => body,
         Err(error) => return proxy_error_response(error),
     };
+    if hold_response && status.is_success() {
+        if response_hold.backend_committed.send(true).is_err() {
+            return proxy_error_response("replication response hold closed after the backend committed");
+        }
+        while !*response_hold.release.borrow() {
+            if response_hold.release.changed().await.is_err() {
+                return proxy_error_response("replication response hold closed before release");
+            }
+        }
+    }
     let mut proxied = Response::builder().status(status);
     for (name, value) in &headers {
         proxied = proxied.header(name, value);
@@ -2073,7 +2166,7 @@ async fn start_replication_counting_proxy(
     backend_url: &str,
     tasks: &mut JoinSet<()>,
 ) -> Result<(String, Arc<AtomicU64>, watch::Sender<bool>), Box<dyn Error + Send + Sync>> {
-    let (proxy_url, request_count, replication_enabled, _held_tagging) =
+    let (proxy_url, request_count, replication_enabled, _held_tagging, _response_hold) =
         start_replication_counting_proxy_with_tag_hold(backend_url, tasks).await?;
     Ok((proxy_url, request_count, replication_enabled))
 }
@@ -2085,7 +2178,16 @@ async fn start_replication_counting_proxy(
 async fn start_replication_counting_proxy_with_tag_hold(
     backend_url: &str,
     tasks: &mut JoinSet<()>,
-) -> Result<(String, Arc<AtomicU64>, watch::Sender<bool>, watch::Sender<Option<String>>), Box<dyn Error + Send + Sync>> {
+) -> Result<
+    (
+        String,
+        Arc<AtomicU64>,
+        watch::Sender<bool>,
+        watch::Sender<Option<String>>,
+        ReplicationResponseHold,
+    ),
+    Box<dyn Error + Send + Sync>,
+> {
     let listener = TcpListener::bind("127.0.0.1:0").await?;
     let proxy_url = format!("http://{}", listener.local_addr()?);
     let backend_url = backend_url.to_string();
@@ -2093,6 +2195,7 @@ async fn start_replication_counting_proxy_with_tag_hold(
     let task_request_count = request_count.clone();
     let (replication_enabled, task_replication_enabled) = watch::channel(true);
     let (held_tagging, task_held_tagging) = watch::channel(None);
+    let (response_hold_runtime, response_hold) = replication_response_hold();
     tasks.spawn(async move {
         let client = local_http_client();
         let mut connections = JoinSet::new();
@@ -2105,6 +2208,7 @@ async fn start_replication_counting_proxy_with_tag_hold(
                     let request_count = task_request_count.clone();
                     let replication_enabled = task_replication_enabled.clone();
                     let held_tagging = task_held_tagging.clone();
+                    let response_hold = response_hold_runtime.clone();
                     connections.spawn(async move {
                         let service = service_fn(move |request| {
                             let backend_url = backend_url.clone();
@@ -2112,6 +2216,7 @@ async fn start_replication_counting_proxy_with_tag_hold(
                             let request_count = request_count.clone();
                             let replication_enabled = replication_enabled.clone();
                             let held_tagging = held_tagging.clone();
+                            let response_hold = response_hold.clone();
                             async move {
                                 Ok::<_, Infallible>(
                                     forward_replication_proxy_request(
@@ -2121,6 +2226,7 @@ async fn start_replication_counting_proxy_with_tag_hold(
                                         &request_count,
                                         replication_enabled,
                                         held_tagging,
+                                        response_hold,
                                     )
                                     .await,
                                 )
@@ -2133,7 +2239,7 @@ async fn start_replication_counting_proxy_with_tag_hold(
             }
         }
     });
-    Ok((proxy_url, request_count, replication_enabled, held_tagging))
+    Ok((proxy_url, request_count, replication_enabled, held_tagging, response_hold))
 }
 
 async fn site_replication_remove(
@@ -3824,6 +3930,12 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR
     let mut source_env = RustFSTestEnvironment::new().await?;
     let mut source_env_vars = replication_fast_env();
     source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    // This matrix verifies request-time rule/admission behavior. Keep the
+    // background existing-object scanner outside the observation window: with
+    // ExistingObjectReplication enabled it may legitimately discover an
+    // object after its tags change, which is a separate data-replication path
+    // that PR #5696 intentionally did not alter.
+    source_env_vars.push(("RUSTFS_SCANNER_START_DELAY_SECS", "300"));
     source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
 
     let mut target_env_a = RustFSTestEnvironment::new().await?;
@@ -4052,6 +4164,13 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR
         .await?;
     assert_replication_key_absent(&target_client_b, target_bucket_b, "tagged/no-match.txt", Duration::from_secs(3)).await?;
 
+    // A metadata edit must not retroactively admit data that failed the tag
+    // filter at PUT time. This is the PR #5696 safety boundary: the target ARN
+    // has no persisted data-admission state for this version, so adding the
+    // matching tag later remains metadata-only and fails closed.
+    put_single_tag_current(&source_client, source_bucket, "tagged/no-match.txt", "route", "tagged").await?;
+    assert_replication_key_absent(&target_client_b, target_bucket_b, "tagged/no-match.txt", Duration::from_secs(3)).await?;
+
     source_client
         .put_object()
         .bucket(source_bucket)
@@ -7155,6 +7274,27 @@ async fn put_single_tag(
     Ok(())
 }
 
+async fn put_single_tag_current(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    tag_key: &str,
+    tag_value: &str,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    client
+        .put_object_tagging()
+        .bucket(bucket)
+        .key(key)
+        .tagging(
+            aws_sdk_s3::types::Tagging::builder()
+                .tag_set(aws_sdk_s3::types::Tag::builder().key(tag_key).value(tag_value).build()?)
+                .build()?,
+        )
+        .send()
+        .await?;
+    Ok(())
+}
+
 async fn get_single_tag(
     client: &Client,
     bucket: &str,
@@ -7202,6 +7342,28 @@ async fn wait_for_single_tag(
     }
 }
 
+/// Poll one site until `tag_key` is absent from the selected version.
+async fn wait_for_tag_absent(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    tag_key: &str,
+    site: &str,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+    loop {
+        let observed = get_single_tag(client, bucket, key, version_id, tag_key).await?;
+        if observed.is_none() {
+            return Ok(());
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return Err(format!("{site}: {bucket}/{key}?versionId={version_id} tag {tag_key} remained {observed:?}").into());
+        }
+        sleep(Duration::from_millis(200)).await;
+    }
+}
+
 /// Tag key the dual-node LWW scenario edits on both sites.
 const LWW_TAG_KEY: &str = "owner";
 
@@ -7249,6 +7411,265 @@ async fn wait_for_proxy_replication_requests(
     }
 }
 
+/// rustfs/backlog#2099: metadata admission must not drop a tag edit made after
+/// the target has committed the initial object but before the source persists
+/// that replication as COMPLETED.
+#[tokio::test]
+async fn test_site_replication_tagging_during_initial_pending_window_converges() -> TestResult {
+    init_logging();
+
+    // `RustFSTestEnvironment::start_rustfs_server_with_env` resolves (and on a
+    // cold checkout builds) this binary synchronously. Keep that setup outside
+    // the scenario timeout so 180 seconds measures the runtime race rather
+    // than compilation latency.
+    let _rustfs_binary = rustfs_binary_path();
+
+    match timeout(Duration::from_secs(180), async {
+        const PAYLOAD: &str = "tagging during pending replication";
+        const TAG_KEY: &str = "window";
+        const TAG_VALUE: &str = "pending";
+        const DELETE_PAYLOAD: &str = "tag deletion during pending replication";
+        const DELETE_TAG_KEY: &str = "remove";
+        const DELETE_TAG_VALUE: &str = "while-pending";
+
+        let mut site_env = replication_fast_env();
+        site_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+
+        let mut site_a_env = RustFSTestEnvironment::new().await?;
+        site_a_env.start_rustfs_server_with_env(vec![], &site_env).await?;
+
+        let mut site_b_env = RustFSTestEnvironment::new().await?;
+        site_b_env.start_rustfs_server_with_env(vec![], &site_env).await?;
+
+        let site_a_client = site_a_env.create_s3_client();
+        let site_b_client = site_b_env.create_s3_client();
+        let mut proxy_tasks = JoinSet::new();
+        let (
+            site_b_proxy,
+            _site_b_replication_requests,
+            _site_b_replication_enabled,
+            _site_b_held_tagging,
+            mut site_b_response_hold,
+        ) = start_replication_counting_proxy_with_tag_hold(&site_b_env.url, &mut proxy_tasks).await?;
+
+        let add_status = site_replication_add(
+            &site_a_env,
+            &[
+                PeerSite {
+                    name: "pending-site-a".to_string(),
+                    endpoint: site_a_env.url.clone(),
+                    access_key: site_a_env.access_key.clone(),
+                    secret_key: site_a_env.secret_key.clone(),
+                    ..Default::default()
+                },
+                PeerSite {
+                    name: "pending-site-b".to_string(),
+                    endpoint: site_b_env.url.clone(),
+                    access_key: site_b_env.access_key.clone(),
+                    secret_key: site_b_env.secret_key.clone(),
+                    ..Default::default()
+                },
+            ],
+        )
+        .await?;
+        assert!(add_status.success, "unexpected site add result: {add_status:?}");
+
+        let site_info = wait_for_site_replication_enabled(&site_a_env, 2).await?;
+        wait_for_site_replication_enabled(&site_b_env, 2).await?;
+        let mut site_b_peer = site_info
+            .sites
+            .iter()
+            .find(|peer| peer.endpoint == site_b_env.url.as_str())
+            .ok_or("site B peer missing from replication info")?
+            .clone();
+        site_b_peer.endpoint = site_b_proxy.clone();
+        site_b_peer.sync_state = SyncStatus::Enable;
+        let edit = site_replication_edit(&site_a_env, "", &site_b_peer).await?;
+        assert!(edit.success, "unexpected site B endpoint edit: {edit:?}");
+        for env in [&site_a_env, &site_b_env] {
+            wait_for_site_replication_info(env, |info| info.sites.iter().any(|peer| peer.endpoint == site_b_proxy)).await?;
+        }
+
+        let bucket = "site-repl-tag-pending";
+        let key = "pending-window.txt";
+        site_a_client.create_bucket().bucket(bucket).send().await?;
+        wait_for_bucket_on_target(&site_b_client, bucket).await?;
+
+        site_b_response_hold.arm()?;
+        let put_task = {
+            let client = site_a_client.clone();
+            let bucket = bucket.to_string();
+            let key = key.to_string();
+            tokio::spawn(async move {
+                client
+                    .put_object()
+                    .bucket(bucket)
+                    .key(key)
+                    .body(ByteStream::from_static(PAYLOAD.as_bytes()))
+                    .send()
+                    .await
+            })
+        };
+
+        // The proxy reads B's complete successful response before parking it,
+        // so B's GET proves the object is committed while A's worker is still
+        // unable to persist COMPLETED.
+        site_b_response_hold.wait_for_backend_commit().await?;
+        wait_for_replicated_object(&site_b_client, bucket, key, PAYLOAD).await?;
+        let source_head = site_a_client.head_object().bucket(bucket).key(key).send().await?;
+        let version_id = source_head
+            .version_id()
+            .ok_or("source HEAD omitted the pending version ID")?
+            .to_string();
+        assert_eq!(
+            source_head.replication_status().map(|status| status.as_str()),
+            Some("PENDING"),
+            "source must still report PENDING while the initial replication response is held"
+        );
+
+        let tag_task = {
+            let client = site_a_client.clone();
+            let bucket = bucket.to_string();
+            let key = key.to_string();
+            // The original real-machine failure used the current-version S3
+            // API (no versionId). The delete phase below deliberately keeps an
+            // explicit versionId so both request shapes stay covered.
+            tokio::spawn(async move { put_single_tag_current(&client, &bucket, &key, TAG_KEY, TAG_VALUE).await })
+        };
+        wait_for_single_tag(&site_a_client, bucket, key, &version_id, TAG_KEY, TAG_VALUE, "site A").await?;
+        assert_eq!(
+            head_replication_status(&site_a_client, bucket, key, &version_id)
+                .await?
+                .as_deref(),
+            Some("PENDING"),
+            "tag update must be authored before the initial replication reaches COMPLETED"
+        );
+
+        site_b_response_hold.release()?;
+        let put_output = timeout(Duration::from_secs(60), put_task)
+            .await
+            .map_err(|_| "source PutObject remained blocked after releasing the replication response")???;
+        timeout(Duration::from_secs(60), tag_task)
+            .await
+            .map_err(|_| "PutObjectTagging remained blocked after releasing the replication response")???;
+        assert_eq!(put_output.version_id(), Some(version_id.as_str()));
+
+        wait_for_single_tag(&site_b_client, bucket, key, &version_id, TAG_KEY, TAG_VALUE, "site B").await?;
+        wait_for_version_replication_status(&site_a_client, bucket, key, &version_id, &["COMPLETED"], "site A").await?;
+        wait_for_version_replication_status(&site_b_client, bucket, key, &version_id, &["REPLICA"], "site B").await?;
+
+        let source_state = list_replication_state(&site_a_client, bucket).await?;
+        let target_state = list_replication_state(&site_b_client, bucket).await?;
+        assert_eq!(source_state, target_state, "tag replication must not fork the object version");
+        assert_eq!(source_state.len(), 1, "tag replication must leave exactly one object version");
+        assert_eq!(source_state[0].key, key);
+        assert_eq!(source_state[0].version_id, version_id);
+
+        // Re-arm the same response barrier for an initially tagged object.
+        // Deleting its tag while A is still PENDING proves the same admission
+        // rule covers DeleteObjectTagging rather than only the original PUT
+        // symptom.
+        let delete_key = "pending-delete-window.txt";
+        site_b_response_hold.arm()?;
+        let delete_put_task = {
+            let client = site_a_client.clone();
+            let bucket = bucket.to_string();
+            let key = delete_key.to_string();
+            tokio::spawn(async move {
+                client
+                    .put_object()
+                    .bucket(bucket)
+                    .key(key)
+                    .tagging(format!("{DELETE_TAG_KEY}={DELETE_TAG_VALUE}"))
+                    .body(ByteStream::from_static(DELETE_PAYLOAD.as_bytes()))
+                    .send()
+                    .await
+            })
+        };
+
+        site_b_response_hold.wait_for_backend_commit().await?;
+        wait_for_replicated_object(&site_b_client, bucket, delete_key, DELETE_PAYLOAD).await?;
+        let delete_source_head = site_a_client.head_object().bucket(bucket).key(delete_key).send().await?;
+        let delete_version_id = delete_source_head
+            .version_id()
+            .ok_or("source HEAD omitted the pending tagged version ID")?
+            .to_string();
+        assert_eq!(
+            delete_source_head.replication_status().map(|status| status.as_str()),
+            Some("PENDING"),
+            "source tagged object must remain PENDING while its initial response is held"
+        );
+        wait_for_single_tag(
+            &site_b_client,
+            bucket,
+            delete_key,
+            &delete_version_id,
+            DELETE_TAG_KEY,
+            DELETE_TAG_VALUE,
+            "site B",
+        )
+        .await?;
+
+        let delete_tag_task = {
+            let client = site_a_client.clone();
+            let bucket = bucket.to_string();
+            let key = delete_key.to_string();
+            let version_id = delete_version_id.clone();
+            tokio::spawn(async move {
+                client
+                    .delete_object_tagging()
+                    .bucket(bucket)
+                    .key(key)
+                    .version_id(version_id)
+                    .send()
+                    .await
+            })
+        };
+        wait_for_tag_absent(&site_a_client, bucket, delete_key, &delete_version_id, DELETE_TAG_KEY, "site A").await?;
+        assert_eq!(
+            head_replication_status(&site_a_client, bucket, delete_key, &delete_version_id)
+                .await?
+                .as_deref(),
+            Some("PENDING"),
+            "tag deletion must be authored before the initial replication reaches COMPLETED"
+        );
+
+        site_b_response_hold.release()?;
+        let delete_put_output = timeout(Duration::from_secs(60), delete_put_task)
+            .await
+            .map_err(|_| "source tagged PutObject remained blocked after releasing the replication response")???;
+        timeout(Duration::from_secs(60), delete_tag_task)
+            .await
+            .map_err(|_| "DeleteObjectTagging remained blocked after releasing the replication response")???;
+        assert_eq!(delete_put_output.version_id(), Some(delete_version_id.as_str()));
+
+        wait_for_tag_absent(&site_b_client, bucket, delete_key, &delete_version_id, DELETE_TAG_KEY, "site B").await?;
+        wait_for_version_replication_status(&site_a_client, bucket, delete_key, &delete_version_id, &["COMPLETED"], "site A")
+            .await?;
+        wait_for_version_replication_status(&site_b_client, bucket, delete_key, &delete_version_id, &["REPLICA"], "site B")
+            .await?;
+
+        let source_state = list_replication_state(&site_a_client, bucket).await?;
+        let target_state = list_replication_state(&site_b_client, bucket).await?;
+        assert_eq!(source_state, target_state, "tag deletion must not fork the object version");
+        assert_eq!(source_state.len(), 2, "pending-window scenarios must leave exactly two object versions");
+        assert!(
+            source_state
+                .iter()
+                .any(|entry| entry.key == delete_key && entry.version_id == delete_version_id),
+            "tag deletion must preserve the original version identity"
+        );
+
+        proxy_tasks.abort_all();
+        Ok(())
+    })
+    .await
+    {
+        Ok(result) => result,
+        Err(_) => Err("pending-window site-replication tagging test timed out".into()),
+    }
+}
+
 /// rustfs/backlog#1953 (audit A4/P1-6): receiver-side LWW for replicated
 /// metadata categories, exercised end to end over the real dual-node
 /// active-active site-replication control plane — sender, worker, status
@@ -7285,9 +7706,9 @@ async fn test_site_replication_tagging_lww_converges_active_active_real_dual_nod
         site_b_env.start_rustfs_server_with_env(vec![], &site_env).await?;
 
         let mut proxy_tasks = JoinSet::new();
-        let (site_a_proxy, site_a_replication_requests, _site_a_replication_enabled, site_a_held_tagging) =
+        let (site_a_proxy, site_a_replication_requests, _site_a_replication_enabled, site_a_held_tagging, _site_a_response_hold) =
             start_replication_counting_proxy_with_tag_hold(&site_a_env.url, &mut proxy_tasks).await?;
-        let (site_b_proxy, site_b_replication_requests, _site_b_replication_enabled, site_b_held_tagging) =
+        let (site_b_proxy, site_b_replication_requests, _site_b_replication_enabled, site_b_held_tagging, _site_b_response_hold) =
             start_replication_counting_proxy_with_tag_hold(&site_b_env.url, &mut proxy_tasks).await?;
 
         let site_a_client = site_a_env.create_s3_client();

crates/ecstore/src/bucket/replication/mod.rs

@@ -53,12 +53,12 @@ pub use replication_config_boundary::{
     replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id,
     unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
 };
-pub(crate) use replication_filemeta_boundary::version_purge_statuses_map;
 pub use replication_filemeta_boundary::{
     MrfOpKind, MrfReplicateEntry, REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState,
     ReplicationStatusType, ReplicationType, VersionPurgeStatusType, replication_state_to_filemeta,
     replication_status_to_filemeta, replication_statuses_map, version_purge_status_to_filemeta,
 };
+pub(crate) use replication_filemeta_boundary::{ReplicationGenerationSnapshot, version_purge_statuses_map};
 pub(crate) use replication_filemeta_boundary::{
     replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta,
 };

crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs

@@ -15,7 +15,7 @@
 pub use rustfs_replication::{MrfOpKind, MrfReplicateEntry};
 pub(crate) use rustfs_replication::{
     REPLICATE_EXISTING, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction,
-    ReplicationWorkerOperation, ResyncDecision, get_replication_state, parse_replicate_decision,
+    ReplicationGenerationSnapshot, ReplicationWorkerOperation, ResyncDecision, get_replication_state, parse_replicate_decision,
     replicate_decision_for_admitted_targets, target_reset_header, version_purge_statuses_map,
 };
 pub use rustfs_replication::{

crates/ecstore/src/bucket/replication/replication_object_config.rs

@@ -575,7 +575,7 @@ pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplic
         let mut sopts = opts.clone();
         sopts.target_arn = arn.clone();
 
-        let replicate = cfg.replicate(&sopts) && mopts.metadata_target_is_eligible(&arn);
+        let replicate = metadata_target_should_replicate(&cfg, &sopts, &mopts, &arn);
         let synchronous = if let Some(cli) = cli { cli.replicate_sync } else { false };
 
         dsc.set(ReplicateTargetDecision::new(arn, replicate, synchronous));
@@ -584,6 +584,15 @@ pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplic
     dsc
 }
 
+fn metadata_target_should_replicate(
+    cfg: &ReplicationConfiguration,
+    opts: &ObjectOpts,
+    mopts: &MustReplicateOptions,
+    arn: &str,
+) -> bool {
+    cfg.replicate(opts) && mopts.metadata_target_is_eligible(arn)
+}
+
 #[cfg(test)]
 mod tests {
     use s3s::dto::{
@@ -613,6 +622,46 @@ mod tests {
         }
     }
 
+    #[test]
+    fn metadata_replication_requires_both_current_rule_match_and_historical_admission() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let mut rule = replication_rule();
+        rule.destination.bucket = arn.to_string();
+        rule.filter = Some(ReplicationRuleFilter {
+            prefix: Some("admitted/".to_string()),
+            ..Default::default()
+        });
+        let cfg = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_REPLICATION_STATUS, format!("{arn}=PENDING;"));
+        let admitted = MustReplicateOptions::new(&metadata, String::new(), ReplicationType::Metadata, false);
+        let matching = ObjectOpts {
+            name: "admitted/object".to_string(),
+            target_arn: arn.to_string(),
+            ..Default::default()
+        };
+        assert!(metadata_target_should_replicate(&cfg, &matching, &admitted, arn));
+
+        let rule_mismatch = ObjectOpts {
+            name: "outside/object".to_string(),
+            target_arn: arn.to_string(),
+            ..Default::default()
+        };
+        assert!(
+            !metadata_target_should_replicate(&cfg, &rule_mismatch, &admitted, arn),
+            "historical admission must not bypass the current replication rule"
+        );
+
+        let never_admitted = MustReplicateOptions::new(&HashMap::new(), String::new(), ReplicationType::Metadata, false);
+        assert!(
+            !metadata_target_should_replicate(&cfg, &matching, &never_admitted, arn),
+            "a current rule match must not create historical admission"
+        );
+    }
+
     #[test]
     fn replication_config_empty_and_replicate_follow_config() {
         let empty = ReplicationConfig::default();

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -52,7 +52,6 @@ use super::runtime_boundary as runtime_sources;
 use futures_util::stream::{self, StreamExt};
 use metrics::{counter, histogram};
 use rustfs_utils::hash::HashAlgorithm;
-use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str};
 use std::collections::HashMap;
 use std::collections::hash_map::Entry;
 use std::sync::Arc;
@@ -939,7 +938,11 @@ async fn replay_mrf_object_entry<S: ReplicationStorage>(
         Some(queue_replication_heal(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
     } else {
         let roi = admitted_mrf_replicate_object(oi, entry, entry.op.replication_type());
-        if replicate_object_with_outcome(roi, storage.clone()).await.1 {
+        if replicate_object_with_outcome(roi, storage.clone())
+            .await
+            .1
+            .consumes_mrf_entry()
+        {
             Some(ReplicationQueueAdmission::Queued)
         } else {
             Some(ReplicationQueueAdmission::Missed)
@@ -978,7 +981,11 @@ async fn replay_mrf_metadata_entry<S: ReplicationStorage>(
         Some(queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
     } else {
         let roi = admitted_mrf_replicate_object(oi, entry, ReplicationType::Metadata);
-        if replicate_object_with_outcome(roi, storage.clone()).await.1 {
+        if replicate_object_with_outcome(roi, storage.clone())
+            .await
+            .1
+            .consumes_mrf_entry()
+        {
             Some(ReplicationQueueAdmission::Queued)
         } else {
             Some(ReplicationQueueAdmission::Missed)
@@ -2978,8 +2985,11 @@ fn replicate_object_info_from_object_info(
 ) -> 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)
-        .map(|v| OffsetDateTime::parse(&v, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH));
+    let replication_generation = oi.replication_generation_snapshot();
+    let tm = replication_generation
+        .timestamp
+        .as_deref()
+        .map(|value| OffsetDateTime::parse(value, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH));
     let mut rstate = oi.replication_state();
     rstate.replicate_decision_str = dsc.to_string();
     let asz = oi.get_actual_size_or_physical();
@@ -3006,6 +3016,7 @@ fn replicate_object_info_from_object_info(
         target_statuses: tgt_statuses,
         target_purge_statuses: purge_statuses,
         replication_timestamp: tm,
+        replication_generation,
         user_tags: (*oi.user_tags).clone(),
         checksum,
         retry_count: 0,

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -17,6 +17,8 @@ use super::replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt
 use super::replication_config_store::ReplicationConfigStore;
 use super::replication_error_boundary::{Error, Result, is_err_object_not_found, is_err_version_not_found};
 use super::replication_event_sink::{EventArgs, send_event, send_local_event};
+#[cfg(test)]
+use super::replication_filemeta_boundary::ReplicationGenerationSnapshot;
 use super::replication_filemeta_boundary::{
     REPLICATE_EXISTING, ReplicateDecision, ReplicateObjectInfo, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction,
     ReplicationState, ReplicationStatusType, ReplicationType, VersionPurgeStatusType, get_replication_state,
@@ -47,12 +49,13 @@ use super::replication_resync_boundary::{
 };
 #[cfg(test)]
 use super::replication_resync_boundary::{RESYNC_META_FORMAT, RESYNC_META_VERSION, WIRE_ZERO_TIME_UNIX};
-#[cfg(test)]
-use super::replication_storage_boundary::ReplicationDeletedObject;
 use super::replication_storage_boundary::{
     AdvancedGetOptions, EcstoreObjectOperations, GetObjectReader, HTTPPreconditions, HTTPRangeSpec, ObjectInfo, ObjectOptions,
-    ObjectToDelete, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions,
+    ObjectToDelete, ReplicationObjectIO, ReplicationStatusWritebackCondition, ReplicationStatusWritebackMode, ReplicationStorage,
+    StatObjectOptions, StorageObjectInfoOrErr, WalkOptions,
 };
+#[cfg(test)]
+use super::replication_storage_boundary::{NamespaceLockFence, NamespaceLockSignalTestFence, ReplicationDeletedObject};
 use super::replication_target_boundary::{
     ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions,
     RemotePutObjectResponse, ReplicationTargetStore, S3ClientError, SsecPassthroughCapability, SsecPassthroughGate, TargetClient,
@@ -1625,6 +1628,7 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC
     let mut replication_state = oi.replication_state();
     replication_state.replicate_decision_str = dsc.to_string();
     let actual_size = oi.get_actual_size_or_physical();
+    let replication_generation = oi.replication_generation_snapshot();
 
     Ok(ReplicateObjectInfo {
         name: oi.name.clone(),
@@ -1647,6 +1651,7 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC
         target_statuses,
         target_purge_statuses,
         replication_timestamp: None,
+        replication_generation,
         ssec: replication_object_is_ssec_encrypted(&user_defined),
         user_tags: (*oi.user_tags).clone(),
         checksum: oi.checksum.clone(),
@@ -2931,10 +2936,125 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
     replicate_object_with_outcome(roi, storage).await.0
 }
 
+enum ReplicationStatePersistOutcome {
+    Updated,
+    Superseded,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum ReplicationAttemptDisposition {
+    Persisted,
+    Superseded,
+    Retry,
+}
+
+impl ReplicationAttemptDisposition {
+    pub(crate) fn consumes_mrf_entry(self) -> bool {
+        matches!(self, Self::Persisted | Self::Superseded)
+    }
+}
+
+fn replication_attempt_preflight(roi: &ReplicateObjectInfo) -> Option<(ReplicationState, ReplicationAttemptDisposition)> {
+    roi.replication_generation
+        .invalid
+        .then(|| (roi.replication_state.clone().unwrap_or_default(), ReplicationAttemptDisposition::Retry))
+}
+
+#[derive(Debug, Default, PartialEq, Eq)]
+struct ReplicationTerminalPublication {
+    emit_terminal_failure: bool,
+    emit_event: bool,
+    update_transition_stats: bool,
+    update_same_state_failure_stats: bool,
+}
+
+fn replication_terminal_publication(
+    suppress_terminal_publication: bool,
+    state_update_needed: bool,
+    attempt_status: ReplicationStatusType,
+    previous_internal_matches_attempt: bool,
+) -> ReplicationTerminalPublication {
+    if suppress_terminal_publication {
+        return ReplicationTerminalPublication::default();
+    }
+    ReplicationTerminalPublication {
+        emit_terminal_failure: true,
+        emit_event: true,
+        update_transition_stats: state_update_needed,
+        update_same_state_failure_stats: attempt_status != ReplicationStatusType::Completed && previous_internal_matches_attempt,
+    }
+}
+
+fn replication_status_writeback_mode(state_update_needed: bool) -> ReplicationStatusWritebackMode {
+    if state_update_needed {
+        ReplicationStatusWritebackMode::Update
+    } else {
+        ReplicationStatusWritebackMode::ValidateOnly
+    }
+}
+
+/// Publish a worker's status with a storage-enforced compare-and-set token.
+/// `put_object_metadata` checks the token while it owns the object write lock,
+/// avoiding both a read/write race and any need to hold a hot object lock
+/// across network I/O.
+fn replication_status_writeback_options(
+    roi: &ReplicateObjectInfo,
+    replication_lock_guard: &rustfs_lock::NamespaceLockGuard,
+    new_replication_internal: Option<&String>,
+    mode: ReplicationStatusWritebackMode,
+) -> ObjectOptions {
+    let mut eval_metadata = HashMap::new();
+    if let Some(status) = new_replication_internal {
+        insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, status.clone());
+    }
+    let mut write_opts = ObjectOptions {
+        version_id: roi.version_id.map(|version_id| version_id.to_string()),
+        eval_metadata: Some(eval_metadata),
+        replication_status_writeback: Some(Box::new(ReplicationStatusWritebackCondition {
+            expected_generation: roi.replication_generation.clone(),
+            mode,
+        })),
+        ..Default::default()
+    };
+    // The remote transfer runs under a renewable replication namespace lock.
+    // Carry that guard's loss signal into the storage commit so a worker whose
+    // lease expired while it was doing remote I/O cannot publish over a newer
+    // worker for the same generation.
+    write_opts.add_namespace_lock_guard(replication_lock_guard);
+    write_opts
+}
+
+async fn persist_replication_state_if_current<S: ReplicationStorage>(
+    roi: &ReplicateObjectInfo,
+    storage: &Arc<S>,
+    replication_lock_guard: &rustfs_lock::NamespaceLockGuard,
+    new_replication_internal: Option<&String>,
+    mode: ReplicationStatusWritebackMode,
+    object_info: &mut ObjectInfo,
+) -> Result<ReplicationStatePersistOutcome> {
+    let write_opts = replication_status_writeback_options(roi, replication_lock_guard, new_replication_internal, mode);
+    match storage.put_object_metadata(&roi.bucket, &roi.name, &write_opts).await {
+        Ok(updated) => {
+            *object_info = updated;
+            Ok(ReplicationStatePersistOutcome::Updated)
+        }
+        Err(Error::PreconditionFailed) => Ok(ReplicationStatePersistOutcome::Superseded),
+        Err(error) => Err(error),
+    }
+}
+
 pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
     roi: ReplicateObjectInfo,
     storage: Arc<S>,
-) -> (ReplicationState, bool) {
+) -> (ReplicationState, ReplicationAttemptDisposition) {
+    // Conflicting compatibility aliases, empty opaque timestamps, and invalid
+    // mutation UUIDs are corruption, not evidence that another generation
+    // superseded this task. Fail before any target I/O and keep the MRF entry
+    // retryable instead of repeatedly transmitting and then acknowledging it.
+    if let Some(outcome) = replication_attempt_preflight(&roi) {
+        return outcome;
+    }
+
     let bucket = roi.bucket.clone();
     let object = roi.name.clone();
 
@@ -2963,10 +3083,10 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return (roi.replication_state.unwrap_or_default(), false);
+            return (roi.replication_state.unwrap_or_default(), ReplicationAttemptDisposition::Retry);
         }
     };
-    let _obj_lock_guard = match obj_ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await {
+    let obj_lock_guard = match obj_ns_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await {
         Ok(g) => g,
         Err(e) => {
             debug!(
@@ -2986,7 +3106,7 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
                 user_agent: "Internal: [Replication]".to_string(),
                 ..Default::default()
             });
-            return (roi.replication_state.unwrap_or_default(), false);
+            return (roi.replication_state.unwrap_or_default(), ReplicationAttemptDisposition::Retry);
         }
     };
 
@@ -3057,72 +3177,98 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
     }
 
     let version_id = roi.version_id.map(|v| v.to_string());
-    note_replication_terminal_failure(&bucket, &object, version_id.as_deref(), &rinfos);
 
     let previous_state = roi.replication_state.clone().unwrap_or_default();
-    let merged_state = get_replication_state(&rinfos, &previous_state, version_id);
-    let replication_status = merged_state.composite_replication_status();
+    let mut merged_state = get_replication_state(&rinfos, &previous_state, version_id.clone());
+    let mut replication_status = merged_state.composite_replication_status();
     let new_replication_internal = merged_state.replication_status_internal.clone();
     let mut object_info = roi.to_object_info();
-    let mut state_persisted = true;
+    let mut disposition = ReplicationAttemptDisposition::Persisted;
+    let mut suppress_terminal_publication = false;
+    let state_update_needed = roi.replication_status_internal != new_replication_internal || rinfos.replication_resynced();
+    let writeback_mode = replication_status_writeback_mode(state_update_needed);
 
-    if roi.replication_status_internal != new_replication_internal || rinfos.replication_resynced() {
-        let mut eval_metadata = HashMap::new();
-        if let Some(ref s) = new_replication_internal {
-            insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, s.clone());
-        }
-        let popts = ObjectOptions {
-            version_id: roi.version_id.map(|v| v.to_string()),
-            eval_metadata: Some(eval_metadata),
-            ..Default::default()
-        };
-
-        match storage.put_object_metadata(&bucket, &object, &popts).await {
-            Ok(u) => object_info = u,
-            Err(e) => {
-                state_persisted = false;
-                // Persisting the resynced replication status failed. Don't swallow
-                // it silently — the object's on-disk status now disagrees with the
-                // resync result and needs operator visibility (backlog#799 B23).
-                warn!(
-                    event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
-                    bucket = %bucket,
-                    object = %object,
-                    error = %e,
-                    "Failed to persist resynced replication status metadata"
-                );
-            }
-        }
-
-        if let Some(stats) = runtime_sources::replication_stats() {
-            for tgt in &rinfos.targets {
-                if tgt.replication_status != tgt.prev_replication_status {
-                    stats
-                        .update(&bucket, tgt, tgt.replication_status.clone(), tgt.prev_replication_status.clone())
-                        .await;
+    match persist_replication_state_if_current(
+        &roi,
+        &storage,
+        &obj_lock_guard,
+        new_replication_internal.as_ref(),
+        writeback_mode,
+        &mut object_info,
+    )
+    .await
+    {
+        Ok(ReplicationStatePersistOutcome::Updated) => {}
+        Ok(ReplicationStatePersistOutcome::Superseded) => {
+            // A tag/retention/legal-hold mutation committed while this
+            // worker was in flight. Its PENDING state is authoritative and
+            // must remain discoverable by the queue/MRF/scanner after a
+            // crash or missed admission. Return that newer state to sync
+            // callers and leave its worker to publish the terminal status.
+            suppress_terminal_publication = true;
+            disposition = ReplicationAttemptDisposition::Superseded;
+            let read_opts = ObjectOptions {
+                version_id: roi.version_id.map(|version_id| version_id.to_string()),
+                ..Default::default()
+            };
+            match storage.get_object_info(&bucket, &object, &read_opts).await {
+                Ok(current) => {
+                    object_info = current;
+                    merged_state = object_info.replication_state();
+                    replication_status = merged_state.composite_replication_status();
+                }
+                Err(error) => {
+                    // The CAS result is authoritative: a best-effort refetch
+                    // failure must not turn a superseded task back into a
+                    // retry that can publish the stale generation later.
+                    debug!(
+                        event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                        bucket = %bucket,
+                        object = %object,
+                        error = %error,
+                        reason = "source_snapshot_refetch_failed_after_superseded",
+                        "Could not refresh source state after skipping stale replication status update"
+                    );
                 }
             }
+            debug!(
+                event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                bucket = %bucket,
+                object = %object,
+                reason = "source_replication_snapshot_superseded",
+                "Skipped stale replication status update"
+            );
+        }
+        Err(e) => {
+            disposition = ReplicationAttemptDisposition::Retry;
+            suppress_terminal_publication = true;
+            // Persisting the resynced replication status failed. Don't swallow
+            // it silently — the object's on-disk status now disagrees with the
+            // resync result and needs operator visibility (backlog#799 B23).
+            warn!(
+                event = EVENT_RESYNC_TARGET_OPERATION_FAILED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
+                bucket = %bucket,
+                object = %object,
+                error = %e,
+                "Failed to persist resynced replication status metadata"
+            );
         }
     }
 
-    let event_name = if replication_status == ReplicationStatusType::Completed {
-        EventName::ObjectReplicationComplete.to_string()
-    } else {
-        EventName::ObjectReplicationFailed.to_string()
-    };
+    let publication = replication_terminal_publication(
+        suppress_terminal_publication,
+        state_update_needed,
+        rinfos.replication_status(),
+        roi.replication_status_internal == rinfos.replication_status_internal(),
+    );
 
-    send_local_event(EventArgs {
-        event_name,
-        bucket_name: bucket.clone(),
-        object: object_info,
-        user_agent: "Internal: [Replication]".to_string(),
-        ..Default::default()
-    });
-
-    if rinfos.replication_status() != ReplicationStatusType::Completed
-        && roi.replication_status_internal == rinfos.replication_status_internal()
+    if publication.update_transition_stats
         && let Some(stats) = runtime_sources::replication_stats()
     {
         for tgt in &rinfos.targets {
@@ -3134,7 +3280,39 @@ pub(crate) async fn replicate_object_with_outcome<S: ReplicationStorage>(
         }
     }
 
-    (merged_state, state_persisted)
+    if publication.emit_terminal_failure {
+        note_replication_terminal_failure(&bucket, &object, version_id.as_deref(), &rinfos);
+    }
+
+    let event_name = if replication_status == ReplicationStatusType::Completed {
+        EventName::ObjectReplicationComplete.to_string()
+    } else {
+        EventName::ObjectReplicationFailed.to_string()
+    };
+
+    if publication.emit_event {
+        send_local_event(EventArgs {
+            event_name,
+            bucket_name: bucket.clone(),
+            object: object_info,
+            user_agent: "Internal: [Replication]".to_string(),
+            ..Default::default()
+        });
+    }
+
+    if publication.update_same_state_failure_stats
+        && let Some(stats) = runtime_sources::replication_stats()
+    {
+        for tgt in &rinfos.targets {
+            if tgt.replication_status != tgt.prev_replication_status {
+                stats
+                    .update(&bucket, tgt, tgt.replication_status.clone(), tgt.prev_replication_status.clone())
+                    .await;
+            }
+        }
+    }
+
+    (merged_state, disposition)
 }
 
 /// Emit the operator-visible record of a replication attempt that ended FAILED.
@@ -4567,6 +4745,105 @@ async fn replicate_multipart_parts_and_complete<S: ReplicationObjectIO>(
 mod tests {
     use super::super::replication_filemeta_boundary::ReplicateTargetDecision;
 
+    #[test]
+    fn same_state_terminal_retry_uses_validate_only() {
+        assert_eq!(replication_status_writeback_mode(false), ReplicationStatusWritebackMode::ValidateOnly);
+        assert_eq!(replication_status_writeback_mode(true), ReplicationStatusWritebackMode::Update);
+    }
+
+    #[test]
+    fn superseded_attempt_has_no_terminal_publication_side_effects() {
+        assert!(ReplicationAttemptDisposition::Persisted.consumes_mrf_entry());
+        assert!(ReplicationAttemptDisposition::Superseded.consumes_mrf_entry());
+        assert!(!ReplicationAttemptDisposition::Retry.consumes_mrf_entry());
+        assert_eq!(
+            replication_terminal_publication(true, false, ReplicationStatusType::Failed, true),
+            ReplicationTerminalPublication::default()
+        );
+        assert_eq!(
+            replication_terminal_publication(true, true, ReplicationStatusType::Completed, false),
+            ReplicationTerminalPublication::default()
+        );
+
+        let current = replication_terminal_publication(false, false, ReplicationStatusType::Failed, true);
+        assert!(current.emit_terminal_failure);
+        assert!(current.emit_event);
+        assert!(!current.update_transition_stats);
+        assert!(current.update_same_state_failure_stats);
+
+        assert_eq!(
+            replication_terminal_publication(true, true, ReplicationStatusType::Failed, false),
+            ReplicationTerminalPublication::default(),
+            "a retryable status-persistence failure must not publish a terminal result"
+        );
+    }
+
+    #[test]
+    fn invalid_generation_retries_before_remote_replication() {
+        let mut preserved_state = ReplicationState::default();
+        preserved_state
+            .targets
+            .insert("arn:target".to_string(), ReplicationStatusType::Pending);
+        let invalid = ReplicateObjectInfo {
+            replication_generation: ReplicationGenerationSnapshot {
+                invalid: true,
+                ..Default::default()
+            },
+            replication_state: Some(preserved_state.clone()),
+            ..Default::default()
+        };
+
+        assert_eq!(
+            replication_attempt_preflight(&invalid),
+            Some((preserved_state, ReplicationAttemptDisposition::Retry))
+        );
+        assert!(replication_attempt_preflight(&ReplicateObjectInfo::default()).is_none());
+    }
+
+    #[tokio::test]
+    async fn terminal_writeback_carries_replication_lock_loss_fence() {
+        let lock = rustfs_lock::NamespaceLock::new(
+            "replication-status-writeback-fence".to_string(),
+            Arc::new(rustfs_lock::LocalClient::new()),
+        );
+        let guard = lock
+            .get_write_lock(
+                rustfs_lock::ObjectKey::new("bucket", "/[replicate]/object"),
+                "worker-a",
+                std::time::Duration::from_secs(2),
+            )
+            .await
+            .expect("replication lock should be acquired");
+        let signal = guard
+            .lock_lost_signal()
+            .expect("distributed guard must expose its loss signal");
+        let forced_lost = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let _test_fence = NamespaceLockSignalTestFence::install_with_loss_handle(&signal, Arc::clone(&forced_lost));
+
+        // Build the writeback after remote work has acquired the guard, then
+        // lose the lease before storage reaches its commit fence.
+        let opts = replication_status_writeback_options(
+            &ReplicateObjectInfo::default(),
+            &guard,
+            None,
+            ReplicationStatusWritebackMode::ValidateOnly,
+        );
+        forced_lost.store(true, std::sync::atomic::Ordering::Release);
+
+        assert!(
+            opts.namespace_lock_fence
+                .as_ref()
+                .is_some_and(NamespaceLockFence::is_lock_lost),
+            "terminal CAS must observe a replication lease lost after remote I/O"
+        );
+        assert!(!ReplicationAttemptDisposition::Retry.consumes_mrf_entry());
+        assert_eq!(
+            replication_terminal_publication(true, true, ReplicationStatusType::Failed, false),
+            ReplicationTerminalPublication::default(),
+            "a fenced writeback retry must not publish terminal events or statistics"
+        );
+    }
+
     #[test]
     fn unavailable_object_target_is_persisted_as_failed() {
         let arn = "arn:object-target";

crates/ecstore/src/bucket/replication/replication_storage_boundary.rs

@@ -18,7 +18,11 @@ use tokio_util::sync::CancellationToken;
 use super::replication_error_boundary::Error;
 use super::replication_filemeta_boundary::{replication_state_from_filemeta, version_purge_status_from_filemeta};
 pub(crate) type ReplicationObjectStore = crate::store::ECStore;
-pub(crate) use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
+pub(crate) use crate::object_api::{
+    GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, ReplicationStatusWritebackCondition, ReplicationStatusWritebackMode,
+};
+#[cfg(test)]
+pub(crate) use crate::object_api::{NamespaceLockFence, NamespaceLockSignalTestFence};
 pub(crate) use crate::storage_api_contracts::list::{
     ListOperations, StorageListObjectVersionsInfo, StorageListObjectsV2Info, StorageObjectInfoOrErr, StorageWalkOptions,
 };

crates/ecstore/src/bucket/replication/replication_target_boundary.rs

@@ -849,6 +849,11 @@ mod tests {
         metadata.insert(MINIO_INTERNAL_ENCRYPTION_ALGORITHM_HEADER.to_string(), "DAREv2-HMAC-SHA256".to_string());
         metadata.insert(MINIO_INTERNAL_ENCRYPTION_SSEC_SEALED_KEY_HEADER.to_string(), "sealed".to_string());
         metadata.insert(MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER.to_string(), "true".to_string());
+        rustfs_utils::http::insert_str(
+            &mut metadata,
+            rustfs_utils::http::SUFFIX_REPLICATION_GENERATION,
+            Uuid::from_u128(1).to_string(),
+        );
 
         let object_info = ObjectInfo {
             user_defined: Arc::new(metadata),
@@ -894,6 +899,13 @@ mod tests {
         assert!(!options.user_metadata.contains_key(AMZ_SERVER_SIDE_ENCRYPTION));
         assert!(!options.user_metadata.contains_key(SSEC_ALGORITHM_HEADER));
         assert!(!options.user_metadata.contains_key(INTERNAL_ENCRYPTION_IV_HEADER));
+        assert!(
+            options
+                .user_metadata
+                .keys()
+                .all(|key| !key.contains(rustfs_utils::http::SUFFIX_REPLICATION_GENERATION)),
+            "source-local mutation generation must never cross the replication wire"
+        );
         assert!(
             !options
                 .user_metadata

crates/ecstore/src/object_api/mod.rs

@@ -18,8 +18,9 @@ pub mod object_api_utils;
 
 use crate::bucket::metadata_sys::get_versioning_config;
 use crate::bucket::replication::{
-    DeleteReplicationConfigSnapshot, ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType,
-    replication_status_from_filemeta, replication_statuses_map, version_purge_status_from_filemeta, version_purge_statuses_map,
+    DeleteReplicationConfigSnapshot, ReplicateDecision, ReplicationGenerationSnapshot, ReplicationState, ReplicationStatusType,
+    VersionPurgeStatusType, replication_status_from_filemeta, replication_statuses_map, version_purge_status_from_filemeta,
+    version_purge_statuses_map,
 };
 use crate::bucket::versioning::VersioningApi as _;
 use crate::config::storageclass;

crates/ecstore/src/object_api/types.rs

@@ -19,6 +19,7 @@ use crate::storage_api_contracts::{
         HTTPPreconditions, ObjectLockRetentionOptions, ObjectPreconditionError, ObjectPreconditionPart, ObjectPreconditionState,
     },
 };
+use sha2::{Digest, Sha256};
 use std::io;
 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use tokio::sync::{Mutex, Notify, OwnedRwLockReadGuard};
@@ -980,6 +981,13 @@ pub struct ObjectOptions {
     pub lifecycle_audit_event: LcAuditEvent,
 
     pub eval_metadata: Option<HashMap<String, String>>,
+    /// Internal compare-and-set condition for replication workers publishing
+    /// terminal status after remote I/O. Storage validates it while holding
+    /// the object write lock so an older worker cannot overwrite a newer
+    /// mutation's PENDING state. Keep the condition boxed because
+    /// `ObjectOptions` is passed by value through deep storage futures.
+    #[doc(hidden)]
+    pub replication_status_writeback: Option<Box<ReplicationStatusWritebackCondition>>,
     pub object_lock_retention: Option<ObjectLockRetentionOptions>,
     pub object_lock_delete: Option<crate::storage_api_contracts::object::ObjectLockDeleteOptions>,
     /// Authoritative bucket Object Lock snapshot installed inside `ECStore`
@@ -1000,6 +1008,21 @@ pub struct ObjectOptions {
     pub decommission_capacity_admission: Option<Arc<crate::store::ECStore>>,
 }
 
+#[derive(Clone, Debug, Default)]
+#[doc(hidden)]
+pub struct ReplicationStatusWritebackCondition {
+    pub(crate) expected_generation: ReplicationGenerationSnapshot,
+    pub(crate) mode: ReplicationStatusWritebackMode,
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+#[doc(hidden)]
+pub enum ReplicationStatusWritebackMode {
+    #[default]
+    Update,
+    ValidateOnly,
+}
+
 impl ObjectOptions {
     pub(crate) fn with_capacity_expected_data_bytes(expected_data_bytes: Option<usize>) -> Self {
         Self {
@@ -1086,6 +1109,7 @@ impl std::fmt::Debug for ObjectOptions {
                     || !self.lifecycle_audit_event.event.storage_class.is_empty()),
             )
             .field("eval_metadata_count", &self.eval_metadata.as_ref().map(HashMap::len))
+            .field("replication_status_writeback", &self.replication_status_writeback.is_some())
             .field("object_lock_retention", &self.object_lock_retention.is_some())
             .field("object_lock_delete", &self.object_lock_delete)
             .field("object_lock_config_snapshot", &self.object_lock_config_snapshot.is_some())
@@ -1281,6 +1305,32 @@ impl ObjectOptions {
     }
 }
 
+fn replication_snapshot_internal_value(
+    metadata: &HashMap<String, String>,
+    suffix: &str,
+) -> std::result::Result<Option<String>, ()> {
+    match rustfs_utils::http::get_consistent_str(metadata, suffix) {
+        Some(value) => Ok(Some(value.to_string())),
+        None if rustfs_utils::http::contains_key_str(metadata, suffix) => Err(()),
+        None => Ok(None),
+    }
+}
+
+fn update_replication_fingerprint_bytes(hasher: &mut Sha256, value: &[u8]) {
+    let len = u64::try_from(value.len()).unwrap_or(u64::MAX);
+    hasher.update(len.to_le_bytes());
+    hasher.update(value);
+}
+
+fn update_replication_fingerprint_optional_str(hasher: &mut Sha256, value: Option<&str>) {
+    if let Some(value) = value {
+        hasher.update([1]);
+        update_replication_fingerprint_bytes(hasher, value.as_bytes());
+    } else {
+        hasher.update([0]);
+    }
+}
+
 #[derive(Debug, Default)]
 pub struct ObjectInfo {
     pub bucket: String,
@@ -1369,6 +1419,144 @@ impl Clone for ObjectInfo {
 }
 
 impl ObjectInfo {
+    /// Capture the source mutation snapshot used by replication workers when
+    /// publishing terminal status. The semantic fingerprint is recomputed at
+    /// the storage CAS boundary, so an older writer that preserves an unknown
+    /// UUID and collides on the timestamp still cannot hide a payload change.
+    pub(crate) fn replication_generation_snapshot(&self) -> ReplicationGenerationSnapshot {
+        let timestamp = replication_snapshot_internal_value(&self.user_defined, rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP);
+        let mutation_id =
+            replication_snapshot_internal_value(&self.user_defined, rustfs_utils::http::SUFFIX_REPLICATION_GENERATION);
+        let tagging_timestamp =
+            replication_snapshot_internal_value(&self.user_defined, rustfs_utils::http::SUFFIX_TAGGING_TIMESTAMP);
+        let retention_timestamp =
+            replication_snapshot_internal_value(&self.user_defined, rustfs_utils::http::SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP);
+        let legalhold_timestamp =
+            replication_snapshot_internal_value(&self.user_defined, rustfs_utils::http::SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP);
+
+        let invalid = timestamp.is_err()
+            || mutation_id.is_err()
+            || tagging_timestamp.is_err()
+            || retention_timestamp.is_err()
+            || legalhold_timestamp.is_err();
+        let timestamp = timestamp.unwrap_or_default();
+        let mutation_id = mutation_id.unwrap_or_default();
+        let tagging_timestamp = tagging_timestamp.unwrap_or_default();
+        let retention_timestamp = retention_timestamp.unwrap_or_default();
+        let legalhold_timestamp = legalhold_timestamp.unwrap_or_default();
+        let opaque_timestamp_is_invalid = [
+            timestamp.as_deref(),
+            tagging_timestamp.as_deref(),
+            retention_timestamp.as_deref(),
+            legalhold_timestamp.as_deref(),
+        ]
+        .into_iter()
+        .flatten()
+        .any(str::is_empty);
+        let mutation_id_is_invalid = mutation_id.as_deref().is_some_and(|value| {
+            Uuid::parse_str(value)
+                .ok()
+                .filter(|generation| !generation.is_nil())
+                .is_none()
+        });
+        let invalid =
+            invalid || opaque_timestamp_is_invalid || mutation_id_is_invalid || (mutation_id.is_some() && timestamp.is_none());
+
+        let payload_fingerprint = (!invalid).then(|| {
+            self.replication_payload_fingerprint(
+                tagging_timestamp.as_deref(),
+                retention_timestamp.as_deref(),
+                legalhold_timestamp.as_deref(),
+            )
+        });
+
+        ReplicationGenerationSnapshot {
+            timestamp,
+            mutation_id,
+            payload_fingerprint,
+            invalid,
+        }
+    }
+
+    fn replication_payload_fingerprint(
+        &self,
+        tagging_timestamp: Option<&str>,
+        retention_timestamp: Option<&str>,
+        legalhold_timestamp: Option<&str>,
+    ) -> [u8; 32] {
+        let mut hasher = Sha256::new();
+        hasher.update(b"rustfs-replication-payload-v1");
+
+        // A suspended bucket's current null version is represented as
+        // `Some(Uuid::nil())` at the request/queue boundary and as `None`
+        // when the same xl.meta is reread without versioning flags. They are
+        // one persisted identity, so do not let that representation detail
+        // make a worker permanently supersede its own terminal write-back.
+        if let Some(version_id) = self.version_id.filter(|version_id| !version_id.is_nil()) {
+            hasher.update([1]);
+            hasher.update(version_id.as_bytes());
+        } else {
+            hasher.update([0]);
+        }
+        if let Some(data_dir) = self.data_dir {
+            hasher.update([1]);
+            hasher.update(data_dir.as_bytes());
+        } else {
+            hasher.update([0]);
+        }
+        if let Some(mod_time) = self.mod_time {
+            hasher.update([1]);
+            hasher.update(mod_time.unix_timestamp_nanos().to_le_bytes());
+        } else {
+            hasher.update([0]);
+        }
+
+        update_replication_fingerprint_optional_str(&mut hasher, self.content_type.as_deref());
+        update_replication_fingerprint_optional_str(&mut hasher, self.content_encoding.as_deref());
+        update_replication_fingerprint_optional_str(&mut hasher, self.storage_class.as_deref());
+        if let Some(expires) = self.expires {
+            hasher.update([1]);
+            hasher.update(expires.unix_timestamp_nanos().to_le_bytes());
+        } else {
+            hasher.update([0]);
+        }
+        update_replication_fingerprint_bytes(&mut hasher, self.user_tags.as_bytes());
+        update_replication_fingerprint_optional_str(&mut hasher, tagging_timestamp);
+        update_replication_fingerprint_optional_str(&mut hasher, retention_timestamp);
+        update_replication_fingerprint_optional_str(&mut hasher, legalhold_timestamp);
+
+        let mut target_arns = self
+            .replication_status_internal
+            .as_deref()
+            .map(replication_statuses_map)
+            .unwrap_or_default()
+            .into_keys()
+            .collect::<Vec<_>>();
+        target_arns.sort_unstable();
+        hasher.update(u64::try_from(target_arns.len()).unwrap_or(u64::MAX).to_le_bytes());
+        for arn in target_arns {
+            update_replication_fingerprint_bytes(&mut hasher, arn.as_bytes());
+        }
+        update_replication_fingerprint_bytes(&mut hasher, self.replication_decision.as_bytes());
+
+        let mut user_metadata = self
+            .user_defined
+            .iter()
+            .filter(|(key, _)| {
+                !rustfs_utils::http::is_internal_key(key)
+                    && !key.eq_ignore_ascii_case(rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS)
+            })
+            .collect::<Vec<_>>();
+        user_metadata.sort_unstable_by(|left, right| left.0.cmp(right.0).then_with(|| left.1.cmp(right.1)));
+        hasher.update(u64::try_from(user_metadata.len()).unwrap_or(u64::MAX).to_le_bytes());
+        for (key, value) in user_metadata {
+            update_replication_fingerprint_bytes(&mut hasher, key.as_bytes());
+            update_replication_fingerprint_bytes(&mut hasher, value.as_bytes());
+        }
+
+        hasher.finalize().into()
+    }
+
     pub fn is_compressed(&self) -> bool {
         rustfs_utils::http::contains_key_str(&self.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION)
     }
@@ -2001,6 +2189,13 @@ fn versions_after_marker(file_infos: &rustfs_filemeta::FileInfoVersions, marker:
 mod tests {
     use super::*;
 
+    #[test]
+    fn replication_status_writeback_condition_remains_indirected() {
+        fn assert_indirected(_: &Option<Box<ReplicationStatusWritebackCondition>>) {}
+
+        assert_indirected(&ObjectOptions::default().replication_status_writeback);
+    }
+
     #[test]
     fn object_lock_config_snapshot_is_bound_to_store_bucket_and_incarnation() {
         let store_id = Uuid::new_v4();

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -55,6 +55,7 @@ use crate::api::config::storageclass;
 #[cfg(test)]
 use crate::bucket::metadata_sys::ObjectLockConfigState;
 use crate::bucket::quota::reservation;
+use crate::bucket::replication::ReplicationStatusType;
 use crate::crash_inject::{self, CrashPoint};
 use crate::disk::DiskAPI;
 #[cfg(test)]
@@ -2366,6 +2367,43 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         ensure_data_movement_upload_access(&fi, bucket, object, upload_id, opts)?;
         ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
             .await?;
+        // The request layer computes replication admission immediately before
+        // completion. Apply that metadata only after the staged upload and its
+        // bucket incarnation are validated, while the object/upload commit
+        // locks are still held, so generation, PENDING targets, and the final
+        // object become visible as one commit.
+        if let Some(eval_metadata) = &opts.eval_metadata {
+            for suffix in [
+                rustfs_utils::http::SUFFIX_REPLICATION_GENERATION,
+                rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP,
+                rustfs_utils::http::SUFFIX_REPLICATION_STATUS,
+            ] {
+                rustfs_utils::http::remove_str(&mut fi.metadata, suffix);
+            }
+            // A source-side completion creates a new local object identity;
+            // replica bookkeeping carried by an internal/legacy staged upload
+            // belongs to the source object and must not survive. Authorized
+            // inbound replication owns these fields and remains exempt. Older
+            // and heterogeneous replication clients can send an authorized
+            // REPLICA status on Complete without repeating the source-request
+            // marker, so either authenticated parser result is sufficient.
+            let authorized_inbound_replica =
+                opts.replication_request || opts.delete_marker_replication_status() == ReplicationStatusType::Replica;
+            if !authorized_inbound_replica {
+                fi.metadata
+                    .retain(|key, _| !key.eq_ignore_ascii_case(rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS));
+                for suffix in [
+                    rustfs_utils::http::SUFFIX_REPLICA_STATUS,
+                    rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP,
+                ] {
+                    rustfs_utils::http::remove_str(&mut fi.metadata, suffix);
+                }
+            }
+            for (key, value) in eval_metadata {
+                fi.metadata.insert(key.clone(), value.clone());
+            }
+            fi.replication_state_internal = rustfs_filemeta::get_internal_replication_state(&fi.metadata);
+        }
         let has_layout_candidate = range_seek_rollout_enabled
             && fi
                 .data_dir
@@ -2910,6 +2948,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 meta.mod_time = fi.mod_time;
                 meta.parts.clone_from(&fi.parts);
                 meta.metadata = fi.metadata.clone();
+                meta.replication_state_internal = fi.replication_state_internal.clone();
                 meta.versioned = opts.versioned || opts.version_suspended;
                 meta.version_id = fi.version_id;
                 meta.checksum = fi.checksum.clone();
@@ -8248,6 +8287,215 @@ mod tests {
             assert_eq!(current.version_id, Some(version_id));
         }
 
+        #[tokio::test]
+        #[serial]
+        async fn complete_multipart_upload_commits_replication_admission_metadata_atomically() {
+            let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+            let bucket = "multipart-replication-admission-bucket";
+            let object = "object";
+            make_bucket_on_all(&disk_stores, bucket).await;
+
+            let mut create_replication_metadata = HashMap::new();
+            rustfs_utils::http::insert_str(
+                &mut create_replication_metadata,
+                rustfs_utils::http::SUFFIX_REPLICATION_GENERATION,
+                Uuid::from_u128(1).to_string(),
+            );
+            rustfs_utils::http::insert_str(
+                &mut create_replication_metadata,
+                rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP,
+                "create-time".to_string(),
+            );
+            rustfs_utils::http::insert_str(
+                &mut create_replication_metadata,
+                rustfs_utils::http::SUFFIX_REPLICATION_STATUS,
+                "arn:create-target=PENDING;".to_string(),
+            );
+            rustfs_utils::http::insert_str(
+                &mut create_replication_metadata,
+                rustfs_utils::http::SUFFIX_REPLICA_STATUS,
+                "REPLICA".to_string(),
+            );
+            rustfs_utils::http::insert_str(
+                &mut create_replication_metadata,
+                rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP,
+                "foreign-replica-time".to_string(),
+            );
+            create_replication_metadata
+                .insert(rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS.to_string(), "REPLICA".to_string());
+            let (upload_id, parts) = stage_upload_with_create_opts(
+                &set_disks,
+                bucket,
+                object,
+                b"multipart replication body",
+                &ObjectOptions {
+                    user_defined: create_replication_metadata.clone(),
+                    ..Default::default()
+                },
+            )
+            .await;
+            let generation = Uuid::from_u128(42).to_string();
+            let timestamp = "opaque-completion-time";
+            let status = "arn:complete-target=PENDING;";
+            let mut replication_metadata = HashMap::new();
+            rustfs_utils::http::insert_str(
+                &mut replication_metadata,
+                rustfs_utils::http::SUFFIX_REPLICATION_GENERATION,
+                generation.clone(),
+            );
+            rustfs_utils::http::insert_str(
+                &mut replication_metadata,
+                rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP,
+                timestamp.to_string(),
+            );
+            rustfs_utils::http::insert_str(
+                &mut replication_metadata,
+                rustfs_utils::http::SUFFIX_REPLICATION_STATUS,
+                status.to_string(),
+            );
+
+            let completed = set_disks
+                .clone()
+                .complete_multipart_upload(
+                    bucket,
+                    object,
+                    &upload_id,
+                    parts,
+                    &ObjectOptions {
+                        eval_metadata: Some(replication_metadata),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("multipart completion with replication admission should succeed");
+
+            assert_eq!(
+                rustfs_utils::http::get_str(completed.user_defined.as_ref(), rustfs_utils::http::SUFFIX_REPLICATION_GENERATION,)
+                    .as_deref(),
+                Some(generation.as_str())
+            );
+            assert_eq!(
+                rustfs_utils::http::get_str(completed.user_defined.as_ref(), rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP,)
+                    .as_deref(),
+                Some(timestamp)
+            );
+            assert_eq!(completed.replication_status_internal.as_deref(), Some(status));
+            for suffix in [
+                rustfs_utils::http::SUFFIX_REPLICA_STATUS,
+                rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP,
+            ] {
+                assert!(
+                    !rustfs_utils::http::contains_key_str(completed.user_defined.as_ref(), suffix),
+                    "source completion must clear staged foreign {suffix}"
+                );
+            }
+            assert!(
+                completed
+                    .user_defined
+                    .iter()
+                    .filter(|(key, _)| key.eq_ignore_ascii_case(rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS))
+                    .all(|(_, value)| value != "REPLICA")
+            );
+
+            let disabled_object = "disabled";
+            let (upload_id, parts) = stage_upload_with_create_opts(
+                &set_disks,
+                bucket,
+                disabled_object,
+                b"multipart replication disabled body",
+                &ObjectOptions {
+                    user_defined: create_replication_metadata,
+                    ..Default::default()
+                },
+            )
+            .await;
+            let completed = set_disks
+                .clone()
+                .complete_multipart_upload(
+                    bucket,
+                    disabled_object,
+                    &upload_id,
+                    parts,
+                    &ObjectOptions {
+                        eval_metadata: Some(HashMap::new()),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("completion should clear stale create-time replication admission");
+            assert!(completed.replication_status_internal.is_none());
+            for suffix in [
+                rustfs_utils::http::SUFFIX_REPLICATION_GENERATION,
+                rustfs_utils::http::SUFFIX_REPLICATION_TIMESTAMP,
+                rustfs_utils::http::SUFFIX_REPLICATION_STATUS,
+                rustfs_utils::http::SUFFIX_REPLICA_STATUS,
+                rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP,
+            ] {
+                assert!(
+                    !rustfs_utils::http::contains_key_str(completed.user_defined.as_ref(), suffix),
+                    "disabled completion must clear stale {suffix}"
+                );
+            }
+
+            let inbound_object = "inbound";
+            let mut inbound_replica_metadata = HashMap::new();
+            rustfs_utils::http::insert_str(
+                &mut inbound_replica_metadata,
+                rustfs_utils::http::SUFFIX_REPLICA_STATUS,
+                "REPLICA".to_string(),
+            );
+            rustfs_utils::http::insert_str(
+                &mut inbound_replica_metadata,
+                rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP,
+                "authorized-inbound-time".to_string(),
+            );
+            inbound_replica_metadata.insert(rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS.to_string(), "REPLICA".to_string());
+            let (upload_id, parts) = stage_upload_with_create_opts(
+                &set_disks,
+                bucket,
+                inbound_object,
+                b"authorized inbound multipart body",
+                &ObjectOptions {
+                    user_defined: inbound_replica_metadata,
+                    ..Default::default()
+                },
+            )
+            .await;
+            let mut inbound_complete_opts = ObjectOptions {
+                eval_metadata: Some(HashMap::new()),
+                ..Default::default()
+            };
+            inbound_complete_opts.set_replica_status(ReplicationStatusType::Replica);
+            assert!(
+                !inbound_complete_opts.replication_request,
+                "the compatibility path must not depend on a source-request marker"
+            );
+            let inbound = set_disks
+                .clone()
+                .complete_multipart_upload(bucket, inbound_object, &upload_id, parts, &inbound_complete_opts)
+                .await
+                .expect("authorized REPLICA-only completion must preserve replica bookkeeping");
+            assert_eq!(
+                rustfs_utils::http::get_consistent_str(inbound.user_defined.as_ref(), rustfs_utils::http::SUFFIX_REPLICA_STATUS),
+                Some("REPLICA")
+            );
+            assert_eq!(
+                rustfs_utils::http::get_consistent_str(
+                    inbound.user_defined.as_ref(),
+                    rustfs_utils::http::SUFFIX_REPLICA_TIMESTAMP
+                ),
+                Some("authorized-inbound-time")
+            );
+            assert_eq!(
+                inbound
+                    .user_defined
+                    .iter()
+                    .find(|(key, _)| key.eq_ignore_ascii_case(rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS))
+                    .map(|(_, value)| value.as_str()),
+                Some("REPLICA")
+            );
+        }
+
         #[tokio::test]
         #[serial]
         async fn complete_multipart_upload_replaces_staged_nil_version_id() {

crates/filemeta/src/filemeta.rs

@@ -363,6 +363,25 @@ impl FileMeta {
                                 }
                             }
 
+                            // `fi.metadata` is the authoritative replacement
+                            // for every internal suffix it carries. Remove all
+                            // existing RustFS/MinIO and mixed-case aliases for
+                            // those suffixes before inserting the new map;
+                            // otherwise an old MinIO key survives this RMW and
+                            // conflicts with newly written canonical aliases.
+                            let replaced_internal_suffixes = fi
+                                .metadata
+                                .keys()
+                                .filter_map(|key| rustfs_utils::http::strip_internal_prefix_preserving_case(key))
+                                .map(str::to_ascii_lowercase)
+                                .collect::<std::collections::HashSet<_>>();
+                            if !replaced_internal_suffixes.is_empty() {
+                                obj.meta_sys.retain(|key, _| {
+                                    rustfs_utils::http::strip_internal_prefix_preserving_case(key)
+                                        .is_none_or(|suffix| !replaced_internal_suffixes.contains(&suffix.to_ascii_lowercase()))
+                                });
+                            }
+
                             for (k, v) in fi.metadata.iter() {
                                 // Split metadata into meta_user and meta_sys based on prefix
                                 // This logic must match From<FileInfo> for MetaObject

crates/filemeta/src/filemeta/version.rs

@@ -2861,7 +2861,12 @@ impl From<FileInfo> for MetaObject {
     }
 }
 
-fn get_internal_replication_state(metadata: &HashMap<String, String>) -> Option<ReplicationState> {
+/// Rebuild the structured replication state from its durable internal metadata.
+///
+/// Mutation paths that update internal replication keys on an existing
+/// [`FileInfo`] must use this parser before serializing xl.meta so the metadata
+/// map and the structured state cannot diverge.
+pub fn get_internal_replication_state(metadata: &HashMap<String, String>) -> Option<ReplicationState> {
     let mut rs = ReplicationState::default();
     let mut has = false;
 

crates/replication/src/filemeta.rs

@@ -849,6 +849,20 @@ pub fn parse_replicate_decision(_bucket: &str, s: &str) -> std::io::Result<Repli
     // }
 }
 
+/// Source snapshot used to fence replication terminal-status publication.
+///
+/// Timestamp and mutation id remain opaque because supported RustFS/MinIO
+/// writers may use different textual representations. `invalid` preserves the
+/// distinction between truly absent legacy metadata and corrupt/conflicting
+/// compatibility aliases.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ReplicationGenerationSnapshot {
+    pub timestamp: Option<String>,
+    pub mutation_id: Option<String>,
+    pub payload_fingerprint: Option<[u8; 32]>,
+    pub invalid: bool,
+}
+
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct ReplicateObjectInfo {
     pub name: String,
@@ -871,6 +885,10 @@ pub struct ReplicateObjectInfo {
     pub target_statuses: HashMap<String, ReplicationStatusType>,
     pub target_purge_statuses: HashMap<String, VersionPurgeStatusType>,
     pub replication_timestamp: Option<OffsetDateTime>,
+    /// Exact persisted source snapshot used to fence terminal status
+    /// write-back after remote I/O.
+    #[serde(default)]
+    pub replication_generation: ReplicationGenerationSnapshot,
     pub ssec: bool,
     pub user_tags: String,
     pub checksum: Option<Bytes>,

crates/replication/src/http.rs

@@ -85,6 +85,24 @@ pub(crate) fn get_internal_metadata(map: &HashMap<String, String>, suffix: &str)
         })
 }
 
+/// Return a non-empty internal value only when every RustFS/MinIO alias
+/// present for the suffix agrees. A single alias remains valid for rolling
+/// upgrade compatibility; conflicts fail closed at security boundaries.
+pub(crate) fn get_consistent_internal_metadata<'a>(map: &'a HashMap<String, String>, suffix: &str) -> Option<&'a str> {
+    let (rustfs_key, minio_key) = internal_keys(suffix);
+    let mut value: Option<&String> = None;
+    for (key, candidate) in map {
+        if !key.eq_ignore_ascii_case(&rustfs_key) && !key.eq_ignore_ascii_case(&minio_key) {
+            continue;
+        }
+        if candidate.is_empty() || value.is_some_and(|current| current != candidate) {
+            return None;
+        }
+        value = Some(candidate);
+    }
+    value.map(String::as_str)
+}
+
 pub(crate) fn get_header_metadata(map: &HashMap<String, String>, suffix: &str) -> Option<String> {
     let rustfs_key = rustfs_header_key(suffix);
     let minio_key = minio_header_key(suffix);
@@ -114,8 +132,8 @@ pub(crate) fn insert_internal_metadata(map: &mut HashMap<String, String>, suffix
 #[cfg(test)]
 mod tests {
     use super::{
-        SUFFIX_ACTUAL_SIZE, SUFFIX_REPLICATION_RESET_STATUS, get_header_metadata, get_internal_metadata, has_prefix_fold,
-        insert_internal_metadata, internal_key_rustfs, trim_etag,
+        SUFFIX_ACTUAL_SIZE, SUFFIX_REPLICATION_RESET_STATUS, get_consistent_internal_metadata, get_header_metadata,
+        get_internal_metadata, has_prefix_fold, insert_internal_metadata, internal_key_rustfs, trim_etag,
     };
     use std::collections::HashMap;
 
@@ -136,6 +154,38 @@ mod tests {
         assert_eq!(get_internal_metadata(&metadata, SUFFIX_ACTUAL_SIZE).as_deref(), Some("12"));
     }
 
+    #[test]
+    fn consistent_internal_metadata_accepts_single_or_matching_aliases_and_rejects_conflicts() {
+        for key in ["x-rustfs-internal-replication-status", "X-Minio-Internal-Replication-Status"] {
+            let metadata = HashMap::from([(key.to_string(), "arn:target=PENDING;".to_string())]);
+            assert_eq!(
+                get_consistent_internal_metadata(&metadata, "replication-status"),
+                Some("arn:target=PENDING;")
+            );
+        }
+
+        let matching = HashMap::from([
+            ("x-rustfs-internal-replication-status".to_string(), "arn:target=COMPLETED;".to_string()),
+            ("X-Minio-Internal-Replication-Status".to_string(), "arn:target=COMPLETED;".to_string()),
+        ]);
+        assert_eq!(
+            get_consistent_internal_metadata(&matching, "replication-status"),
+            Some("arn:target=COMPLETED;")
+        );
+
+        let conflicting = HashMap::from([
+            ("x-rustfs-internal-replication-status".to_string(), "arn:target=PENDING;".to_string()),
+            ("x-minio-internal-replication-status".to_string(), "arn:target=REPLICA;".to_string()),
+        ]);
+        assert!(get_consistent_internal_metadata(&conflicting, "replication-status").is_none());
+
+        let empty_alias = HashMap::from([
+            ("x-rustfs-internal-replication-status".to_string(), "arn:target=PENDING;".to_string()),
+            ("x-minio-internal-replication-status".to_string(), String::new()),
+        ]);
+        assert!(get_consistent_internal_metadata(&empty_alias, "replication-status").is_none());
+    }
+
     #[test]
     fn internal_metadata_insert_writes_rustfs_and_minio_keys() {
         let mut metadata = HashMap::new();

crates/replication/src/lib.rs

@@ -48,10 +48,10 @@ pub use delete::{
 pub use filemeta::{
     NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING,
     REPLICATE_INCOMING_DELETE, REPLICATE_MRF, REPLICATE_QUEUED, REPLICATION_RESET, REPLICATION_STATUS, ReplicateDecision,
-    ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState,
-    ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision,
-    VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replicate_decision_for_admitted_targets,
-    replication_statuses_map, target_reset_header, version_purge_statuses_map,
+    ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction,
+    ReplicationGenerationSnapshot, ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation,
+    ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state, parse_replicate_decision,
+    replicate_decision_for_admitted_targets, replication_statuses_map, target_reset_header, version_purge_statuses_map,
 };
 pub use mrf::{
     MRF_ENVELOPE_FORMAT, MRF_ENVELOPE_VERSION, MRF_V2_FILE, MRF_V2_FORMAT, MRF_V2_NAMESPACE, MRF_V2_VERSION, MrfCapabilities,

crates/replication/src/operation.rs

@@ -16,7 +16,7 @@ use std::collections::HashMap;
 
 use crate::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER,
-    SUFFIX_REPLICATION_RESET_STATUS, SUFFIX_REPLICATION_STATUS, get_header_metadata, get_internal_metadata,
+    SUFFIX_REPLICATION_RESET_STATUS, SUFFIX_REPLICATION_STATUS, get_consistent_internal_metadata, get_header_metadata,
 };
 use s3s::dto::ReplicationConfiguration;
 use time::OffsetDateTime;
@@ -99,15 +99,34 @@ impl MustReplicateOptions {
             return true;
         }
 
-        get_internal_metadata(&self.meta, SUFFIX_REPLICATION_STATUS)
-            .as_deref()
-            .and_then(|statuses| {
-                statuses.split(';').find_map(|entry| {
-                    let (target_arn, status) = entry.split_once('=')?;
-                    (target_arn == arn).then(|| ReplicationStatusType::from(status))
-                })
-            })
-            == Some(ReplicationStatusType::Completed)
+        let Some(statuses) = get_consistent_internal_metadata(&self.meta, SUFFIX_REPLICATION_STATUS) else {
+            return false;
+        };
+        let mut admitted = None;
+        for entry in statuses.split(';').filter(|entry| !entry.is_empty()) {
+            let Some((target_arn, status)) = entry.split_once('=') else {
+                continue;
+            };
+            if target_arn != arn {
+                continue;
+            }
+            // Per-target admission is authoritative only when the target has
+            // exactly one unambiguous historical status record.
+            if admitted.is_some() {
+                return false;
+            }
+            admitted = Some(ReplicationStatusType::from(status));
+        }
+
+        matches!(
+            admitted,
+            Some(
+                ReplicationStatusType::Pending
+                    | ReplicationStatusType::Failed
+                    | ReplicationStatusType::Completed
+                    | ReplicationStatusType::CompletedLegacy
+            )
+        )
     }
 }
 
@@ -376,22 +395,72 @@ mod tests {
     }
 
     #[test]
-    fn metadata_replication_requires_completed_target_state() {
+    fn metadata_replication_requires_prior_target_admission() {
         let arn = "arn:rustfs:replication:target";
-        for status in ["PENDING", "FAILED"] {
+        for status in ["PENDING", "FAILED", "COMPLETED", "COMPLETE"] {
             let mut meta = HashMap::new();
             insert_internal_metadata(&mut meta, SUFFIX_REPLICATION_STATUS, format!("{arn}={status};"));
             let options = MustReplicateOptions::new(&meta, String::new(), ReplicationType::Metadata, false);
 
-            assert!(!options.metadata_target_is_eligible(arn));
+            assert!(
+                options.metadata_target_is_eligible(arn),
+                "known target state {status} must remain eligible"
+            );
+            assert!(!options.metadata_target_is_eligible("arn:rustfs:replication:missing"));
         }
 
-        let mut meta = HashMap::new();
-        insert_internal_metadata(&mut meta, SUFFIX_REPLICATION_STATUS, format!("{arn}=COMPLETED;"));
-        let options = MustReplicateOptions::new(&meta, String::new(), ReplicationType::Metadata, false);
+        for status in ["", "REPLICA", "UNKNOWN"] {
+            let mut meta = HashMap::new();
+            insert_internal_metadata(&mut meta, SUFFIX_REPLICATION_STATUS, format!("{arn}={status};"));
+            let options = MustReplicateOptions::new(&meta, String::new(), ReplicationType::Metadata, false);
+            assert!(
+                !options.metadata_target_is_eligible(arn),
+                "unsupported target state {status:?} must fail closed"
+            );
+        }
 
-        assert!(options.metadata_target_is_eligible(arn));
-        assert!(!options.metadata_target_is_eligible("arn:rustfs:replication:missing"));
+        for key in ["x-rustfs-internal-replication-status", "X-Minio-Internal-Replication-Status"] {
+            let meta = HashMap::from([(key.to_string(), format!("{arn}=PENDING;"))]);
+            assert!(
+                MustReplicateOptions::new(&meta, String::new(), ReplicationType::Metadata, false)
+                    .metadata_target_is_eligible(arn),
+                "a single compatibility alias must preserve rolling-upgrade admission"
+            );
+        }
+
+        let matching_aliases = HashMap::from([
+            ("x-rustfs-internal-replication-status".to_string(), format!("{arn}=COMPLETED;")),
+            ("X-Minio-Internal-Replication-Status".to_string(), format!("{arn}=COMPLETED;")),
+        ]);
+        assert!(
+            MustReplicateOptions::new(&matching_aliases, String::new(), ReplicationType::Metadata, false)
+                .metadata_target_is_eligible(arn)
+        );
+
+        for conflicting in [
+            HashMap::from([
+                ("x-rustfs-internal-replication-status".to_string(), format!("{arn}=PENDING;")),
+                ("x-minio-internal-replication-status".to_string(), format!("{arn}=REPLICA;")),
+            ]),
+            HashMap::from([
+                ("x-rustfs-internal-replication-status".to_string(), format!("{arn}=PENDING;")),
+                ("x-minio-internal-replication-status".to_string(), String::new()),
+            ]),
+            HashMap::from([(
+                "x-rustfs-internal-replication-status".to_string(),
+                format!("{arn}=PENDING;{arn}=REPLICA;"),
+            )]),
+            HashMap::from([(
+                "x-rustfs-internal-replication-status".to_string(),
+                format!("{arn}=PENDING;{arn}=PENDING;"),
+            )]),
+        ] {
+            assert!(
+                !MustReplicateOptions::new(&conflicting, String::new(), ReplicationType::Metadata, false)
+                    .metadata_target_is_eligible(arn),
+                "conflicting, empty, or duplicate admission evidence must fail closed"
+            );
+        }
 
         // A replica-side metadata edit (active-active) has no per-target
         // internal status; eligibility is left to the ReplicaModifications rule.

crates/utils/src/http/metadata_compat.rs

@@ -75,6 +75,8 @@ pub const SUFFIX_REPLICA_STATUS: &str = "replica-status";
 pub const SUFFIX_REPLICA_TIMESTAMP: &str = "replica-timestamp";
 pub const SUFFIX_REPLICATION_STATUS: &str = "replication-status";
 pub const SUFFIX_REPLICATION_TIMESTAMP: &str = "replication-timestamp";
+/// Source-local opaque mutation id fencing replication status write-back.
+pub const SUFFIX_REPLICATION_GENERATION: &str = "replication-generation";
 pub const SUFFIX_TAGGING_TIMESTAMP: &str = "tagging-timestamp";
 pub const SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP: &str = "objectlock-retention-timestamp";
 pub const SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP: &str = "objectlock-legalhold-timestamp";