fix(replication): keep first-replication miss quiet in multipart LWW log

Version-absent is the normal first delivery of a version, not a
degraded comparison skip; only real read errors (quorum loss) warrant
the warn added in the previous commit.

.github/workflows/package.yml

@@ -189,6 +189,7 @@ jobs:
     timeout-minutes: 30
     strategy:
       fail-fast: false
+      max-parallel: 1
       matrix:
         include:
           - arch: x86_64
@@ -510,15 +511,13 @@ jobs:
 
           CHECKSUM_DIR="$(mktemp -d)"
           gh release download "$TAG" -p 'SHA256SUMS' -p 'SHA512SUMS' \
-            -D "$CHECKSUM_DIR" --clobber 2>/dev/null || true
+            -D "$CHECKSUM_DIR" --clobber
 
           for spec in "SHA256SUMS:sha256sum" "SHA512SUMS:sha512sum"; do
             asset="${spec%%:*}"
             checksum_cmd="${spec##*:}"
             checksum_file="${CHECKSUM_DIR}/${asset}"
 
-            touch "$checksum_file"
-
             for f in "$DEB_FILE" "$RPM_FILE"; do
               if [[ -n "$f" && -f "$f" ]]; then
                 base="$(basename "$f")"
@@ -531,7 +530,8 @@ jobs:
                 grep -Fv -- "$base" "$checksum_file" > "${checksum_file}.tmp" || true
                 grep -Fv -- "$github_base" "${checksum_file}.tmp" > "${checksum_file}.tmp2" || true
                 mv "${checksum_file}.tmp2" "$checksum_file"
-                (cd "$(dirname "$f")" && "$checksum_cmd" -- "$github_base") >> "$checksum_file"
+                digest=$("$checksum_cmd" -- "$f" | awk '{print $1}')
+                printf '%s  %s\n' "$digest" "$github_base" >> "$checksum_file"
               fi
             done
 

crates/e2e_test/src/lib.rs

@@ -61,6 +61,11 @@ mod get_codec_streaming_compat_test;
 #[cfg(test)]
 mod version_id_regression_test;
 
+// Receiver-side replication LWW (rustfs/backlog#1953): stale inbound
+// replication metadata must not overwrite a newer local category state.
+#[cfg(test)]
+mod replication_lww_receiver_test;
+
 // Data usage regression tests
 #[cfg(test)]
 mod data_usage_test;

crates/e2e_test/src/replication_lww_receiver_test.rs

@@ -0,0 +1,152 @@
+#![cfg(test)]
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Receiver-side replication LWW over the wire (rustfs/backlog#1953, audit
+//! A4/P1-6).
+//!
+//! In an active-active topology both sites' metadata states arrive at the
+//! peer as authorized replication PUTs carrying per-category source
+//! timestamps (`x-rustfs-source-replication-tagging-timestamp` header
+//! family). Before the fix the receiver applied them unconditionally, so a
+//! stale delivery overwrote a newer local state and the two sites diverged
+//! permanently while both reported COMPLETED. This test drives one live
+//! `rustfs` server with simulated inbound replication PUTs for the same
+//! object version and asserts the newer tagging state wins regardless of
+//! delivery order, while a stale delivery still succeeds at the object level
+//! (a failure would loop through MRF re-delivering the stale value).
+
+use crate::common::{RustFSTestEnvironment, init_logging};
+use aws_sdk_s3::Client;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
+use serial_test::serial;
+
+type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
+
+const HDR_SOURCE_REPLICATION_REQUEST: &str = "x-rustfs-source-replication-request";
+const HDR_SOURCE_VERSION_ID: &str = "x-rustfs-source-version-id";
+const HDR_SOURCE_MTIME: &str = "x-rustfs-source-mtime";
+const HDR_SOURCE_TAGGING_TIMESTAMP: &str = "x-rustfs-source-replication-tagging-timestamp";
+
+const SOURCE_MTIME: &str = "2026-01-01T00:00:00Z";
+const T_STALE: &str = "2026-01-01T00:00:01Z";
+const T_LOCAL: &str = "2026-02-01T00:00:00Z";
+const T_NEWER: &str = "2026-03-01T00:00:00Z";
+
+/// Simulated inbound authorized replication PUT: same object version, tags and
+/// the source-authored tagging timestamp carried in transport headers.
+async fn inbound_replication_put(
+    client: &Client,
+    bucket: &str,
+    key: &str,
+    version_id: &str,
+    tags: &str,
+    tagging_timestamp: &str,
+) -> TestResult {
+    let version_id = version_id.to_string();
+    let tagging_timestamp = tagging_timestamp.to_string();
+    client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .body(ByteStream::from_static(b"lww-e2e-body"))
+        .tagging(tags)
+        .customize()
+        .mutate_request(move |req| {
+            req.headers_mut().insert(HDR_SOURCE_REPLICATION_REQUEST, "true");
+            req.headers_mut().insert(HDR_SOURCE_VERSION_ID, version_id.clone());
+            req.headers_mut().insert(HDR_SOURCE_MTIME, SOURCE_MTIME);
+            req.headers_mut()
+                .insert(HDR_SOURCE_TAGGING_TIMESTAMP, tagging_timestamp.clone());
+        })
+        .send()
+        .await?;
+    Ok(())
+}
+
+async fn tag_value(client: &Client, bucket: &str, key: &str, version_id: &str, tag_key: &str) -> Option<String> {
+    let tagging = client
+        .get_object_tagging()
+        .bucket(bucket)
+        .key(key)
+        .version_id(version_id)
+        .send()
+        .await
+        .expect("object tagging should be readable");
+    tagging
+        .tag_set()
+        .iter()
+        .find(|tag| tag.key() == tag_key)
+        .map(|tag| tag.value().to_string())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn receiver_lww_keeps_newer_tags_across_delivery_orders() -> TestResult {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    let client = env.create_s3_client();
+
+    let bucket = "replication-lww-receiver";
+    let key = "object";
+    client.create_bucket().bucket(bucket).send().await?;
+    client
+        .put_bucket_versioning()
+        .bucket(bucket)
+        .versioning_configuration(
+            VersioningConfiguration::builder()
+                .status(BucketVersioningStatus::Enabled)
+                .build(),
+        )
+        .send()
+        .await?;
+
+    // First delivery establishes version V with tags stamped T_LOCAL.
+    let version_id = uuid::Uuid::new_v4().to_string();
+    inbound_replication_put(&client, bucket, key, &version_id, "site=local", T_LOCAL).await?;
+    assert_eq!(
+        tag_value(&client, bucket, key, &version_id, "site").await.as_deref(),
+        Some("local"),
+        "the first delivery must establish the tagged version"
+    );
+
+    // A stale delivery (older source timestamp) must succeed at the object
+    // level but must NOT overwrite the newer tags.
+    inbound_replication_put(&client, bucket, key, &version_id, "site=stale", T_STALE).await?;
+    assert_eq!(
+        tag_value(&client, bucket, key, &version_id, "site").await.as_deref(),
+        Some("local"),
+        "a stale inbound delivery must not overwrite newer tags (rustfs/backlog#1953)"
+    );
+
+    // A newer delivery still converges the version onto the newest state.
+    inbound_replication_put(&client, bucket, key, &version_id, "site=newer", T_NEWER).await?;
+    assert_eq!(
+        tag_value(&client, bucket, key, &version_id, "site").await.as_deref(),
+        Some("newer"),
+        "a newer inbound delivery must overwrite older tags"
+    );
+
+    client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(&version_id)
+        .send()
+        .await?;
+    env.delete_test_bucket(bucket).await.ok();
+    Ok(())
+}

crates/ecstore/src/bucket/migration.rs

@@ -41,6 +41,7 @@ const IAM_FORMAT_FILE_PATH: &str = "config/iam/format.json";
 const IAM_USERS_PREFIX: &str = "config/iam/users/";
 const IAM_SERVICE_ACCOUNTS_PREFIX: &str = "config/iam/service-accounts/";
 const IAM_STS_PREFIX: &str = "config/iam/sts/";
+const MINIO_GO_ZERO_TIME: OffsetDateTime = time::macros::datetime!(0001-01-01 00:00 UTC);
 const IAM_GROUPS_PREFIX: &str = "config/iam/groups/";
 const IAM_POLICIES_PREFIX: &str = "config/iam/policies/";
 const IAM_POLICY_DB_PREFIX: &str = "config/iam/policydb/";
@@ -120,6 +121,15 @@ fn normalize_iam_config_blob(path: &str, data: &[u8]) -> std::result::Result<Opt
     if is_identity_path(path) {
         let mut identity: UserIdentity =
             serde_json::from_slice(data).map_err(|err| format!("parse IAM identity failed: {err}"))?;
+        if (path.starts_with(IAM_USERS_PREFIX) || path.starts_with(IAM_SERVICE_ACCOUNTS_PREFIX))
+            && identity
+                .credentials
+                .expiration
+                .as_ref()
+                .is_some_and(|expiration| *expiration == MINIO_GO_ZERO_TIME || *expiration == OffsetDateTime::UNIX_EPOCH)
+        {
+            identity.credentials.expiration = None;
+        }
         if identity.update_at.is_none() {
             identity.update_at = Some(OffsetDateTime::now_utc());
         }
@@ -441,7 +451,10 @@ mod tests {
     use crate::bucket::replication::{
         BucketReplicationResyncStatus, ReplicationMigrationBridge, ResyncStatusType, TargetReplicationResyncStatus,
     };
+    use rustfs_policy::auth::UserIdentity;
     use std::collections::HashMap;
+    use time::OffsetDateTime;
+    use time::format_description::well_known::Rfc3339;
 
     #[test]
     fn test_normalize_policy_mapping_legacy_timestamp_and_fields() {
@@ -493,6 +506,54 @@ mod tests {
         assert!(v.get("updatedAt").is_some(), "normalize should backfill updatedAt");
     }
 
+    #[test]
+    fn test_normalize_minio_permanent_credential_expiration() {
+        let cases = [
+            ("config/iam/users/alice/identity.json", "0001-01-01T00:00:00Z", true),
+            ("config/iam/users/alice/identity.json", "1970-01-01T00:00:00Z", true),
+            ("config/iam/service-accounts/svc/identity.json", "0001-01-01T00:00:00Z", true),
+            ("config/iam/service-accounts/svc/identity.json", "1970-01-01T00:00:00Z", true),
+            ("config/iam/service-accounts/svc/identity.json", "1970-01-01T00:00:00.000000001Z", false),
+            ("config/iam/sts/temp/identity.json", "0001-01-01T00:00:00Z", false),
+            ("config/iam/sts/temp/identity.json", "1970-01-01T00:00:00Z", false),
+            ("config/iam/users/alice/identity.json", "1969-12-31T23:59:59Z", false),
+            ("config/iam/users/alice/identity.json", "1970-01-01T00:00:00.000000001Z", false),
+            ("config/iam/users/alice/identity.json", "0001-01-01T00:00:00.000000001Z", false),
+            ("config/iam/users/alice/identity.json", "2030-01-01T00:00:00Z", false),
+        ];
+
+        for (path, expiration, should_clear) in cases {
+            let input = serde_json::json!({
+                "version": 1,
+                "credentials": {
+                    "accessKey": "test-access",
+                    "secretKey": "test-secret",
+                    "sessionToken": "test-session-token",
+                    "parentUser": "test-parent",
+                    "expiration": expiration,
+                }
+            });
+            let output = normalize_iam_config_blob(path, &serde_json::to_vec(&input).expect("serialize identity fixture"))
+                .expect("normalize should succeed")
+                .expect("identity path should be supported");
+            let identity: UserIdentity = serde_json::from_slice(&output).expect("deserialize normalized identity");
+
+            assert_eq!(identity.credentials.access_key, "test-access");
+            assert_eq!(identity.credentials.secret_key, "test-secret");
+            assert_eq!(identity.credentials.session_token, "test-session-token");
+            assert_eq!(identity.credentials.parent_user, "test-parent");
+            if should_clear {
+                assert_eq!(identity.credentials.expiration, None, "path: {path}, expiration: {expiration}");
+            } else {
+                assert_eq!(
+                    identity.credentials.expiration,
+                    Some(OffsetDateTime::parse(expiration, &Rfc3339).expect("parse expected expiration")),
+                    "path: {path}, expiration: {expiration}"
+                );
+            }
+        }
+    }
+
     #[test]
     fn test_normalize_bucket_meta_blob_resync_reencode() {
         let path = ".buckets/test/.replication/resync.bin";

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

@@ -3868,6 +3868,7 @@ async fn replicate_object_with_multipart<S: ReplicationObjectIO>(ctx: MultipartR
                 actual_size,
                 object_info.etag.clone().unwrap_or_default(),
                 object_info.mod_time,
+                &put_opts.internal,
             ),
         )
         .await

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

@@ -472,6 +472,7 @@ pub(crate) fn replication_complete_multipart_options(
     actual_size: String,
     source_etag: String,
     source_mtime: Option<OffsetDateTime>,
+    source_internal: &AdvancedPutOptions,
 ) -> PutObjectOptions {
     let mut user_metadata = HashMap::new();
     insert_header_map(&mut user_metadata, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, actual_size);
@@ -484,6 +485,14 @@ pub(crate) fn replication_complete_multipart_options(
             // mtime must degrade to epoch so header() suppresses the header
             // instead of asserting the replication time as the object's mtime.
             source_mtime: source_mtime.unwrap_or(OffsetDateTime::UNIX_EPOCH),
+            // Carry the per-category LWW timestamps on the complete request as
+            // well: the receiver's CompleteMultipartUpload options builder
+            // parses the same headers, so the multipart transport gets the
+            // same receiver-side LWW as the single-PUT transport
+            // (rustfs/backlog#1953). Epoch values keep the headers suppressed.
+            tagging_timestamp: source_internal.tagging_timestamp,
+            retention_timestamp: source_internal.retention_timestamp,
+            legalhold_timestamp: source_internal.legalhold_timestamp,
             replication_status: ReplicationStatusType::Replica,
             replication_request: true,
             ..Default::default()
@@ -649,20 +658,39 @@ mod tests {
     #[test]
     fn replication_complete_multipart_options_sets_actual_size() {
         let source_mtime = OffsetDateTime::from_unix_timestamp(1_716_170_000).expect("valid test timestamp");
+        let source_internal = AdvancedPutOptions {
+            tagging_timestamp: OffsetDateTime::from_unix_timestamp(1_716_170_100).expect("valid test timestamp"),
+            retention_timestamp: OffsetDateTime::from_unix_timestamp(1_716_170_200).expect("valid test timestamp"),
+            legalhold_timestamp: OffsetDateTime::from_unix_timestamp(1_716_170_300).expect("valid test timestamp"),
+            ..Default::default()
+        };
         let options = replication_complete_multipart_options(
             "1024".to_string(),
             "0123456789abcdef0123456789abcdef-3".to_string(),
             Some(source_mtime),
+            &source_internal,
         );
         assert_eq!(options.internal.source_etag, "0123456789abcdef0123456789abcdef-3");
         assert_eq!(options.internal.source_mtime, source_mtime);
 
+        // The complete request must carry the same per-category LWW timestamps
+        // as the initiate request; the receiver reads them from the complete
+        // headers (rustfs/backlog#1953).
+        assert_eq!(options.internal.tagging_timestamp, source_internal.tagging_timestamp);
+        assert_eq!(options.internal.retention_timestamp, source_internal.retention_timestamp);
+        assert_eq!(options.internal.legalhold_timestamp, source_internal.legalhold_timestamp);
+
         // Absent source mtime must degrade to epoch (header suppressed), not
         // the AdvancedPutOptions default of now_utc() — that default would
         // stamp the replication time as the replica's mtime and break the
-        // multipart HEAD convergence.
-        let options_no_mtime = replication_complete_multipart_options("1024".to_string(), String::new(), None);
+        // multipart HEAD convergence. Unset category timestamps stay epoch so
+        // header() keeps suppressing them.
+        let options_no_mtime =
+            replication_complete_multipart_options("1024".to_string(), String::new(), None, &AdvancedPutOptions::default());
         assert_eq!(options_no_mtime.internal.source_mtime.unix_timestamp(), 0);
+        assert_eq!(options_no_mtime.internal.tagging_timestamp.unix_timestamp(), 0);
+        assert_eq!(options_no_mtime.internal.retention_timestamp.unix_timestamp(), 0);
+        assert_eq!(options_no_mtime.internal.legalhold_timestamp.unix_timestamp(), 0);
 
         assert_eq!(
             get_header_map(&options.user_metadata, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE).as_deref(),

crates/ecstore/src/core/pools.rs

@@ -94,6 +94,9 @@ const DECOMMISSION_BUCKET_CONCURRENCY_DEFAULT_CAP: usize = 4;
 const DECOMMISSION_TARGET_CAPACITY_OVERHEAD_PERCENT: usize = 30;
 const DECOMMISSION_LISTING_MAX_ATTEMPTS: usize = 3;
 const DECOMMISSION_LISTING_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(5);
+/// Background decommission walks must tolerate slow object migrations; the
+/// stall timeout is the drive-health bound, not the total listing duration.
+const DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
 
 pub const POOL_META_NAME: &str = "pool.bin";
 pub const POOL_META_FORMAT: u16 = 1;
@@ -5047,6 +5050,8 @@ impl SetDisks {
                 path: bucket_info.prefix.clone(),
                 recursive: true,
                 min_disks: listing_quorum,
+                skip_walkdir_total_timeout: true,
+                walkdir_stall_timeout: Some(DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT),
                 agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))),
                 partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
                     let resolver = resolver.clone();

crates/ecstore/src/disk/os.rs

@@ -23,7 +23,7 @@ use std::{
     io,
     path::{Component, Path, PathBuf},
     sync::{Arc, LazyLock, Weak},
-    time::Instant,
+    time::{Duration, Instant},
 };
 use tokio::fs;
 use tokio::sync::{
@@ -328,6 +328,9 @@ const ENV_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE: &str = "RUSTFS_EXPERIMENTAL_DST_DIR
 const DEFAULT_DST_DIR_FSYNC_GROUP_COMMIT_ENABLE: bool = false;
 const ENV_FILE_FDATASYNC_GROUP_COMMIT_ENABLE: &str = "RUSTFS_EXPERIMENTAL_FILE_FDATASYNC_GROUP_COMMIT_ENABLE";
 const DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_ENABLE: bool = false;
+const ENV_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS: &str = "RUSTFS_EXPERIMENTAL_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS";
+const DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS: u64 = 0;
+const MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS: u64 = 1_000;
 #[cfg(not(test))]
 const MAX_DST_DIR_FSYNC_GROUPS: usize = 1024;
 #[cfg(test)]
@@ -354,6 +357,16 @@ static DST_DIR_FSYNC_GROUP_COMMIT_ENABLED: LazyLock<bool> = LazyLock::new(|| {
 static FILE_FDATASYNC_GROUP_COMMIT_ENABLED: LazyLock<bool> = LazyLock::new(|| {
     rustfs_utils::get_env_bool(ENV_FILE_FDATASYNC_GROUP_COMMIT_ENABLE, DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_ENABLE)
 });
+fn file_fdatasync_group_commit_wait_duration(wait_micros: u64) -> Duration {
+    Duration::from_micros(wait_micros.min(MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS))
+}
+
+static FILE_FDATASYNC_GROUP_COMMIT_WAIT: LazyLock<Duration> = LazyLock::new(|| {
+    file_fdatasync_group_commit_wait_duration(rustfs_utils::get_env_u64(
+        ENV_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS,
+        DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS,
+    ))
+});
 
 #[cfg(test)]
 mod dst_dir_fsync_group_commit_override {
@@ -402,6 +415,7 @@ mod file_fdatasync_group_commit_override {
     use std::sync::{Mutex, MutexGuard, PoisonError, RwLock};
 
     static OVERRIDE: RwLock<Option<bool>> = RwLock::new(None);
+    static WAIT_OVERRIDE_MICROS: RwLock<Option<u64>> = RwLock::new(None);
     static SERIAL: Mutex<()> = Mutex::new(());
 
     pub(crate) fn get() -> Option<bool> {
@@ -415,6 +429,7 @@ mod file_fdatasync_group_commit_override {
     impl Drop for OverrideGuard {
         fn drop(&mut self) {
             *OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = None;
+            *WAIT_OVERRIDE_MICROS.write().unwrap_or_else(PoisonError::into_inner) = None;
         }
     }
 
@@ -423,6 +438,14 @@ mod file_fdatasync_group_commit_override {
         *OVERRIDE.write().unwrap_or_else(PoisonError::into_inner) = Some(enabled);
         OverrideGuard { _serial: serial }
     }
+
+    pub(crate) fn set_wait_micros(wait_micros: u64) {
+        *WAIT_OVERRIDE_MICROS.write().unwrap_or_else(PoisonError::into_inner) = Some(wait_micros);
+    }
+
+    pub(crate) fn wait_micros() -> Option<u64> {
+        *WAIT_OVERRIDE_MICROS.read().unwrap_or_else(PoisonError::into_inner)
+    }
 }
 
 #[cfg(test)]
@@ -430,6 +453,11 @@ pub(crate) fn set_file_fdatasync_group_commit_for_test(enabled: bool) -> file_fd
     file_fdatasync_group_commit_override::set(enabled)
 }
 
+#[cfg(test)]
+fn set_file_fdatasync_group_commit_wait_for_test(wait_micros: u64) {
+    file_fdatasync_group_commit_override::set_wait_micros(wait_micros);
+}
+
 fn file_fdatasync_group_commit_enabled() -> bool {
     #[cfg(test)]
     if let Some(enabled) = file_fdatasync_group_commit_override::get() {
@@ -439,6 +467,15 @@ fn file_fdatasync_group_commit_enabled() -> bool {
     *FILE_FDATASYNC_GROUP_COMMIT_ENABLED
 }
 
+fn file_fdatasync_group_commit_wait() -> Duration {
+    #[cfg(test)]
+    if let Some(wait_micros) = file_fdatasync_group_commit_override::wait_micros() {
+        return file_fdatasync_group_commit_wait_duration(wait_micros);
+    }
+
+    *FILE_FDATASYNC_GROUP_COMMIT_WAIT
+}
+
 #[derive(Clone, Eq, Hash, PartialEq)]
 struct DstDirFsyncGroupKey {
     canonical_path: PathBuf,
@@ -934,6 +971,10 @@ async fn run_file_fdatasync_group_worker(group: Arc<FileFdatasyncGroup>) {
         #[cfg(test)]
         file_sync_probe::run_before_group_batch();
         tokio::task::yield_now().await;
+        let wait = file_fdatasync_group_commit_wait();
+        if !wait.is_zero() {
+            tokio::time::sleep(wait).await;
+        }
         let (batch, batch_file_count): (Vec<FileFdatasyncWaiter>, usize) = {
             let mut group_state = group.inner.lock();
             let batch_file_count = group_state.pending_files;
@@ -6075,6 +6116,7 @@ mod tests {
         use std::sync::mpsc;
 
         let _group_commit = set_file_fdatasync_group_commit_for_test(true);
+        set_file_fdatasync_group_commit_wait_for_test(0);
         clear_file_fdatasync_group_commit_for_test();
         let temp_dir = tempdir().expect("create temp dir");
         let first_dir = temp_dir.path().join("first");
@@ -6141,12 +6183,105 @@ mod tests {
         assert_eq!(file_fdatasync_group_commit_counts_for_test(), (0, 0, 0));
     }
 
+    #[test]
+    fn file_fdatasync_group_commit_wait_duration_uses_default_and_cap() {
+        assert_eq!(DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS, 0);
+        assert_eq!(
+            file_fdatasync_group_commit_wait_duration(DEFAULT_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS),
+            Duration::ZERO
+        );
+        assert_eq!(file_fdatasync_group_commit_wait_duration(250), Duration::from_micros(250));
+        assert_eq!(
+            file_fdatasync_group_commit_wait_duration(MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS),
+            Duration::from_micros(MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS)
+        );
+        assert_eq!(
+            file_fdatasync_group_commit_wait_duration(u64::MAX),
+            Duration::from_micros(MAX_FILE_FDATASYNC_GROUP_COMMIT_WAIT_MICROS)
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread", start_paused = true)]
+    #[serial_test::serial(file_sync_probe)]
+    async fn file_fdatasync_group_commit_wait_budget_batches_late_follower() {
+        use std::sync::mpsc;
+
+        let _group_commit = set_file_fdatasync_group_commit_for_test(true);
+        let wait_budget_micros = 1_000;
+        let wait_budget = file_fdatasync_group_commit_wait_duration(wait_budget_micros);
+        set_file_fdatasync_group_commit_wait_for_test(wait_budget_micros);
+        clear_file_fdatasync_group_commit_for_test();
+        let temp_dir = tempdir().expect("create temp dir");
+        let first_dir = temp_dir.path().join("first");
+        let second_dir = temp_dir.path().join("second");
+        std::fs::create_dir(&first_dir).expect("create first dir");
+        std::fs::create_dir(&second_dir).expect("create second dir");
+        std::fs::write(first_dir.join("part.1"), b"first").expect("write first part");
+        std::fs::write(second_dir.join("part.1"), b"second").expect("write second part");
+        let _probe = file_sync_probe::set_blocking(temp_dir.path());
+        let (entered_tx, entered_rx) = mpsc::channel();
+        file_sync_probe::set_before_group_batch(move || {
+            entered_tx.send(()).expect("signal first file fdatasync group worker");
+        });
+
+        let limiter = file_sync_limiter();
+        let first_limiter = limiter.clone();
+        let first_path = first_dir.clone();
+        let first = tokio::spawn(async move { sync_dir_files_with_limiter(first_path, first_limiter).await });
+        tokio::task::spawn_blocking(move || entered_rx.recv_timeout(Duration::from_secs(30)))
+            .await
+            .expect("group worker hook waiter should run")
+            .expect("first file fdatasync group worker should start");
+
+        let second_limiter = limiter.clone();
+        let second_path = second_dir.clone();
+        let second = tokio::spawn(async move { sync_dir_files_with_limiter(second_path, second_limiter).await });
+        tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                if file_fdatasync_group_commit_counts_for_test().1 == 2 {
+                    return;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("second waiter should enqueue during the configured wait budget");
+        tokio::time::advance(wait_budget).await;
+        tokio::task::yield_now().await;
+        file_sync_probe::wait_for_active(1).await;
+
+        assert_eq!(
+            file_sync_probe::group_batches(),
+            vec![2],
+            "configured wait budget should let a follower join the leader's batch"
+        );
+        file_sync_probe::release();
+        first
+            .await
+            .expect("join first wait-budget file sync")
+            .expect("first wait-budget file sync must succeed");
+        second
+            .await
+            .expect("join second wait-budget file sync")
+            .expect("second wait-budget file sync must succeed");
+        assert!(
+            fsync_dir_recorder::was_fsynced(&first_dir),
+            "first source directory must still be fsynced"
+        );
+        assert!(
+            fsync_dir_recorder::was_fsynced(&second_dir),
+            "second source directory must still be fsynced"
+        );
+        assert_eq!(file_fdatasync_group_commit_counts_for_test(), (0, 0, 0));
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     #[serial_test::serial(file_sync_probe)]
     async fn file_fdatasync_group_commit_failure_fails_all_waiters_before_dir_fsync() {
         use std::sync::mpsc;
 
         let _group_commit = set_file_fdatasync_group_commit_for_test(true);
+        set_file_fdatasync_group_commit_wait_for_test(0);
         clear_file_fdatasync_group_commit_for_test();
         let temp_dir = tempdir().expect("create temp dir");
         let first_dir = temp_dir.path().join("first");

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

@@ -2223,6 +2223,57 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             fi.set_data_moved();
         }
 
+        // Receiver-side LWW (rustfs/backlog#1953): the multipart replication
+        // transport carries the category values at CreateMultipartUpload (in
+        // the staged upload metadata) and the source category timestamps on
+        // the complete request. Read the destination version under the held
+        // object write lock and keep any category this site modified more
+        // recently. Read failures (version absent on first replication, quorum
+        // errors) keep today's overwrite semantics: failing the complete would
+        // loop through MRF, re-delivering the stale value forever.
+        if crate::set_disk::ops::object::replication_lww_applicable(opts)
+            && let Some(version_id) = fi.version_id
+        {
+            match self
+                .get_object_info(
+                    bucket,
+                    object,
+                    &ObjectOptions {
+                        version_id: Some(version_id.to_string()),
+                        no_lock: true,
+                        metadata_cache_safe: false,
+                        versioned: opts.versioned,
+                        version_suspended: opts.version_suspended,
+                        ..Default::default()
+                    },
+                )
+                .await
+            {
+                Ok(existing) => {
+                    let stored = crate::set_disk::ops::object::stored_replication_category_metadata(&existing);
+                    crate::set_disk::ops::object::merge_replication_metadata_lww(&mut fi.metadata, &stored, opts);
+                }
+                // Version absent: first replication of this version, nothing
+                // local to compare — the normal path, not a degraded one.
+                Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {}
+                Err(err) => {
+                    // Degraded path: without the stored state the inbound
+                    // metadata is applied unchanged — exactly the overwrite
+                    // LWW exists to prevent — so this must be operator-visible.
+                    warn!(
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_SET_DISK,
+                        bucket,
+                        object,
+                        version_id = %version_id,
+                        error = %err,
+                        state = "replication_lww_read_unavailable",
+                        "SetDisk multipart replication LWW read skipped; inbound metadata applied without comparison"
+                    );
+                }
+            }
+        }
+
         for meta in parts_metadatas.iter_mut() {
             if meta.has_valid_erasure_geometry() {
                 meta.size = fi.size;
@@ -6960,6 +7011,97 @@ mod tests {
                 .await
         }
 
+        /// Receiver-side LWW on the multipart replication transport
+        /// (rustfs/backlog#1953): a metadata-only replication of a multipart
+        /// source object rides CreateMultipartUpload (category values in the
+        /// upload metadata) + CompleteMultipartUpload (category timestamps in
+        /// the complete options). A stale inbound tagging timestamp must not
+        /// overwrite a newer locally-tagged destination version.
+        #[tokio::test]
+        #[serial]
+        async fn complete_multipart_upload_stale_replication_tags_keep_local() {
+            use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
+            use rustfs_utils::http::{SUFFIX_TAGGING_TIMESTAMP, get_str};
+            use time::format_description::well_known::Rfc3339;
+
+            const T_OLD: &str = "2026-01-01T00:00:00Z";
+            const T_LOCAL: &str = "2026-02-01T00:00:00Z";
+
+            let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+            let bucket = "multipart-replication-lww-bucket";
+            let object = "object";
+            make_bucket_on_all(&disk_stores, bucket).await;
+
+            // Local destination version with newer tags.
+            let version_id = Uuid::new_v4();
+            let mut local_metadata = HashMap::new();
+            local_metadata.insert(AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string());
+            rustfs_utils::http::insert_str(&mut local_metadata, SUFFIX_TAGGING_TIMESTAMP, T_LOCAL.to_string());
+            let mut local_reader = PutObjReader::from_vec(b"local body".to_vec());
+            set_disks
+                .put_object(
+                    bucket,
+                    object,
+                    &mut local_reader,
+                    &ObjectOptions {
+                        versioned: true,
+                        version_id: Some(version_id.to_string()),
+                        user_defined: local_metadata,
+                        // Explicit-version PUTs require the bucket Object Lock snapshot.
+                        object_lock_config_snapshot: Some(Arc::new(crate::set_disk::ObjectLockConfigSnapshot::new(
+                            crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+                        ))),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("local versioned put should commit");
+
+            // Inbound replication upload carrying older tags for the same version.
+            let mut inbound_metadata = HashMap::new();
+            inbound_metadata.insert(AMZ_OBJECT_TAGGING.to_string(), "site=remote".to_string());
+            rustfs_utils::http::insert_str(&mut inbound_metadata, SUFFIX_TAGGING_TIMESTAMP, T_OLD.to_string());
+            let create_opts = ObjectOptions {
+                versioned: true,
+                user_defined: inbound_metadata,
+                ..Default::default()
+            };
+            let (upload_id, parts) =
+                stage_upload_with_create_opts(&set_disks, bucket, object, &payload(0x5a), &create_opts).await;
+            rewrite_staged_upload_version_id(&set_disks, bucket, object, &upload_id, Some(version_id)).await;
+
+            let complete_opts = ObjectOptions {
+                versioned: true,
+                replication_request: true,
+                replication_tagging_timestamp: Some(OffsetDateTime::parse(T_OLD, &Rfc3339).expect("test timestamp should parse")),
+                ..Default::default()
+            };
+            set_disks
+                .clone()
+                .complete_multipart_upload(bucket, object, &upload_id, parts, &complete_opts)
+                .await
+                .expect("replication multipart completion should succeed even when a category keeps local values");
+
+            let info = set_disks
+                .get_object_info(
+                    bucket,
+                    object,
+                    &ObjectOptions {
+                        versioned: true,
+                        version_id: Some(version_id.to_string()),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("completed version should be readable");
+            assert_eq!(
+                info.user_tags.as_str(),
+                "site=local",
+                "older inbound multipart tags must not overwrite newer local tags"
+            );
+            assert_eq!(get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(), Some(T_LOCAL));
+        }
+
         #[tokio::test]
         #[serial]
         async fn complete_multipart_upload_assigns_completion_version_id() {

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

@@ -1880,6 +1880,110 @@ fn delete_file_info_with_replication_transport_metadata(fi: &FileInfo) -> FileIn
     transported
 }
 
+/// True when an authorized replication write carries at least one per-category
+/// source timestamp, i.e. receiver-side LWW has something to judge.
+pub(in crate::set_disk) fn replication_lww_applicable(opts: &ObjectOptions) -> bool {
+    opts.replication_request
+        && (opts.replication_tagging_timestamp.is_some()
+            || opts.replication_retention_timestamp.is_some()
+            || opts.replication_legalhold_timestamp.is_some())
+}
+
+/// The stored per-category state of a destination version, as compared by
+/// [`merge_replication_metadata_lww`]. `ObjectInfo::from_file_info`
+/// externalizes tags into `user_tags` (stripping the metadata key), so the
+/// tag value is folded back into map form here.
+pub(in crate::set_disk) fn stored_replication_category_metadata(existing: &ObjectInfo) -> HashMap<String, String> {
+    let mut stored = (*existing.user_defined).clone();
+    if !existing.user_tags.is_empty() {
+        stored.insert(rustfs_utils::http::headers::AMZ_OBJECT_TAGGING.to_string(), (*existing.user_tags).clone());
+    }
+    stored
+}
+
+/// Receiver-side last-writer-wins for authorized replication writes
+/// (rustfs/backlog#1953, audit A4/P1-6). Metadata-only replication reuses the
+/// whole-object transports, so in active-active topologies an inbound write
+/// carries the source's tags / retention / legal hold verbatim and would
+/// otherwise overwrite a category the destination modified more recently —
+/// both sites end up permanently diverged while reporting COMPLETED.
+///
+/// Judged per category, only when the inbound request carries that category's
+/// source timestamp (`ObjectOptions::replication_*_timestamp`):
+/// - stored timestamp newer than inbound: the local category values and
+///   timestamp are kept; the rest of the write proceeds per the inbound
+///   metadata and the object-level result stays successful (failing the write
+///   instead would loop through MRF, re-delivering the stale value forever);
+/// - otherwise the inbound category wins and its internal timestamp key is
+///   pinned to the source-authored time — the PUT path re-stamps the
+///   object-lock timestamps with the receiver's clock
+///   (`parse_object_lock_retention` / `parse_object_lock_legal_hold` insert
+///   `now()` via `eval_metadata`), which would make the replica's clock the
+///   LWW authority and wedge later convergence;
+/// - no stored timestamp (pre-P1-6 data) or no inbound timestamp: the current
+///   overwrite behavior is preserved.
+///
+/// Returns whether `inbound` was modified. Callers must hold the object write
+/// lock so the stored values compared here are the ones being replaced.
+pub(in crate::set_disk) fn merge_replication_metadata_lww(
+    inbound: &mut HashMap<String, String>,
+    existing: &HashMap<String, String>,
+    opts: &ObjectOptions,
+) -> bool {
+    use rustfs_utils::http::headers::{
+        AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, AMZ_OBJECT_TAGGING,
+    };
+    use rustfs_utils::http::metadata_compat::{
+        SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP, get_str,
+        remove_str,
+    };
+    use time::format_description::well_known::Rfc3339;
+
+    let categories: [(Option<OffsetDateTime>, &str, &[&str]); 3] = [
+        (opts.replication_tagging_timestamp, SUFFIX_TAGGING_TIMESTAMP, &[AMZ_OBJECT_TAGGING]),
+        (
+            opts.replication_retention_timestamp,
+            SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP,
+            &[AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER],
+        ),
+        (
+            opts.replication_legalhold_timestamp,
+            SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
+            &[AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER],
+        ),
+    ];
+
+    let mut changed = false;
+    for (inbound_timestamp, timestamp_suffix, value_keys) in categories {
+        let Some(inbound_timestamp) = inbound_timestamp else { continue };
+        let is_category_value_key = |key: &str| value_keys.iter().any(|value_key| key.eq_ignore_ascii_case(value_key));
+        let stored_timestamp = get_str(existing, timestamp_suffix).and_then(|value| OffsetDateTime::parse(&value, &Rfc3339).ok());
+        if stored_timestamp.is_some_and(|stored| stored > inbound_timestamp) {
+            inbound.retain(|key, _| !is_category_value_key(key));
+            remove_str(inbound, timestamp_suffix);
+            for (key, value) in existing {
+                if is_category_value_key(key) {
+                    inbound.insert(key.clone(), value.clone());
+                }
+            }
+            // Restore the winning timestamp via insert_str, not a verbatim key
+            // copy: a MinIO-written version may carry only the
+            // x-minio-internal- key, and the dual-key invariant requires every
+            // write to produce both keys.
+            if let Some(stored_value) = get_str(existing, timestamp_suffix) {
+                rustfs_utils::http::insert_str(inbound, timestamp_suffix, stored_value);
+            }
+            changed = true;
+        } else if let Ok(source_authored) = inbound_timestamp.format(&Rfc3339)
+            && get_str(inbound, timestamp_suffix).as_deref() != Some(source_authored.as_str())
+        {
+            rustfs_utils::http::insert_str(inbound, timestamp_suffix, source_authored);
+            changed = true;
+        }
+    }
+    changed
+}
+
 impl SetDisks {
     pub(in crate::set_disk) async fn persist_old_data_cleanup_receipts(
         &self,
@@ -2560,6 +2664,22 @@ impl SetDisks {
                         if check_object_lock_for_deletion_with_state(object_lock_config.state(), &existing, false)?.is_some() {
                             return Err(StorageError::PrefixAccessDenied(bucket.to_string(), object.to_string()));
                         }
+                        // Receiver-side LWW (rustfs/backlog#1953): reuse this
+                        // commit-lock read of the destination version so a
+                        // category (tags / retention / legal hold) modified
+                        // more recently on this site is kept instead of being
+                        // overwritten by the inbound replication metadata.
+                        if replication_lww_applicable(opts) {
+                            let stored = stored_replication_category_metadata(&existing);
+                            let mut merged = parts_metadatas[response_metadata_slot].metadata.clone();
+                            if merge_replication_metadata_lww(&mut merged, &stored, opts) {
+                                for (pfi, disk) in parts_metadatas.iter_mut().zip(shuffle_disks.iter()) {
+                                    if disk.is_some() {
+                                        pfi.metadata = merged.clone();
+                                    }
+                                }
+                            }
+                        }
                     }
                     Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {}
                     Err(err) => return Err(err),
@@ -7887,6 +8007,357 @@ mod replication_quota_safety_tests {
     }
 }
 
+#[cfg(test)]
+mod replication_lww_tests {
+    //! Receiver-side LWW for authorized replication writes (rustfs/backlog#1953,
+    //! audit A4/P1-6): an inbound replication PUT whose per-category timestamp
+    //! (tags / retention / legal hold) is older than the destination version's
+    //! stored timestamp must keep the local category values instead of
+    //! overwriting them; categories are judged independently and the write
+    //! itself still succeeds.
+
+    use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
+    use super::*;
+    use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
+    use rustfs_utils::http::headers::{
+        AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, AMZ_OBJECT_TAGGING,
+    };
+    use rustfs_utils::http::{
+        SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_TAGGING_TIMESTAMP, get_str,
+        insert_str,
+    };
+    use time::format_description::well_known::Rfc3339;
+
+    const T_OLD: &str = "2026-01-01T00:00:00Z";
+    const T_LOCAL: &str = "2026-02-01T00:00:00Z";
+    const T_NEW: &str = "2026-03-01T00:00:00Z";
+
+    fn parse_ts(value: &str) -> OffsetDateTime {
+        OffsetDateTime::parse(value, &Rfc3339).expect("test timestamp should parse")
+    }
+
+    async fn make_bucket(disks: &[DiskStore], bucket: &str) {
+        for disk in disks {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+    }
+
+    async fn put_version(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, version_id: &str, opts: &ObjectOptions) {
+        let mut reader = PutObjReader::from_vec(b"lww-body".to_vec());
+        set_disks
+            .put_object(bucket, object, &mut reader, opts)
+            .await
+            .expect("versioned put should commit");
+        assert_eq!(opts.version_id.as_deref(), Some(version_id));
+    }
+
+    fn versioned_opts(version_id: &str, user_defined: HashMap<String, String>) -> ObjectOptions {
+        ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id.to_string()),
+            user_defined,
+            // Explicit-version PUTs require the bucket Object Lock snapshot.
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
+                crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+            ))),
+            ..Default::default()
+        }
+    }
+
+    /// Local state: version `version_id` with tags "site=local" stamped `T_LOCAL`.
+    async fn seed_local_tagged_version(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, version_id: &str) {
+        let mut user_defined = HashMap::new();
+        user_defined.insert(AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string());
+        insert_str(&mut user_defined, SUFFIX_TAGGING_TIMESTAMP, T_LOCAL.to_string());
+        put_version(set_disks, bucket, object, version_id, &versioned_opts(version_id, user_defined)).await;
+    }
+
+    fn inbound_tagging_opts(version_id: &str, tags: &str, timestamp: &str) -> ObjectOptions {
+        let mut user_defined = HashMap::new();
+        user_defined.insert(AMZ_OBJECT_TAGGING.to_string(), tags.to_string());
+        insert_str(&mut user_defined, SUFFIX_TAGGING_TIMESTAMP, timestamp.to_string());
+        ObjectOptions {
+            replication_request: true,
+            replication_tagging_timestamp: Some(parse_ts(timestamp)),
+            ..versioned_opts(version_id, user_defined)
+        }
+    }
+
+    async fn version_info(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, version_id: &str) -> ObjectInfo {
+        set_disks
+            .get_object_info(bucket, object, &versioned_opts(version_id, HashMap::new()))
+            .await
+            .expect("version should be readable")
+    }
+
+    #[tokio::test]
+    async fn inbound_stale_tagging_keeps_newer_local_tags() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-tagging-stale";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+        seed_local_tagged_version(&set_disks, bucket, object, &version_id).await;
+
+        put_version(
+            &set_disks,
+            bucket,
+            object,
+            &version_id,
+            &inbound_tagging_opts(&version_id, "site=remote", T_OLD),
+        )
+        .await;
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(
+            info.user_tags.as_str(),
+            "site=local",
+            "older inbound tags must not overwrite newer local tags"
+        );
+        assert_eq!(
+            get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(),
+            Some(T_LOCAL),
+            "the winning local tagging timestamp must be preserved"
+        );
+    }
+
+    #[tokio::test]
+    async fn inbound_newer_tagging_overwrites_local_tags() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-tagging-newer";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+        seed_local_tagged_version(&set_disks, bucket, object, &version_id).await;
+
+        put_version(
+            &set_disks,
+            bucket,
+            object,
+            &version_id,
+            &inbound_tagging_opts(&version_id, "site=remote", T_NEW),
+        )
+        .await;
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(
+            info.user_tags.as_str(),
+            "site=remote",
+            "newer inbound tags must overwrite older local tags"
+        );
+        assert_eq!(get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(), Some(T_NEW));
+    }
+
+    #[tokio::test]
+    async fn inbound_wins_when_local_has_no_tagging_timestamp() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-tagging-no-local-ts";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+        // Pre-P1-6 data: local tags without a stored tagging timestamp.
+        let mut user_defined = HashMap::new();
+        user_defined.insert(AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string());
+        put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, user_defined)).await;
+
+        put_version(
+            &set_disks,
+            bucket,
+            object,
+            &version_id,
+            &inbound_tagging_opts(&version_id, "site=remote", T_OLD),
+        )
+        .await;
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(
+            info.user_tags.as_str(),
+            "site=remote",
+            "without a local timestamp the inbound category must win (pre-LWW data compatibility)"
+        );
+    }
+
+    #[tokio::test]
+    async fn categories_are_judged_independently() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-category-independent";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+
+        // Local: newer tags (T_LOCAL), older *cleared* retention (T_OLD) —
+        // timestamp key only, the shape a replicated retention clear stores.
+        // (An active local retention would already block the overwrite at the
+        // WORM gate; the LWW-reachable retention states are cleared/expired.)
+        let mut local = HashMap::new();
+        local.insert(AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string());
+        insert_str(&mut local, SUFFIX_TAGGING_TIMESTAMP, T_LOCAL.to_string());
+        insert_str(&mut local, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_OLD.to_string());
+        put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
+
+        // Inbound: older tags (T_OLD), newer retention (T_NEW).
+        let mut inbound = HashMap::new();
+        inbound.insert(AMZ_OBJECT_TAGGING.to_string(), "site=remote".to_string());
+        insert_str(&mut inbound, SUFFIX_TAGGING_TIMESTAMP, T_OLD.to_string());
+        inbound.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "COMPLIANCE".to_string());
+        inbound.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2028-01-01T00:00:00Z".to_string());
+        insert_str(&mut inbound, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_NEW.to_string());
+        let opts = ObjectOptions {
+            replication_request: true,
+            replication_tagging_timestamp: Some(parse_ts(T_OLD)),
+            replication_retention_timestamp: Some(parse_ts(T_NEW)),
+            ..versioned_opts(&version_id, inbound)
+        };
+        put_version(&set_disks, bucket, object, &version_id, &opts).await;
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(info.user_tags.as_str(), "site=local", "the stale tagging category must keep local values");
+        assert_eq!(
+            info.user_defined.get(AMZ_OBJECT_LOCK_MODE_LOWER).map(String::as_str),
+            Some("COMPLIANCE"),
+            "the newer retention category must be applied in the same write"
+        );
+        assert_eq!(get_str(&info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP).as_deref(), Some(T_NEW));
+    }
+
+    #[tokio::test]
+    async fn inbound_stale_legal_hold_keeps_local_value() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-legalhold-stale";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+
+        // Local: legal hold released (OFF) at T_LOCAL. (A local hold that is
+        // still ON already blocks the overwrite at the WORM gate; the
+        // LWW-reachable divergence is a stale inbound ON resurrecting a hold
+        // that was released more recently on this site.)
+        let mut local = HashMap::new();
+        local.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "OFF".to_string());
+        insert_str(&mut local, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, T_LOCAL.to_string());
+        put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
+
+        let mut inbound = HashMap::new();
+        inbound.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "ON".to_string());
+        insert_str(&mut inbound, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, T_OLD.to_string());
+        let opts = ObjectOptions {
+            replication_request: true,
+            replication_legalhold_timestamp: Some(parse_ts(T_OLD)),
+            ..versioned_opts(&version_id, inbound)
+        };
+        put_version(&set_disks, bucket, object, &version_id, &opts).await;
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(
+            info.user_defined.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER).map(String::as_str),
+            Some("OFF"),
+            "a stale inbound legal hold must not resurrect a hold released more recently"
+        );
+        assert_eq!(
+            get_str(&info.user_defined, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP).as_deref(),
+            Some(T_LOCAL)
+        );
+    }
+
+    /// Dual-key invariant under LWW: a MinIO-written destination version may
+    /// carry only the x-minio-internal timestamp key; when the local category
+    /// wins, the restored map must still hold BOTH compatibility keys.
+    #[test]
+    fn local_win_restores_both_internal_timestamp_keys_for_minio_only_metadata() {
+        let mut inbound = HashMap::new();
+        inbound.insert(AMZ_OBJECT_TAGGING.to_string(), "site=remote".to_string());
+        insert_str(&mut inbound, SUFFIX_TAGGING_TIMESTAMP, T_OLD.to_string());
+        let existing = HashMap::from([
+            (AMZ_OBJECT_TAGGING.to_string(), "site=local".to_string()),
+            ("X-Minio-Internal-Tagging-Timestamp".to_string(), T_LOCAL.to_string()),
+        ]);
+        let opts = ObjectOptions {
+            replication_request: true,
+            replication_tagging_timestamp: Some(parse_ts(T_OLD)),
+            ..Default::default()
+        };
+
+        assert!(merge_replication_metadata_lww(&mut inbound, &existing, &opts));
+        assert_eq!(inbound.get(AMZ_OBJECT_TAGGING).map(String::as_str), Some("site=local"));
+        assert_eq!(
+            inbound.get("x-rustfs-internal-tagging-timestamp").map(String::as_str),
+            Some(T_LOCAL),
+            "the RustFS twin key must be materialized even when the source version only had the MinIO key"
+        );
+        assert_eq!(inbound.get("x-minio-internal-tagging-timestamp").map(String::as_str), Some(T_LOCAL));
+    }
+
+    /// When the inbound category wins, the stored timestamp must be the
+    /// source-authored one: the PUT path's eval_metadata stamps the
+    /// object-lock timestamps with the receiver's clock
+    /// (`parse_object_lock_retention`), which would otherwise make this
+    /// replica's clock the LWW authority and wedge later convergence.
+    #[tokio::test]
+    async fn inbound_win_pins_stored_timestamp_to_source_authored_value() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-retention-ts-pinned";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+
+        // Local cleared retention at T_OLD.
+        let mut local = HashMap::new();
+        insert_str(&mut local, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_OLD.to_string());
+        put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
+
+        // Inbound newer retention: the source authored T_LOCAL, but the PUT
+        // path's eval_metadata stomped the metadata key with receiver-now
+        // (simulated by T_NEW here).
+        let mut inbound = HashMap::new();
+        inbound.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), "GOVERNANCE".to_string());
+        inbound.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), "2028-01-01T00:00:00Z".to_string());
+        insert_str(&mut inbound, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, T_NEW.to_string());
+        let opts = ObjectOptions {
+            replication_request: true,
+            replication_retention_timestamp: Some(parse_ts(T_LOCAL)),
+            ..versioned_opts(&version_id, inbound)
+        };
+        put_version(&set_disks, bucket, object, &version_id, &opts).await;
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert_eq!(
+            get_str(&info.user_defined, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP).as_deref(),
+            Some(T_LOCAL),
+            "the stored category timestamp must be the source-authored time, not the receiver's clock"
+        );
+        assert_eq!(info.user_defined.get(AMZ_OBJECT_LOCK_MODE_LOWER).map(String::as_str), Some("GOVERNANCE"));
+    }
+
+    #[tokio::test]
+    async fn newer_local_tag_deletion_survives_stale_inbound_tags() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "lww-tagging-deleted";
+        let object = "object";
+        let version_id = Uuid::new_v4().to_string();
+        make_bucket(&disk_stores, bucket).await;
+        // Local DeleteObjectTagging state: no tags, but a newer tagging timestamp.
+        let mut local = HashMap::new();
+        insert_str(&mut local, SUFFIX_TAGGING_TIMESTAMP, T_LOCAL.to_string());
+        put_version(&set_disks, bucket, object, &version_id, &versioned_opts(&version_id, local)).await;
+
+        put_version(
+            &set_disks,
+            bucket,
+            object,
+            &version_id,
+            &inbound_tagging_opts(&version_id, "site=remote", T_OLD),
+        )
+        .await;
+
+        let info = version_info(&set_disks, bucket, object, &version_id).await;
+        assert!(
+            info.user_tags.is_empty(),
+            "a newer local tag deletion must not be resurrected by older inbound tags"
+        );
+        assert_eq!(get_str(&info.user_defined, SUFFIX_TAGGING_TIMESTAMP).as_deref(), Some(T_LOCAL));
+    }
+}
+
 #[cfg(test)]
 mod inline_put_commit_path_tests {
     use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;

crates/ecstore/src/store/mod.rs

@@ -343,6 +343,23 @@ impl ECStore {
         let (decommission, rebalance) = tokio::join!(self.is_decommission_running(), self.is_rebalance_started());
         decommission || rebalance
     }
+
+    /// Returns whether scanner metadata may still be hidden by a local
+    /// data-movement state. Terminal failed/canceled decommission entries
+    /// remain suspended until an operator clears or retries them, so they are
+    /// a publication barrier even after the worker has stopped.
+    pub async fn scanner_data_usage_publication_blocked(&self) -> bool {
+        if self.scanner_data_movement_active().await {
+            return true;
+        }
+
+        let pool_meta = self.pool_meta.read().await;
+        pool_meta.pools.iter().any(|pool| {
+            pool.decommission
+                .as_ref()
+                .is_some_and(|info| !info.queued && (info.failed || info.canceled))
+        })
+    }
 }
 
 // impl Clone for ECStore {
@@ -875,6 +892,7 @@ impl crate::storage_api_contracts::admin::StorageAdminApi for ECStore {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
     use crate::layout::endpoints::{Endpoints, PoolEndpoints, SetupType};
     use crate::runtime::global::reset_local_disk_test_state;
     use crate::runtime::sources::{clear_local_disk_id_map_for_test, local_disk_path_by_id};
@@ -911,6 +929,72 @@ mod tests {
         })
     }
 
+    #[tokio::test]
+    async fn scanner_data_usage_publication_blocks_active_and_unqueued_terminal_decommission() {
+        let store = build_store_with_ctx(Arc::new(InstanceContext::new()));
+        let cases = [
+            (
+                "active",
+                PoolDecommissionInfo {
+                    start_time: Some(OffsetDateTime::now_utc()),
+                    ..Default::default()
+                },
+                true,
+            ),
+            (
+                "failed",
+                PoolDecommissionInfo {
+                    failed: true,
+                    ..Default::default()
+                },
+                true,
+            ),
+            (
+                "canceled",
+                PoolDecommissionInfo {
+                    canceled: true,
+                    ..Default::default()
+                },
+                true,
+            ),
+            (
+                "queued_failed",
+                PoolDecommissionInfo {
+                    failed: true,
+                    queued: true,
+                    ..Default::default()
+                },
+                false,
+            ),
+            (
+                "complete",
+                PoolDecommissionInfo {
+                    complete: true,
+                    ..Default::default()
+                },
+                false,
+            ),
+            ("idle", PoolDecommissionInfo::default(), false),
+        ];
+
+        for (name, decommission, expected) in cases {
+            *store.pool_meta.write().await = PoolMeta {
+                pools: vec![PoolStatus {
+                    id: 0,
+                    cmd_line: format!("scanner-publication-{name}"),
+                    last_update: OffsetDateTime::now_utc(),
+                    decommission: Some(decommission),
+                }],
+                ..Default::default()
+            };
+            assert_eq!(
+                store.scanner_data_usage_publication_blocked().await,
+                expected,
+                "unexpected scanner publication barrier state for {name}"
+            );
+        }
+    }
+
     // The object graph is the isolation carrier: two ECStore instances holding
     // distinct contexts report independent erasure state through their real
     // `&self` accessors — no cross-contamination.

crates/iam/tests/ecstore_test_compat/mod.rs

@@ -17,11 +17,11 @@
 //! All direct `rustfs_ecstore` facade imports used by tests in this crate
 //! must go through this module (architecture migration rule:
 //! `check_architecture_migration_rules.sh`). Keep the surface minimal —
-//! only what the tests actually need to build a temp-disk ECStore fixture
-//! and to flip the erasure setup type for lock-quorum fault injection.
+//! only what the tests actually need to run storage-backed IAM scenarios.
 
 #[allow(unused_imports)]
 pub(crate) mod fixture {
+    pub(crate) use rustfs_ecstore::api::bucket::migration::try_migrate_iam_config;
     pub(crate) use rustfs_ecstore::api::layout::SetupType;
 
     // `update_erasure_type` is a write-side global facade entry. Its use is

crates/iam/tests/minio_iam_migration_test.rs

@@ -0,0 +1,182 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+mod ecstore_test_compat;
+
+use ecstore_test_compat::fixture::try_migrate_iam_config;
+use rustfs_credentials::{get_global_action_cred, init_global_action_credentials};
+use rustfs_iam::store::object::{
+    IAM_CONFIG_POLICY_DB_SERVICE_ACCOUNTS_PREFIX, IAM_CONFIG_POLICY_DB_USERS_PREFIX, IAM_CONFIG_SERVICE_ACCOUNTS_PREFIX,
+    IAM_CONFIG_USERS_PREFIX, ObjectStore,
+};
+use rustfs_iam::store::{Store, UserType};
+use rustfs_iam::utils::generate_jwt;
+use rustfs_policy::auth::UserIdentity;
+use serde_json::{Value, json};
+use std::collections::HashMap;
+
+const LEGACY_META_BUCKET: &str = ".minio.sys";
+const REGULAR_USER: &str = "minio-user";
+const SERVICE_ACCOUNT: &str = "minio-service-account";
+
+async fn seed_legacy_iam_object(env: &rustfs_test_utils::TestECStoreEnv, path: &str, value: &Value) {
+    env.put_object_bytes(
+        LEGACY_META_BUCKET,
+        path,
+        serde_json::to_vec(value).expect("legacy IAM object must serialize"),
+    )
+    .await;
+}
+
+fn assert_identity_fields(actual: &UserIdentity, expected: &Value) {
+    assert_eq!(
+        serde_json::to_value(actual).expect("loaded identity must serialize"),
+        *expected,
+        "migration must preserve every credential field except expiration",
+    );
+}
+
+async fn assert_identity_survives(
+    store: &ObjectStore,
+    identity_path: &str,
+    name: &str,
+    user_type: UserType,
+    source: &Value,
+    expected_policy: &Value,
+) {
+    let mut expected = source.clone();
+    expected["credentials"]["expiration"] = Value::Null;
+
+    let persisted: UserIdentity = store
+        .load_iam_config(identity_path)
+        .await
+        .expect("migrated identity must be persisted");
+    assert_identity_fields(&persisted, &expected);
+
+    for _ in 0..2 {
+        let actual = store
+            .load_user_identity(name, user_type)
+            .await
+            .expect("migrated permanent identity must remain loadable");
+        assert_identity_fields(&actual, &expected);
+    }
+
+    let mut mappings = HashMap::new();
+    store
+        .load_mapped_policy(name, user_type, false, &mut mappings)
+        .await
+        .expect("loading the identity must not delete its policy mapping");
+    let actual_policy = mappings.get(name).expect("migrated policy mapping must exist");
+    assert_eq!(
+        serde_json::to_value(actual_policy).expect("loaded policy mapping must serialize"),
+        *expected_policy,
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn minio_permanent_identities_survive_migration_and_repeated_iam_loads() {
+    if get_global_action_cred().is_none() {
+        init_global_action_credentials(Some("MINIOMIGRATIONROOT".to_string()), Some("minio-migration-root-secret".to_string()))
+            .expect("root credentials must initialize for JWT validation");
+    }
+
+    let temp_dir = tempfile::TempDir::with_prefix("rustfs_minio_iam_migration_").expect("temp directory must be created");
+    let env = rustfs_test_utils::TestECStoreEnv::builder()
+        .base_dir(temp_dir.path())
+        .init_bucket_metadata(false)
+        .build()
+        .await;
+    for disk_path in &env.disk_paths {
+        tokio::fs::create_dir_all(disk_path.join(LEGACY_META_BUCKET))
+            .await
+            .expect("legacy metadata volume must be created");
+    }
+
+    let regular_source = json!({
+        "version": 1,
+        "credentials": {
+            "accessKey": REGULAR_USER,
+            "secretKey": "regular-user-secret",
+            "sessionToken": "",
+            "expiration": "0001-01-01T00:00:00Z",
+            "status": "on",
+            "parentUser": "regular-parent",
+            "groups": ["engineering", "operations"],
+            "claims": {"tenant": "alpha"},
+            "name": "MinIO regular user",
+            "description": "migrated regular identity"
+        },
+        "updatedAt": "2025-03-07T12:00:00Z"
+    });
+    let service_claims = json!({"sa-policy": "inherited-policy", "tenant": "alpha"});
+    let service_secret = "service-account-secret";
+    let service_source = json!({
+        "version": 1,
+        "credentials": {
+            "accessKey": SERVICE_ACCOUNT,
+            "secretKey": service_secret,
+            "sessionToken": generate_jwt(&service_claims, service_secret).expect("service-account JWT must be generated"),
+            "expiration": "1970-01-01T00:00:00Z",
+            "status": "on",
+            "parentUser": REGULAR_USER,
+            "groups": ["service-accounts"],
+            "claims": service_claims,
+            "name": "MinIO service account",
+            "description": "migrated service identity"
+        },
+        "updatedAt": "2025-03-07T12:00:00Z"
+    });
+    let regular_policy_source = json!({"version": 1, "policy": "readwrite", "updatedAt": "2025-03-07T12:00:00Z"});
+    let service_policy_source = json!({"version": 1, "policy": "readonly", "updatedAt": "2025-03-07T12:00:00Z"});
+
+    let regular_identity_path = format!("{}{REGULAR_USER}/identity.json", IAM_CONFIG_USERS_PREFIX.as_str());
+    let service_identity_path = format!("{}{SERVICE_ACCOUNT}/identity.json", IAM_CONFIG_SERVICE_ACCOUNTS_PREFIX.as_str());
+
+    seed_legacy_iam_object(&env, &regular_identity_path, &regular_source).await;
+    seed_legacy_iam_object(&env, &service_identity_path, &service_source).await;
+    seed_legacy_iam_object(
+        &env,
+        &format!("{}{REGULAR_USER}.json", IAM_CONFIG_POLICY_DB_USERS_PREFIX.as_str()),
+        &regular_policy_source,
+    )
+    .await;
+    seed_legacy_iam_object(
+        &env,
+        &format!("{}{SERVICE_ACCOUNT}.json", IAM_CONFIG_POLICY_DB_SERVICE_ACCOUNTS_PREFIX.as_str()),
+        &service_policy_source,
+    )
+    .await;
+
+    try_migrate_iam_config(env.ecstore.clone(), None).await;
+
+    let store = ObjectStore::new(env.ecstore);
+    assert_identity_survives(
+        &store,
+        &regular_identity_path,
+        REGULAR_USER,
+        UserType::Reg,
+        &regular_source,
+        &regular_policy_source,
+    )
+    .await;
+    assert_identity_survives(
+        &store,
+        &service_identity_path,
+        SERVICE_ACCOUNT,
+        UserType::Svc,
+        &service_source,
+        &service_policy_source,
+    )
+    .await;
+}

crates/scanner/src/scanner.rs

@@ -1081,18 +1081,6 @@ async fn run_data_scanner_cycle(
         }
     };
     let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
-    let storeapi_clone = storeapi.clone();
-    let ctx_clone = ctx.clone();
-    let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move {
-        store_data_usage_in_backend_with_outcome_for_epoch_and_baseline(
-            ctx_clone,
-            storeapi_clone,
-            receiver,
-            Some(leader_epoch),
-            Some(usage_persist_baseline),
-        )
-        .await
-    }));
 
     let done_cycle = Metrics::time(Metric::ScanCycle);
     let cycle_budget = ScannerCycleBudget::new(ctx, cycle_budget_config);
@@ -1107,47 +1095,78 @@ async fn run_data_scanner_cycle(
             scan_mode,
         )
         .await;
+    let publication_defer_reason = match &scan_result {
+        Ok(result) => final_data_usage_publication_defer_reason(storeapi.as_ref(), result.status).await,
+        Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
+    };
     let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled();
-    let usage_persist_outcome = match wait_for_data_usage_persist_task(ctx, &mut usage_persist_task, usage_persist_timeout).await
-    {
-        DataUsagePersistTaskResult::Completed(outcome) => outcome,
-        DataUsagePersistTaskResult::JoinFailed(err) => {
-            error!(
-                target: "rustfs::scanner",
-                event = EVENT_SCANNER_PERSIST_STATE,
-                component = LOG_COMPONENT_SCANNER,
-                subsystem = LOG_SUBSYSTEM_RUNTIME,
-                cycle = cycle_info.current,
-                state = "usage_persist_task_failed",
-                error = %err,
-                "Scanner data usage persistence task failed"
-            );
-            DataUsagePersistOutcome::Failed
+    let usage_persist_outcome = match publication_defer_reason {
+        Some(reason) => {
+            drop(receiver);
+            DataUsagePersistOutcome::Deferred(reason)
         }
-        DataUsagePersistTaskResult::Cancelled => {
-            debug!(
-                target: "rustfs::scanner",
-                event = EVENT_SCANNER_PERSIST_STATE,
-                component = LOG_COMPONENT_SCANNER,
-                subsystem = LOG_SUBSYSTEM_RUNTIME,
-                cycle = cycle_info.current,
-                state = "usage_persist_task_cancelled",
-                "Scanner data usage persistence task cancelled"
-            );
-            DataUsagePersistOutcome::Failed
-        }
-        DataUsagePersistTaskResult::TimedOut => {
-            error!(
-                target: "rustfs::scanner",
-                event = EVENT_SCANNER_PERSIST_STATE,
-                component = LOG_COMPONENT_SCANNER,
-                subsystem = LOG_SUBSYSTEM_RUNTIME,
-                cycle = cycle_info.current,
-                timeout = ?usage_persist_timeout,
-                state = "usage_persist_task_timed_out",
-                "Scanner data usage persistence task timed out"
-            );
-            DataUsagePersistOutcome::Failed
+        None => {
+            // ScannerIO emits its complete or observational update only after
+            // all set workers finish. Persist after the final activity fence;
+            // this also avoids blocking the scanner on a denied publication.
+            let storeapi_clone = storeapi.clone();
+            let ctx_clone = ctx.clone();
+            let route_probe_store = storeapi.clone();
+            let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move {
+                store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe(
+                    ctx_clone,
+                    storeapi_clone,
+                    receiver,
+                    Some(leader_epoch),
+                    Some(usage_persist_baseline),
+                    move || {
+                        let storeapi = route_probe_store.clone();
+                        async move { storeapi.scanner_data_usage_publication_blocked().await }
+                    },
+                )
+                .await
+            }));
+            match wait_for_data_usage_persist_task(ctx, &mut usage_persist_task, usage_persist_timeout).await {
+                DataUsagePersistTaskResult::Completed(outcome) => outcome,
+                DataUsagePersistTaskResult::JoinFailed(err) => {
+                    error!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_PERSIST_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        cycle = cycle_info.current,
+                        state = "usage_persist_task_failed",
+                        error = %err,
+                        "Scanner data usage persistence task failed"
+                    );
+                    DataUsagePersistOutcome::Failed
+                }
+                DataUsagePersistTaskResult::Cancelled => {
+                    debug!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_PERSIST_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        cycle = cycle_info.current,
+                        state = "usage_persist_task_cancelled",
+                        "Scanner data usage persistence task cancelled"
+                    );
+                    DataUsagePersistOutcome::Failed
+                }
+                DataUsagePersistTaskResult::TimedOut => {
+                    error!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_PERSIST_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        cycle = cycle_info.current,
+                        timeout = ?usage_persist_timeout,
+                        state = "usage_persist_task_timed_out",
+                        "Scanner data usage persistence task timed out"
+                    );
+                    DataUsagePersistOutcome::Failed
+                }
+            }
         }
     };
     let unresolved_heal_work = global_metrics().current_scan_cycle_has_unresolved_heal_work();
@@ -1191,33 +1210,51 @@ async fn run_data_scanner_cycle(
         mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
         return ScannerCycleOutcome::Failed;
     }
-    if let Some(required_cycle) = scan_cycle_result.required_cycle_floor() {
-        warn!(
-            target: "rustfs::scanner",
-            event = EVENT_SCANNER_CYCLE_STATE,
-            component = LOG_COMPONENT_SCANNER,
-            subsystem = LOG_SUBSYSTEM_RUNTIME,
-            cycle = cycle_info.current,
-            required_cycle,
-            state = "cache_cycle_ahead",
-            "Scanner cycle is recovering to a newer durable cache generation"
-        );
-        emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None);
-        return if persist_required_scanner_cycle_floor(
-            ctx,
-            storeapi.clone(),
-            cycle_info,
-            cycle_revision,
-            leader_epoch,
-            required_cycle,
-            &mut cycle_metrics_guard,
-        )
-        .await
-        {
-            ScannerCycleOutcome::Partial
-        } else {
-            ScannerCycleOutcome::Failed
-        };
+    match scanner_cycle_pre_commit_outcome(scan_cycle_result.required_cycle_floor(), &usage_persist_outcome) {
+        Some(ScannerCyclePreCommitOutcome::RecoverCacheCycle(required_cycle)) => {
+            warn!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_CYCLE_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                cycle = cycle_info.current,
+                required_cycle,
+                state = "cache_cycle_ahead",
+                "Scanner cycle is recovering to a newer durable cache generation"
+            );
+            emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None);
+            return if persist_required_scanner_cycle_floor(
+                ctx,
+                storeapi.clone(),
+                cycle_info,
+                cycle_revision,
+                leader_epoch,
+                required_cycle,
+                &mut cycle_metrics_guard,
+            )
+            .await
+            {
+                ScannerCycleOutcome::Partial
+            } else {
+                ScannerCycleOutcome::Failed
+            };
+        }
+        Some(ScannerCyclePreCommitOutcome::Deferred(reason)) => {
+            info!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_CYCLE_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                cycle = cycle_info.current,
+                reason = reason.as_str(),
+                state = "deferred",
+                "Scanner cycle deferred before data usage publication"
+            );
+            emit_scan_cycle_deferred(cycle_start.elapsed());
+            mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
+            return ScannerCycleOutcome::Deferred(reason);
+        }
+        None => {}
     }
     if usage_persist_outcome == DataUsagePersistOutcome::Failed {
         error!(
@@ -2000,6 +2037,56 @@ impl Drop for ScannerScanModeGuard {
     }
 }
 
+async fn final_data_usage_publication_defer_reason(
+    storeapi: &ECStore,
+    status: ScannerCycleStatus,
+) -> Option<ScannerCycleDeferReason> {
+    match status {
+        ScannerCycleStatus::Complete | ScannerCycleStatus::Superseded => {
+            if storeapi.scanner_data_usage_publication_blocked().await {
+                return Some(ScannerCycleDeferReason::DataMovement);
+            }
+            if status == ScannerCycleStatus::Complete {
+                let distributed = storeapi.setup_is_dist_erasure().await;
+                match probe_scanner_activity(storeapi, distributed).await {
+                    Ok(snapshot) if scanner_activity_allows_usage_publication(&snapshot) => None,
+                    Ok(_) => Some(ScannerCycleDeferReason::DataMovement),
+                    Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
+                }
+            } else {
+                // A superseded cycle is explicitly observational and cannot
+                // replace the authoritative snapshot. It may still be
+                // persisted as a convergence baseline for the next cycle.
+                None
+            }
+        }
+        ScannerCycleStatus::Deferred(reason) => Some(reason),
+        // Incomplete cycles do not publish a usage snapshot. Keep the
+        // decision permissive so existing partial-cycle handling remains
+        // unchanged if a future scanner path emits a bookkeeping update.
+        ScannerCycleStatus::Incomplete => None,
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum ScannerCyclePreCommitOutcome {
+    RecoverCacheCycle(u64),
+    Deferred(ScannerCycleDeferReason),
+}
+
+fn scanner_cycle_pre_commit_outcome(
+    required_cycle_floor: Option<u64>,
+    usage_persist_outcome: &DataUsagePersistOutcome,
+) -> Option<ScannerCyclePreCommitOutcome> {
+    // Keep the publication barrier fail-closed: `.bloomcycle.bin` uses the
+    // same routed writer and its floor must remain pending while data movement
+    // hides the source pool.
+    match usage_persist_outcome {
+        DataUsagePersistOutcome::Deferred(reason) => Some(ScannerCyclePreCommitOutcome::Deferred(*reason)),
+        _ => required_cycle_floor.map(ScannerCyclePreCommitOutcome::RecoverCacheCycle),
+    }
+}
+
 fn scanner_cycle_completion_outcome(
     scan_status: ScannerCycleStatus,
     usage_persist_outcome: DataUsagePersistOutcome,
@@ -2007,6 +2094,7 @@ fn scanner_cycle_completion_outcome(
     has_failed_dirty_usage: bool,
 ) -> ScannerCycleOutcome {
     match (scan_status, usage_persist_outcome) {
+        (_, DataUsagePersistOutcome::Deferred(reason)) => ScannerCycleOutcome::Deferred(reason),
         (_, DataUsagePersistOutcome::Failed) => ScannerCycleOutcome::Failed,
         (ScannerCycleStatus::Deferred(reason), DataUsagePersistOutcome::NoUpdate)
             if !has_dirty_usage && !has_failed_dirty_usage =>

crates/scanner/src/scanner/tests.rs

@@ -153,6 +153,7 @@ struct MemoryConfigStore {
     objects: Mutex<HashMap<String, Vec<u8>>>,
     revisions: Mutex<HashMap<String, u64>>,
     fail_put_number: Mutex<HashMap<String, usize>>,
+    object_not_found_put_number: Mutex<HashMap<String, usize>>,
     error_after_commit_put_number: Mutex<HashMap<String, usize>>,
     interleaving_puts: Mutex<HashMap<String, (usize, Vec<u8>)>>,
     cancel_after_interleaving_puts: Mutex<HashMap<String, CancellationToken>>,
@@ -224,6 +225,9 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore {
         if self.fail_put_number.lock().await.get(&key) == Some(&put_count) {
             return Err(EcstoreError::other("injected put failure"));
         }
+        if self.object_not_found_put_number.lock().await.get(&key) == Some(&put_count) {
+            return Err(EcstoreError::ObjectNotFound(bucket.to_string(), object.to_string()));
+        }
 
         let interleaving_data = {
             let mut interleaving_puts = self.interleaving_puts.lock().await;
@@ -1431,6 +1435,170 @@ async fn test_store_data_usage_in_backend_preserves_newer_snapshot() {
     assert_eq!(outcome, DataUsagePersistOutcome::Current);
 }
 
+#[tokio::test]
+async fn test_usage_save_object_not_found_defers_only_with_a_fresh_route_barrier() {
+    for (route_blocked, expected) in [
+        (true, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement)),
+        (false, DataUsagePersistOutcome::Failed),
+    ] {
+        let store = Arc::new(MemoryConfigStore::default());
+        let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
+        let baseline = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(10)), 1);
+        let baseline_data = serde_json::to_vec(&baseline).expect("baseline usage snapshot should encode");
+        store.objects.lock().await.insert(key.clone(), baseline_data.clone());
+        store.revisions.lock().await.insert(key.clone(), 1);
+        store.object_not_found_put_number.lock().await.insert(key.clone(), 1);
+
+        let (sender, receiver) = mpsc::channel(1);
+        sender
+            .send(complete_usage_with_bucket_count(
+                Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)),
+                2,
+            ))
+            .await
+            .expect("new usage snapshot should enqueue");
+        drop(sender);
+        let probe_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let route_probe_calls = probe_calls.clone();
+
+        let outcome = store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe(
+            CancellationToken::new(),
+            store.clone(),
+            receiver,
+            None,
+            Some(DataUsagePersistBaseline {
+                data: Some(Bytes::from(baseline_data.clone())),
+                revision: DataUsageCacheRevision::Etag("memory-1".to_string()),
+            }),
+            move || {
+                let probe_calls = route_probe_calls.clone();
+                async move {
+                    let call = probe_calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                    route_blocked && call > 1
+                }
+            },
+        )
+        .await;
+
+        assert_eq!(outcome, expected);
+        assert_eq!(
+            probe_calls.load(std::sync::atomic::Ordering::SeqCst),
+            3,
+            "ObjectNotFound must be followed by a fresh route-barrier probe"
+        );
+        assert_eq!(
+            store.objects.lock().await.get(&key),
+            Some(&baseline_data),
+            "a route failure must not replace the authoritative baseline"
+        );
+    }
+}
+
+#[tokio::test]
+async fn test_usage_save_route_barrier_prevents_missing_snapshot_creation() {
+    for observational in [false, true] {
+        let store = Arc::new(MemoryConfigStore::default());
+        let target_path = if observational {
+            DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()
+        } else {
+            DATA_USAGE_OBJ_NAME_PATH.as_str()
+        };
+        let target_key = memory_config_key(RUSTFS_META_BUCKET, target_path);
+        let mut incoming = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1);
+        incoming.usage_snapshot_converged = Some(!observational);
+        let (sender, receiver) = mpsc::channel(1);
+        sender.send(incoming).await.expect("usage snapshot should enqueue");
+        drop(sender);
+
+        let outcome = store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe(
+            CancellationToken::new(),
+            store.clone(),
+            receiver,
+            None,
+            Some(DataUsagePersistBaseline {
+                data: None,
+                revision: DataUsageCacheRevision::Missing,
+            }),
+            || async { true },
+        )
+        .await;
+
+        assert_eq!(outcome, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
+        assert!(!store.objects.lock().await.contains_key(&target_key));
+        assert_eq!(
+            store.put_counts.lock().await.get(&target_key),
+            None,
+            "the final pool-state fence must run before the first PUT"
+        );
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn test_usage_route_barrier_precedes_durable_reconciliation() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
+    let snapshot = complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)), 1);
+    let snapshot_data = serde_json::to_vec(&snapshot).expect("usage snapshot should encode");
+    let (sender, receiver) = mpsc::channel(1);
+    sender.send(snapshot).await.expect("usage snapshot should enqueue");
+    drop(sender);
+
+    let outcome = store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe(
+        CancellationToken::new(),
+        store.clone(),
+        receiver,
+        None,
+        Some(DataUsagePersistBaseline {
+            data: Some(Bytes::from(snapshot_data)),
+            revision: DataUsageCacheRevision::Etag("memory-1".to_string()),
+        }),
+        || async { true },
+    )
+    .await;
+
+    assert_eq!(outcome, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
+    assert_eq!(store.put_counts.lock().await.get(&key), None);
+}
+
+#[tokio::test]
+#[serial]
+async fn test_deferred_usage_save_keeps_last_real_save_metric() {
+    let metrics = global_metrics();
+    metrics.record_scanner_usage_save_result(ScannerUsageSaveResult::Success);
+    let before = metrics.report().await.usage_freshness;
+
+    let store = Arc::new(MemoryConfigStore::default());
+    let (sender, receiver) = mpsc::channel(1);
+    sender
+        .send(complete_usage_with_bucket_count(
+            Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(20)),
+            1,
+        ))
+        .await
+        .expect("usage snapshot should enqueue");
+    drop(sender);
+
+    let outcome = store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe(
+        CancellationToken::new(),
+        store,
+        receiver,
+        None,
+        Some(DataUsagePersistBaseline {
+            data: None,
+            revision: DataUsageCacheRevision::Missing,
+        }),
+        || async { true },
+    )
+    .await;
+
+    assert_eq!(outcome, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
+    let after = metrics.report().await.usage_freshness;
+    assert_eq!(after.last_usage_save_result, before.last_usage_save_result);
+    assert_eq!(after.last_usage_save_result_code, before.last_usage_save_result_code);
+    assert_eq!(after.last_usage_save_unix_secs, before.last_usage_save_unix_secs);
+}
+
 #[tokio::test]
 async fn test_store_data_usage_in_backend_fences_interleaving_newer_writer() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -2325,6 +2493,15 @@ async fn test_store_data_usage_in_backend_reports_missing_snapshot() {
 
 #[test]
 fn test_scanner_cycle_completion_prioritizes_persist_failure() {
+    assert_eq!(
+        scanner_cycle_completion_outcome(
+            ScannerCycleStatus::Complete,
+            DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement),
+            true,
+            false,
+        ),
+        ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement)
+    );
     assert_eq!(
         scanner_cycle_completion_outcome(
             ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable),
@@ -2421,6 +2598,33 @@ fn test_scanner_cycle_completion_prioritizes_persist_failure() {
     );
 }
 
+#[test]
+fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
+    for reason in [
+        ScannerCycleDeferReason::DataMovement,
+        ScannerCycleDeferReason::ActivityBaselineUnavailable,
+    ] {
+        let deferred = DataUsagePersistOutcome::Deferred(reason);
+        assert_eq!(
+            scanner_cycle_pre_commit_outcome(Some(19), &deferred),
+            Some(ScannerCyclePreCommitOutcome::Deferred(reason)),
+            "a blocked publication must not persist the routed scanner cycle floor"
+        );
+        assert_eq!(
+            scanner_cycle_pre_commit_outcome(None, &deferred),
+            Some(ScannerCyclePreCommitOutcome::Deferred(reason))
+        );
+    }
+    assert_eq!(
+        scanner_cycle_pre_commit_outcome(Some(19), &DataUsagePersistOutcome::Saved),
+        Some(ScannerCyclePreCommitOutcome::RecoverCacheCycle(19))
+    );
+    assert_eq!(
+        scanner_cycle_pre_commit_outcome(Some(19), &DataUsagePersistOutcome::Failed),
+        Some(ScannerCyclePreCommitOutcome::RecoverCacheCycle(19))
+    );
+}
+
 #[test]
 #[serial]
 fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
@@ -2448,6 +2652,23 @@ fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
     assert!(!crate::scanner_io::dirty_usage_buckets_pending());
 }
 
+#[test]
+#[serial]
+fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() {
+    crate::scanner_io::clear_dirty_usage_bucket("photos");
+    crate::scanner_io::record_dirty_usage_bucket("photos");
+    let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests();
+    let deferred = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Complete, Some(dirty_snapshot));
+
+    let (outcome, _, acknowledgements) =
+        finalize_scanner_cycle_result(deferred, DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
+
+    assert_eq!(outcome, ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
+    assert!(acknowledgements.is_empty());
+    assert!(crate::scanner_io::dirty_usage_buckets_pending());
+    crate::scanner_io::clear_dirty_usage_bucket("photos");
+}
+
 #[tokio::test]
 async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
     let pending = remote_dirty_usage_acknowledgement_pending(7, 1, std::future::ready(Ok::<bool, std::io::Error>(true))).await;

crates/scanner/src/scanner/usage_store.rs

@@ -22,6 +22,10 @@ pub(super) enum DataUsagePersistOutcome {
     AlreadyDurable,
     PriorCycleDurable,
     Saved,
+    /// The metadata route is temporarily unavailable (for example while a
+    /// terminal decommission state keeps the source pool suspended).  The
+    /// caller must retry without acknowledging dirty usage.
+    Deferred(ScannerCycleDeferReason),
     Failed,
 }
 
@@ -92,10 +96,33 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch(
 pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline(
     ctx: CancellationToken,
     storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
-    mut receiver: mpsc::Receiver<DataUsageInfo>,
+    receiver: mpsc::Receiver<DataUsageInfo>,
     leader_epoch: Option<u64>,
     initial_baseline: Option<DataUsagePersistBaseline>,
 ) -> DataUsagePersistOutcome {
+    store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe(
+        ctx,
+        storeapi,
+        receiver,
+        leader_epoch,
+        initial_baseline,
+        || async { false },
+    )
+    .await
+}
+
+pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline_and_route_probe<F, Fut>(
+    ctx: CancellationToken,
+    storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
+    mut receiver: mpsc::Receiver<DataUsageInfo>,
+    leader_epoch: Option<u64>,
+    initial_baseline: Option<DataUsagePersistBaseline>,
+    route_probe: F,
+) -> DataUsagePersistOutcome
+where
+    F: Fn() -> Fut + Send + Sync,
+    Fut: Future<Output = bool> + Send,
+{
     let mut outcome = DataUsagePersistOutcome::NoUpdate;
     let mut next_baseline = initial_baseline;
 
@@ -113,6 +140,19 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
         } else {
             DATA_USAGE_OBJ_NAME_PATH.as_str()
         };
+        if route_probe().await {
+            debug!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                path = %target_path,
+                state = "publication_blocked_before_reconcile",
+                "Scanner data usage publication deferred by the pool-state fence"
+            );
+            outcome = DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
+            break;
+        }
 
         if observational && data_usage_info.usage_snapshot_authoritative_baseline.is_none() {
             let authoritative_data = match next_baseline.as_ref() {
@@ -275,6 +315,18 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
             if ctx.is_cancelled() {
                 break 'updates;
             }
+            if route_probe().await {
+                debug!(
+                    target: "rustfs::scanner",
+                    event = EVENT_SCANNER_PERSIST_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_RUNTIME,
+                    path = %target_path,
+                    state = "publication_blocked_before_save",
+                    "Scanner data usage publication deferred by the final pool-state fence"
+                );
+                break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
+            }
 
             let done_save = Metrics::time(Metric::SaveUsage);
             let save_result = save_config_shared_with_preconditions(
@@ -313,6 +365,33 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
                         "Scanner data usage CAS conflict will be reconciled"
                     );
                 }
+                Err(e @ EcstoreError::ObjectNotFound(_, _)) => {
+                    let route_blocked = route_probe().await;
+                    if route_blocked {
+                        warn!(
+                            target: "rustfs::scanner",
+                            event = EVENT_SCANNER_PERSIST_STATE,
+                            component = LOG_COMPONENT_SCANNER,
+                            subsystem = LOG_SUBSYSTEM_RUNTIME,
+                            path = %target_path,
+                            state = "publication_deferred",
+                            error = %e,
+                            "Scanner data usage route is blocked by data movement; retrying later"
+                        );
+                        break DataUsagePersistOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
+                    }
+                    error!(
+                        target: "rustfs::scanner",
+                        event = EVENT_SCANNER_PERSIST_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_RUNTIME,
+                        path = %target_path,
+                        state = "save_failed",
+                        error = %e,
+                        "Scanner data usage save failed"
+                    );
+                    break DataUsagePersistOutcome::Failed;
+                }
                 Err(e) => {
                     error!(
                         target: "rustfs::scanner",
@@ -370,6 +449,13 @@ pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_basel
                 outcome = DataUsagePersistOutcome::Failed;
                 continue;
             }
+            DataUsagePersistOutcome::Deferred(reason) => {
+                // A deferred publication is an intentional retryable state, not a
+                // failed save. Keep the last real save result so admin freshness
+                // reporting does not turn a pool-recovery fence into a false error.
+                outcome = DataUsagePersistOutcome::Deferred(reason);
+                break 'updates;
+            }
             DataUsagePersistOutcome::Saved => {
                 if observational {
                     invalidate_admin_data_usage_snapshot_cache().await;

crates/scanner/src/scanner_io/io_cycle.rs

@@ -49,6 +49,25 @@ impl ScannerIOCycle for ECStore {
     ) -> Result<ScannerCycleResult> {
         let child_token = ctx.child_token();
 
+        // Check the local pool metadata before listing buckets. A failed or
+        // canceled decommission remains suspended after its worker exits, so
+        // starting a scan in that state could build a snapshot that cannot be
+        // routed to the authoritative metadata object.
+        if self.scanner_data_usage_publication_blocked().await {
+            debug!(
+                target: "rustfs::scanner::io",
+                event = EVENT_SCANNER_SET_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_IO,
+                state = "cycle_data_usage_route_blocked",
+                "Scanner cycle deferred while data usage metadata remains hidden by data movement"
+            );
+            return Ok(ScannerCycleResult::new(
+                ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement),
+                None,
+            ));
+        }
+
         let distributed = self.setup_is_dist_erasure().await;
         let activity_before = match scanner_activity_preflight(crate::scanner::probe_scanner_activity(self, distributed).await) {
             ScannerActivityPreflight::Ready(snapshot) => snapshot,

crates/scanner/src/scanner_io/tests.rs

@@ -17,7 +17,9 @@ use super::io_disk::tier_stats_template;
 use super::*;
 use crate::scanner_budget::ScannerCycleBudgetConfig;
 use crate::scanner_folder::ScannerItem;
-use crate::storage_api::owner::{EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats};
+use crate::storage_api::owner::{
+    EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats,
+};
 use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _};
 use crate::{
     DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerObjectOptions,
@@ -182,6 +184,39 @@ async fn scanner_cycle_is_deferred_while_rebalance_is_active() {
     assert!(receiver.recv().await.is_none(), "rebalance-deferred cycle must not publish usage");
 }
 
+#[tokio::test]
+#[serial]
+async fn scanner_cycle_is_deferred_while_terminal_decommission_is_blocked() {
+    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+    for decommission in [
+        EcstorePoolDecommissionInfo {
+            failed: true,
+            ..Default::default()
+        },
+        EcstorePoolDecommissionInfo {
+            canceled: true,
+            ..Default::default()
+        },
+    ] {
+        store.pool_meta.write().await.pools[0].decommission = Some(decommission);
+        assert!(store.scanner_data_usage_publication_blocked().await);
+
+        let ctx = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+        let (updates, mut receiver) = mpsc::channel(1);
+        let result = tokio::time::timeout(
+            Duration::from_secs(30),
+            ScannerIOCycle::nsscanner_with_status(store.as_ref(), ctx, budget, updates, 1, 1, HealScanMode::Normal),
+        )
+        .await
+        .expect("terminal-decommission-deferred scanner cycle should finish")
+        .expect("terminal-decommission-deferred scanner cycle should succeed");
+
+        assert_eq!(result.status, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement));
+        assert!(receiver.recv().await.is_none(), "blocked cycle must not publish usage");
+    }
+}
+
 #[tokio::test]
 async fn data_usage_publish_fails_when_receiver_is_closed() {
     let (updates, receiver) = mpsc::channel(1);
@@ -236,6 +271,10 @@ async fn multi_pool_scanner_cycle_publishes_combined_usage() {
     assert_eq!(bucket_usage.size, 11);
     assert_eq!(usage.objects_total_count, 2);
     assert_eq!(usage.objects_total_size, 11);
+    assert!(
+        receiver.recv().await.is_none(),
+        "a scanner cycle must publish at most one terminal usage snapshot"
+    );
 }
 
 #[tokio::test]

crates/scanner/src/storage_api.rs

@@ -47,6 +47,8 @@ pub(crate) use rustfs_ecstore::api::bucket::versioning_sys::BucketVersioningSys
 pub(crate) use rustfs_ecstore::api::cache::{
     ListPathRawOptions as EcstoreListPathRawOptions, list_path_raw as ecstore_list_path_raw,
 };
+#[cfg(test)]
+pub(crate) use rustfs_ecstore::api::capacity::PoolDecommissionInfo as EcstorePoolDecommissionInfo;
 pub(crate) use rustfs_ecstore::api::capacity::{
     is_reserved_or_invalid_bucket as ecstore_is_reserved_or_invalid_bucket, path2_bucket_object as ecstore_path2_bucket_object,
     path2_bucket_object_with_base_path as ecstore_path2_bucket_object_with_base_path,
@@ -127,9 +129,9 @@ pub(crate) mod owner {
     #[cfg(test)]
     pub(crate) use super::{
         EcstoreDiskOption, EcstoreDiskStore, EcstoreEndpoint, EcstoreEndpointServerPools, EcstoreEndpoints,
-        EcstoreInstanceContext, EcstorePoolEndpoints, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta,
-        EcstoreRebalanceStats, ecstore_config_init, ecstore_init_bucket_metadata_sys, ecstore_init_local_disks_with_instance_ctx,
-        ecstore_new_disk,
+        EcstoreInstanceContext, EcstorePoolDecommissionInfo, EcstorePoolEndpoints, EcstoreRebalStatus, EcstoreRebalanceInfo,
+        EcstoreRebalanceMeta, EcstoreRebalanceStats, ecstore_config_init, ecstore_init_bucket_metadata_sys,
+        ecstore_init_local_disks_with_instance_ctx, ecstore_new_disk,
     };
 }
 

rustfs/src/app/object_usecase.rs

@@ -3969,6 +3969,14 @@ fn delete_creates_delete_marker(opts: &ObjectOptions) -> bool {
     opts.version_id.is_none() && opts.versioned && !opts.version_suspended
 }
 
+/// `DeleteObjects` is idempotent. A raw filesystem `NotFound` can cross the
+/// distributed delete path instead of its usual typed missing-object error.
+fn is_delete_objects_not_found(error: &EcstoreError) -> bool {
+    is_err_object_not_found(error)
+        || is_err_version_not_found(error)
+        || matches!(error, StorageError::Io(source) if source.kind() == std::io::ErrorKind::NotFound)
+}
+
 /// Bounded concurrency for the per-object pre-delete stat fanout in
 /// `execute_delete_objects` (backlog#929 / HP-8). Keeps the metadata reads for
 /// a 1000-key batch from serializing while capping the disk fanout pressure.
@@ -4030,6 +4038,27 @@ fn delete_response_version_id(version_id: Option<Uuid>, synthetic_version_id: bo
     }
 }
 
+fn reduce_delete_objects_result<'a>(
+    object: &ObjectToDelete,
+    deleted: &'a StorageDeletedObject,
+    error: Option<&EcstoreError>,
+    synthetic_version_id: bool,
+) -> Result<&'a StorageDeletedObject, s3s::dto::Error> {
+    match error {
+        None => Ok(deleted),
+        Some(error) if is_delete_objects_not_found(error) => Ok(deleted),
+        Some(error) => {
+            let api_error = ApiError::from(error.clone());
+            Err(s3s::dto::Error {
+                code: Some(api_error.code.as_str().to_string()),
+                key: Some(object.object_name.clone()),
+                message: Some(api_error.message),
+                version_id: delete_response_version_id(object.version_id, synthetic_version_id),
+            })
+        }
+    }
+}
+
 fn resolve_put_object_extract_options(headers: &HeaderMap) -> S3Result<PutObjectExtractOptions> {
     let prefix = snowball_meta_value(headers, SNOWBALL_PREFIX_HEADER_KEYS, SNOWBALL_PREFIX_SUFFIX_LOWER)
         .map(|value| normalize_snowball_prefix(&value))
@@ -8476,39 +8505,31 @@ impl DefaultObjectUsecase {
         for (i, err) in errs.iter().enumerate() {
             let didx = object_to_delete_idx[i];
 
-            if err.is_none()
-                || err
-                    .clone()
-                    .is_some_and(|v| is_err_object_not_found(&v) || is_err_version_not_found(&v))
-            {
-                delete_results[didx].delete_object = Some(dobjs[i].clone());
-                let (versioned, version_suspended) = object_versioning[i];
-                let creates_delete_marker = object_to_delete[i].version_id.is_none() && versioned && !version_suspended;
-                if creates_delete_marker {
-                    record_bucket_delete_marker_memory(&bucket).await;
-                } else {
-                    let size = object_sizes[i].max(0) as u64;
-                    record_bucket_object_delete_memory(
-                        &bucket,
-                        size,
-                        existing_object_infos[i].is_some() && object_to_delete[i].version_id.is_none(),
-                    )
-                    .await;
+            match reduce_delete_objects_result(
+                &object_to_delete[i],
+                &dobjs[i],
+                err.as_ref(),
+                delete_results[didx].synthetic_version_id,
+            ) {
+                Ok(deleted_object) => {
+                    delete_results[didx].delete_object = Some(deleted_object.clone());
+                    let (versioned, version_suspended) = object_versioning[i];
+                    let creates_delete_marker = object_to_delete[i].version_id.is_none() && versioned && !version_suspended;
+                    if creates_delete_marker {
+                        record_bucket_delete_marker_memory(&bucket).await;
+                    } else {
+                        let size = object_sizes[i].max(0) as u64;
+                        record_bucket_object_delete_memory(
+                            &bucket,
+                            size,
+                            existing_object_infos[i].is_some() && object_to_delete[i].version_id.is_none(),
+                        )
+                        .await;
+                    }
+                }
+                Err(error) => {
+                    delete_results[didx].error = Some(error);
                 }
-                continue;
-            }
-
-            if let Some(err) = err.clone() {
-                let api_error = ApiError::from(err);
-                delete_results[didx].error = Some(s3s::dto::Error {
-                    code: Some(api_error.code.as_str().to_string()),
-                    key: Some(object_to_delete[i].object_name.clone()),
-                    message: Some(api_error.message),
-                    version_id: delete_response_version_id(
-                        object_to_delete[i].version_id,
-                        delete_results[didx].synthetic_version_id,
-                    ),
-                });
             }
         }
 
@@ -17692,6 +17713,35 @@ mod tests {
         assert_eq!(internal_version_id, None);
     }
 
+    #[test]
+    fn delete_objects_treats_raw_io_not_found_as_idempotent() {
+        assert!(is_delete_objects_not_found(&StorageError::FileNotFound));
+        assert!(is_delete_objects_not_found(&StorageError::Io(std::io::Error::from(
+            std::io::ErrorKind::NotFound,
+        ))));
+        assert!(!is_delete_objects_not_found(&StorageError::Io(std::io::Error::from(
+            std::io::ErrorKind::PermissionDenied,
+        ))));
+        assert!(!is_delete_objects_not_found(&StorageError::DiskNotFound));
+    }
+
+    #[test]
+    fn delete_objects_result_reducer_reports_raw_not_found_as_deleted() {
+        let object = ObjectToDelete {
+            object_name: "missing-key".to_string(),
+            ..Default::default()
+        };
+        let deleted = StorageDeletedObject {
+            object_name: object.object_name.clone(),
+            ..Default::default()
+        };
+        let error = StorageError::Io(std::io::Error::from(std::io::ErrorKind::NotFound));
+
+        let deleted = reduce_delete_objects_result(&object, &deleted, Some(&error), false)
+            .expect("raw not-found must produce a deleted result");
+        assert_eq!(deleted.object_name, "missing-key");
+    }
+
     #[test]
     fn recursive_force_delete_requires_administrative_or_replica_context() {
         let mut headers = HeaderMap::new();

rustfs/src/server/http.rs

@@ -1537,7 +1537,7 @@ fn process_connection(
                 None
             }
         };
-        // ── Canonical Middleware Stack Order (outermost → innermost) ──
+        // ── Canonical External Middleware Stack Order (outermost → innermost) ──
         // This order MUST be preserved across refactorings.
         // Only AddExtensionLayer (layers 1-2) are per-connection; most remaining layers are stateless.
         //
@@ -1565,6 +1565,8 @@ fn process_connection(
         // 22. PublicHealthEndpointLayer              — handles public health before s3s host parsing
         // 23. VirtualHostStyleHintLayer              — actionable error for unroutable virtual-hosted-style (conditional)
         // 24. DoubleSlashListBucketsCompatLayer      — rewrites `GET //` to `GET /` for ListBuckets (MinIO browser compat)
+        // The internode lane below intentionally keeps only the shared
+        // transport/auth/observability subset needed by `/rustfs/rpc/...`.
         // ─────────────────────────────────────────────────────────────
         let build_external_stack = |service| {
             ServiceBuilder::new()
@@ -1747,16 +1749,9 @@ fn process_connection(
                 .layer(PropagateRequestIdLayer::x_request_id())
                 .layer(CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new(compression_config.clone())))
                 .option_layer(compression_config.enabled.then_some(PathCategoryInjectionLayer))
-                .layer(S3ErrorMessageCompatLayer)
-                .layer(IcebergRestErrorCompatLayer)
-                .layer(ObjectAttributesEtagFixLayer)
-                .layer(ConditionalCorsLayer::new())
-                .option_layer(if is_console { Some(RedirectLayer) } else { None })
-                .layer(BodylessStatusFixLayer)
-                .layer(HeadRequestBodyFixLayer)
-                .layer(PublicHealthEndpointLayer::new(Arc::clone(&server_ctx)))
-                .option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer))
-                .layer(DoubleSlashListBucketsCompatLayer)
+                // The internode lane only serves `/rustfs/rpc/...` gRPC requests.
+                // Keep safety/observability layers above, but leave S3/REST
+                // compatibility rewrites on the external lane.
                 .service(service)
         };
         let external_stack_service = build_external_stack(external_service);

rustfs/src/storage/options.rs

@@ -554,10 +554,11 @@ fn apply_replication_timestamps_from_headers(headers: &HeaderMap<HeaderValue>, o
 
     // Persist into the internal metadata keys so a later outbound replication
     // pass (replication_target_boundary) reads the source's modification
-    // times instead of falling back to mod_time.
-    // TODO(P1-6): receiver-side LWW is still missing — when the stored
-    // per-category timestamp is newer than the inbound one, the existing
-    // tags/retention/legal-hold should win instead of being overwritten.
+    // times instead of falling back to mod_time. Receiver-side LWW happens at
+    // the set layer under the object write lock
+    // (ecstore set_disk::ops::object::merge_replication_metadata_lww,
+    // rustfs/backlog#1953): a category whose stored timestamp is newer than
+    // the inbound one keeps the local values.
     for (timestamp, suffix) in [
         (opts.replication_tagging_timestamp, SUFFIX_TAGGING_TIMESTAMP),
         (opts.replication_retention_timestamp, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP),