fix(storage): preserve scanner checkpoints and bound MRF journal I/O (#8692)

* fix(scanner): preserve checkpoints on identity read failures

Distinguish confirmed missing incarnation sidecars from lookup errors before
cache preparation. Preserve durable checkpoint progress when identity reads
are unavailable and report post-scan uncertainty separately from replacement.

Reuse the transaction-fenced reader for optional and required lookups while
retaining existing missing, malformed, system-bucket and lock-loss behavior.
Verify preservation and recovery through a real coordinator with the identity
sidecar write-locked, including unchanged cache revisions and dirty work.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(ecstore): share required incarnation conversion

Keep the missing-incarnation error contract in one conversion used by both
required lookup entry points. This removes a duplicate formatted error site
and satisfies the existing error-format ratchet without changing its baseline.

Pin the full missing-identity error text in the real SetDisks regression.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(heal): bound cross-disk MRF checkpoint I/O

Run independent journal disks with bounded concurrency while retaining each
replica's payload, companion and manifest ordering and the mirror phases.
Share encoded bodies, deduplicate local roots and collect every result
without changing receipt deadlines or persistence failure semantics.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/bucket/metadata_sys.rs

@@ -525,6 +525,15 @@ pub(crate) async fn get_bucket_incarnation_id_in(ctx: &crate::runtime::instance:
     sys.get_bucket_incarnation_id_from_disk(bucket).await
 }
 
+pub(crate) async fn find_bucket_incarnation_id_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+) -> Result<Option<Uuid>> {
+    let sys = bucket_metadata_sys_of(ctx)?;
+    let sys = sys.read().await.clone();
+    sys.find_bucket_incarnation_id_from_disk(bucket).await
+}
+
 /// Directory loss is recoverable only for a committed, non-retired generation.
 /// The caller retains the bucket lifecycle fence through the ensuing reconstruction.
 pub(crate) async fn validate_bucket_reconstruction_in(
@@ -2623,6 +2632,11 @@ impl BucketMetadataSys {
     }
 
     async fn get_bucket_incarnation_id_from_disk(&self, bucket: &str) -> Result<Uuid> {
+        let incarnation_id = self.find_bucket_incarnation_id_from_disk(bucket).await?;
+        Self::require_bucket_incarnation_id(bucket, incarnation_id)
+    }
+
+    async fn find_bucket_incarnation_id_from_disk(&self, bucket: &str) -> Result<Option<Uuid>> {
         let transaction_lock = self
             .object_store()
             .new_ns_lock(RUSTFS_META_BUCKET, &bucket_metadata_transaction_lock_key(bucket))
@@ -2630,7 +2644,7 @@ impl BucketMetadataSys {
         let _transaction_guard = transaction_lock
             .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
             .await?;
-        self.get_bucket_incarnation_id_under_transaction_lock(bucket, &_transaction_guard)
+        self.find_bucket_incarnation_id_under_transaction_lock(bucket, &_transaction_guard)
             .await
     }
 
@@ -2639,14 +2653,26 @@ impl BucketMetadataSys {
         bucket: &str,
         transaction_guard: &rustfs_lock::NamespaceLockGuard,
     ) -> Result<Uuid> {
+        let incarnation_id = self
+            .find_bucket_incarnation_id_under_transaction_lock(bucket, transaction_guard)
+            .await?;
+        Self::require_bucket_incarnation_id(bucket, incarnation_id)
+    }
+
+    fn require_bucket_incarnation_id(bucket: &str, incarnation_id: Option<Uuid>) -> Result<Uuid> {
+        incarnation_id.ok_or_else(|| Error::other(format!("bucket incarnation metadata is not authoritative: {bucket}")))
+    }
+
+    async fn find_bucket_incarnation_id_under_transaction_lock(
+        &self,
+        bucket: &str,
+        transaction_guard: &rustfs_lock::NamespaceLockGuard,
+    ) -> Result<Option<Uuid>> {
         let incarnation_id = load_bucket_incarnation(self.object_store(), bucket).await?;
         if transaction_guard.is_lock_lost() {
             return Err(Error::other(format!("bucket incarnation metadata transaction lock was lost: {bucket}")));
         }
-        match incarnation_id {
-            Some(incarnation_id) if !incarnation_id.is_nil() => Ok(incarnation_id),
-            _ => Err(Error::other(format!("bucket incarnation metadata is not authoritative: {bucket}"))),
-        }
+        Ok(incarnation_id)
     }
 
     async fn get_metadata_authority(&self, bucket: &str) -> Result<BucketMetadataAuthority> {

crates/ecstore/src/set_disk/mod.rs

@@ -5346,6 +5346,14 @@ impl SetDisks {
         metadata_sys::get_bucket_incarnation_id_in(&self.ctx, bucket).await
     }
 
+    /// Read a persisted identity without treating lookup failures as legacy absence.
+    pub async fn find_bucket_incarnation_id_from_disk(&self, bucket: &str) -> Result<Option<Uuid>> {
+        if crate::bucket::utils::is_meta_bucketname(bucket) {
+            return Err(Error::other("system metadata bucket has no bucket incarnation"));
+        }
+        metadata_sys::find_bucket_incarnation_id_in(&self.ctx, bucket).await
+    }
+
     /// Admit one short scanner cache publication under this set's instance
     /// movement fence. The caller must hold the returned guard through its
     /// final conditional cache write; no scan-round work belongs under it.
@@ -8013,10 +8021,27 @@ mod tests {
                 .await
                 .expect("system metadata identity lookup must not reacquire the held pool metadata lock")
                 .expect_err("system metadata buckets have no user bucket incarnation");
+            tokio::time::timeout(Duration::from_secs(30), set.find_bucket_incarnation_id_from_disk(bucket))
+                .await
+                .expect("optional identity lookup must not reacquire the held pool metadata lock")
+                .expect_err("system metadata buckets must not become a legacy identity");
         }
 
         drop(_pool_meta_guard);
         let bucket = "user-incarnation-boundary";
+        assert_eq!(
+            set.find_bucket_incarnation_id_from_disk(bucket)
+                .await
+                .expect("a confirmed absent identity is distinct from a read error"),
+            None,
+        );
+        assert_eq!(
+            set.bucket_incarnation_id_from_disk(bucket)
+                .await
+                .expect_err("required lookup retains its missing-identity error")
+                .to_string(),
+            format!("Io error: bucket incarnation metadata is not authoritative: {bucket}"),
+        );
         let incarnation = Uuid::new_v4();
         crate::bucket::metadata::save_bucket_incarnation(Arc::clone(&store), bucket, incarnation)
             .await
@@ -8027,6 +8052,28 @@ mod tests {
                 .expect("user bucket identities must still load from the metadata owner"),
             incarnation,
         );
+        assert_eq!(
+            set.find_bucket_incarnation_id_from_disk(bucket)
+                .await
+                .expect("persisted identity"),
+            Some(incarnation)
+        );
+
+        let identity_key = format!(
+            "{}/{bucket}/{}",
+            crate::disk::BUCKET_META_PREFIX,
+            crate::bucket::metadata::BUCKET_INCARNATION_FILE
+        );
+        for invalid in [Vec::new(), vec![1; 15], Uuid::nil().as_bytes().to_vec()] {
+            crate::config::com::save_config(store.clone(), &identity_key, invalid)
+                .await
+                .expect("persist an invalid identity fixture");
+            assert!(
+                set.find_bucket_incarnation_id_from_disk(bucket).await.is_err(),
+                "invalid identity bytes must not enable the legacy rebuild path"
+            );
+            assert!(set.bucket_incarnation_id_from_disk(bucket).await.is_err());
+        }
     }
 
     async fn make_test_set_disks_with_ctx(

crates/heal/src/heal/mrf_queue.rs

@@ -27,6 +27,7 @@
 
 use super::{DiskStore, HealDiskExt as _, local_disk_map_read};
 use crate::heal::manager::{HealManager, MrfRepairNoticeTarget};
+use futures::{StreamExt, stream};
 use metrics::{counter, gauge};
 use rustfs_common::mrf_channel::{
     MRF_MAX_ATTEMPTS, MRF_MAX_DELETE_MARKER_BYTES, MrfDeleteMarkerPurge, MrfDeleteMarkerPurgeIdentity, MrfDurableAdmissionError,
@@ -48,6 +49,7 @@ pub use partial_write::{MrfLegacyResponsibility, MrfOperatorAcceptance};
 use partial_write::{PartialWrites, ResponsibilityCheckpoint};
 
 const MRF_LIFECYCLE_CONTROL_CAPACITY: usize = 128;
+const MRF_JOURNAL_DISK_CONCURRENCY: usize = 4;
 const MRF_LIFECYCLE_LIST_MAX_LIMIT: usize = 256;
 
 static GLOBAL_MRF_LIFECYCLE_SENDER: std::sync::OnceLock<mpsc::Sender<MrfLifecycleCommand>> = std::sync::OnceLock::new();
@@ -712,7 +714,29 @@ fn intent_digest(intent: &MrfIntent) -> Option<[u8; 32]> {
 
 async fn journal_disks() -> Vec<DiskStore> {
     let map = local_disk_map_read().await;
-    map.values().flatten().cloned().collect()
+    distinct_journal_disks(map.values().flatten())
+}
+
+fn distinct_journal_disks<'a>(disks: impl IntoIterator<Item = &'a DiskStore>) -> Vec<DiskStore> {
+    let mut roots = HashSet::new();
+    disks
+        .into_iter()
+        .filter(|disk| {
+            !crate::heal::storage_api::owner::EcstoreDiskAPI::is_local(disk.as_ref())
+                || roots.insert(crate::heal::storage_api::owner::EcstoreDiskAPI::path(disk.as_ref()))
+        })
+        .cloned()
+        .collect()
+}
+
+// Bound disk fan-out without serializing independent fsyncs. The caller keeps
+// each disk's commit steps in one future and waits for every replica result.
+fn journal_disk_io<I, F>(operations: I) -> impl std::future::Future<Output = Vec<F::Output>>
+where
+    I: IntoIterator<Item = F>,
+    F: std::future::Future,
+{
+    stream::iter(operations).buffered(MRF_JOURNAL_DISK_CONCURRENCY).collect()
 }
 
 async fn read_journal(path: &str) -> Option<Vec<u8>> {
@@ -750,9 +774,15 @@ async fn read_mrf_lifecycle_checkpoint(
 /// the next tick retries the persist.
 async fn write_journal(path: &str, data: &[u8]) -> bool {
     let payload = bytes::Bytes::copy_from_slice(data);
+    let disks = journal_disks().await;
+    let results = journal_disk_io(disks.into_iter().map(|disk| {
+        let payload = payload.clone();
+        async move { disk.write_all(super::RUSTFS_META_BUCKET, path, payload).await }
+    }))
+    .await;
     let mut any_persisted = false;
-    for disk in journal_disks().await {
-        match disk.write_all(super::RUSTFS_META_BUCKET, path, payload.clone()).await {
+    for result in results {
+        match result {
             Ok(()) => any_persisted = true,
             Err(err) => warn_mrf_journal_write(&err),
         }
@@ -766,15 +796,17 @@ async fn delete_journal(path: &str) -> bool {
         counter!("rustfs_heal_mrf_journal_delete_failures_total").increment(1);
         return false;
     }
+    let results = journal_disk_io(disks.into_iter().map(|disk| async move {
+        disk.delete(
+            super::RUSTFS_META_BUCKET,
+            path,
+            crate::heal::storage_api::owner::EcstoreDeleteOptions::default(),
+        )
+        .await
+    }))
+    .await;
     let mut all_deleted = true;
-    for disk in disks {
-        let result = disk
-            .delete(
-                super::RUSTFS_META_BUCKET,
-                path,
-                crate::heal::storage_api::owner::EcstoreDeleteOptions::default(),
-            )
-            .await;
+    for result in results {
         if let Err(err) = result {
             // Delete is idempotent: a compatibility mirror that was never
             // written (or was already removed) is clean, not a retry state.
@@ -2024,6 +2056,60 @@ mod tests {
     use std::sync::Arc as StdArc;
     use std::time::{Duration as StdDuration, Instant};
 
+    #[tokio::test(start_paused = true)]
+    async fn journal_disk_io_bounds_fanout_and_collects_every_result() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        let active = AtomicUsize::new(0);
+        let peak = AtomicUsize::new(0);
+        let completed = AtomicUsize::new(0);
+        let results = journal_disk_io((0..9).map(|index| {
+            let (active, peak, completed) = (&active, &peak, &completed);
+            async move {
+                let running = active.fetch_add(1, Ordering::SeqCst) + 1;
+                peak.fetch_max(running, Ordering::SeqCst);
+                tokio::time::sleep(Duration::from_secs(1)).await;
+                active.fetch_sub(1, Ordering::SeqCst);
+                completed.fetch_add(1, Ordering::SeqCst);
+                if index == 0 || index == 8 { Err(index) } else { Ok(index) }
+            }
+        }))
+        .await;
+
+        assert_eq!(peak.load(Ordering::SeqCst), MRF_JOURNAL_DISK_CONCURRENCY);
+        assert_eq!(active.load(Ordering::SeqCst), 0);
+        assert_eq!(completed.load(Ordering::SeqCst), 9, "an early failure must not skip later replicas");
+        assert_eq!(results, vec![Err(0), Ok(1), Ok(2), Ok(3), Ok(4), Ok(5), Ok(6), Ok(7), Err(8)]);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn journal_disk_io_cancellation_drops_inflight_operations() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        struct InFlight<'a>(&'a AtomicUsize);
+        impl Drop for InFlight<'_> {
+            fn drop(&mut self) {
+                self.0.fetch_sub(1, Ordering::SeqCst);
+            }
+        }
+        let started = AtomicUsize::new(0);
+        let active = AtomicUsize::new(0);
+        let work = journal_disk_io((0..9).map(|_| {
+            let (started, active) = (&started, &active);
+            async move {
+                started.fetch_add(1, Ordering::SeqCst);
+                active.fetch_add(1, Ordering::SeqCst);
+                let _guard = InFlight(active);
+                std::future::pending::<()>().await;
+            }
+        }));
+        tokio::time::timeout(Duration::from_secs(1), work)
+            .await
+            .expect_err("the unfinished fan-out must remain cancellable");
+        assert_eq!(started.load(Ordering::SeqCst), MRF_JOURNAL_DISK_CONCURRENCY);
+        assert_eq!(active.load(Ordering::SeqCst), 0, "fan-out must not detach replica tasks");
+    }
+
     const W13_EVIDENCE_DIR_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_EVIDENCE_DIR";
     const W13_SOURCE_REVISION_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_SOURCE_REVISION";
     const W13_SELECTION_ENV: &str = "RUSTFS_SCANNER_HEAL_W13_SELECTION";

crates/heal/src/heal/mrf_queue/snapshot.rs

@@ -28,7 +28,7 @@
 //! One surviving committed replica supports process restart recovery only;
 //! this reader does not establish a replication quorum or a power-loss policy.
 
-use super::{MRF_JOURNAL_PATH, MRF_SCOPED_JOURNAL_PATH, decode_journal};
+use super::{MRF_JOURNAL_PATH, MRF_SCOPED_JOURNAL_PATH, decode_journal, distinct_journal_disks, journal_disk_io};
 use crate::heal::RUSTFS_META_BUCKET;
 use crate::heal::storage_api::owner::{
     EcstoreConditionalFileUpdate, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskStore,
@@ -307,7 +307,7 @@ async fn read_committed_with_stats(
 async fn cas_replace(
     disk: &EcstoreDiskStore,
     path: &str,
-    replacement: &[u8],
+    replacement: EcstoreDiskBytes,
     limit: usize,
 ) -> Result<EcstoreConditionalFileUpdate, SnapshotError> {
     let expected = read_bounded(disk, path, limit).await?.map(EcstoreDiskBytes::from);
@@ -318,17 +318,11 @@ async fn cas_replace_expected(
     disk: &EcstoreDiskStore,
     path: &str,
     expected: Option<EcstoreDiskBytes>,
-    replacement: &[u8],
+    replacement: EcstoreDiskBytes,
 ) -> Result<EcstoreConditionalFileUpdate, SnapshotError> {
-    EcstoreDiskAPI::compare_and_update_file(
-        disk.as_ref(),
-        RUSTFS_META_BUCKET,
-        path,
-        expected,
-        Some(EcstoreDiskBytes::copy_from_slice(replacement)),
-    )
-    .await
-    .map_err(SnapshotError::Disk)
+    EcstoreDiskAPI::compare_and_update_file(disk.as_ref(), RUSTFS_META_BUCKET, path, expected, Some(replacement))
+        .await
+        .map_err(SnapshotError::Disk)
 }
 
 async fn cas_delete_expected(
@@ -391,56 +385,58 @@ pub async fn publish_committed_snapshot_with_companion(
     if decode_journal(payload).1 != 0 {
         return Err(SnapshotError::Corrupt);
     }
-    let current = read_committed(disks, limit).await?;
+    // Endpoint aliases can share the same canonical local root and CAS lock.
+    let disks = distinct_journal_disks(disks.iter());
+    let current = read_committed(&disks, limit).await?;
     if current.as_ref().is_some_and(|snapshot| snapshot.sequence() >= sequence) {
         return Err(SnapshotError::Conflict);
     }
     let slot = current.as_ref().map_or(0, |snapshot| 1usize.saturating_sub(snapshot.slot()));
-    let manifest = Manifest::encode(owner, sequence, payload)?;
+    let manifest = EcstoreDiskBytes::from(Manifest::encode(owner, sequence, payload)?);
+    let payload = EcstoreDiskBytes::copy_from_slice(payload);
+    let companion = companion.map(|(paths, bytes, limit)| (paths, EcstoreDiskBytes::copy_from_slice(bytes), limit));
+    // Share encoded bodies across replicas; parallel fsync must not multiply
+    // the large payload/lifecycle allocations. CAS steps stay ordered per disk.
+    let manifest = &manifest;
+    let payload = &payload;
+    let companion = &companion;
+    let results = journal_disk_io(disks.into_iter().map(|disk| async move {
+        let mut payload_written = false;
+        let manifest_written: Result<bool, SnapshotError> = async {
+            let expected_manifest = read_bounded(&disk, MANIFEST_PATHS[slot], MANIFEST_LEN).await?;
+            validate_reusable_manifest_slot(expected_manifest.as_deref(), sequence, limit)?;
+            match cas_replace(&disk, PAYLOAD_PATHS[slot], payload.clone(), limit).await? {
+                EcstoreConditionalFileUpdate::Updated => payload_written = true,
+                EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch => return Ok(false),
+            }
+            if let Some((companion_paths, companion_bytes, companion_limit)) = companion {
+                match cas_replace(&disk, companion_paths[slot], companion_bytes.clone(), *companion_limit).await? {
+                    EcstoreConditionalFileUpdate::Updated => {}
+                    EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch => return Ok(false),
+                }
+            }
+            Ok(matches!(
+                cas_replace_expected(
+                    &disk,
+                    MANIFEST_PATHS[slot],
+                    expected_manifest.map(EcstoreDiskBytes::from),
+                    manifest.clone(),
+                )
+                .await?,
+                EcstoreConditionalFileUpdate::Updated
+            ))
+        }
+        .await;
+        (payload_written, manifest_written)
+    }))
+    .await;
     let mut payload_replicas = 0usize;
     let mut manifest_replicas = 0usize;
     let mut first_error = None;
-    for disk in disks {
-        let expected_manifest = match read_bounded(disk, MANIFEST_PATHS[slot], MANIFEST_LEN).await {
-            Ok(expected) => expected,
-            Err(error) => {
-                if first_error.is_none() {
-                    first_error = Some(error);
-                }
-                continue;
-            }
-        };
-        if let Err(error) = validate_reusable_manifest_slot(expected_manifest.as_deref(), sequence, limit) {
-            if first_error.is_none() {
-                first_error = Some(error);
-            }
-            continue;
-        }
-        match cas_replace(disk, PAYLOAD_PATHS[slot], payload, limit).await {
-            Ok(EcstoreConditionalFileUpdate::Updated) => payload_replicas += 1,
-            Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => continue,
-            Err(error) => {
-                if first_error.is_none() {
-                    first_error = Some(error);
-                }
-                continue;
-            }
-        }
-        if let Some((companion_paths, companion_bytes, companion_limit)) = companion {
-            match cas_replace(disk, companion_paths[slot], companion_bytes, companion_limit).await {
-                Ok(EcstoreConditionalFileUpdate::Updated) => {}
-                Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => continue,
-                Err(error) => {
-                    if first_error.is_none() {
-                        first_error = Some(error);
-                    }
-                    continue;
-                }
-            }
-        }
-        match cas_replace_expected(disk, MANIFEST_PATHS[slot], expected_manifest.map(EcstoreDiskBytes::from), &manifest).await {
-            Ok(EcstoreConditionalFileUpdate::Updated) => manifest_replicas += 1,
-            Ok(EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch) => {}
+    for (payload_written, result) in results {
+        payload_replicas += usize::from(payload_written);
+        match result {
+            Ok(written) => manifest_replicas += usize::from(written),
             Err(error) => {
                 if first_error.is_none() {
                     first_error = Some(error);
@@ -1450,6 +1446,62 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn committed_snapshot_writer_retains_payload_count_when_one_companion_fails() {
+        let root = TempDir::new().expect("test directory");
+        let first = disk(&root, "first").await;
+        let second = disk(&root, "second").await;
+        let paths = super::super::MRF_LIFECYCLE_PATHS;
+        std::fs::create_dir(EcstoreDiskAPI::path(first.as_ref()).join(RUSTFS_META_BUCKET).join(paths[0]))
+            .expect("block only the first replica's companion");
+        let owner = Uuid::new_v4();
+        let bytes = payload("healthy-companion-replica");
+        let disks = [first.clone(), second];
+        let publication =
+            publish_committed_snapshot_with_companion(&disks, owner, 1, &bytes, 4096, Some((&paths, b"lifecycle", 4096)))
+                .await
+                .expect("a failed companion must not stop another replica's commit");
+        assert_eq!(publication.payload_replicas, 2);
+        assert_eq!(publication.manifest_replicas, 1);
+        assert!(
+            read_bounded(&first, MANIFEST_PATHS[0], MANIFEST_LEN)
+                .await
+                .expect("read the failed replica's manifest")
+                .is_none(),
+            "a failed companion must not publish its commit manifest"
+        );
+        assert_eq!(
+            read_committed_companion(&disks, owner, 1, &paths, 4096, 4096)
+                .await
+                .expect("read the healthy committed companion")
+                .as_deref(),
+            Some(b"lifecycle".as_slice())
+        );
+    }
+
+    #[tokio::test]
+    async fn committed_snapshot_writer_deduplicates_disk_roots_before_parallel_cas() {
+        let root = TempDir::new().expect("test directory");
+        let first = disk(&root, "first").await;
+        let alias = disk(&root, "first").await;
+        let second = disk(&root, "second").await;
+        let bytes = payload("distinct-replicas");
+        let disks = [first.clone(), first, alias, second];
+        let publication = publish_committed_snapshot(&disks, Uuid::new_v4(), 1, &bytes, 4096)
+            .await
+            .expect("duplicate handles must not contend with their own replica CAS");
+        assert_eq!(publication.payload_replicas, 2, "count canonical disk roots, not handles");
+        assert_eq!(publication.manifest_replicas, 2);
+        assert_eq!(
+            read_committed(&disks, 4096)
+                .await
+                .expect("read both independent replicas")
+                .expect("committed snapshot")
+                .payload(),
+            bytes
+        );
+    }
+
     #[tokio::test]
     async fn committed_snapshot_writer_publishes_100k_records_with_bounded_readback() {
         let root = TempDir::new().expect("test directory");
@@ -1521,16 +1573,17 @@ mod tests {
             .await
             .ok();
         assert_eq!(
-            cas_replace(&store, PAYLOAD_PATHS[1], &next, 4096)
+            cas_replace(&store, PAYLOAD_PATHS[1], EcstoreDiskBytes::copy_from_slice(&next), 4096)
                 .await
                 .expect("successor payload CAS"),
             EcstoreConditionalFileUpdate::Updated
         );
         install(&store, MANIFEST_PATHS[1], &damaged_manifest).await;
 
-        let manifest_update = cas_replace_expected(&store, MANIFEST_PATHS[1], expected_manifest, &manifest(owner, 2, &next))
-            .await
-            .expect("successor manifest CAS");
+        let manifest_update =
+            cas_replace_expected(&store, MANIFEST_PATHS[1], expected_manifest, manifest(owner, 2, &next).into())
+                .await
+                .expect("successor manifest CAS");
         assert_eq!(manifest_update, EcstoreConditionalFileUpdate::Mismatch);
 
         let reopened = disk(&root, "disk").await;

crates/scanner/src/remote_scanner/stream.rs

@@ -740,7 +740,11 @@ async fn scan_and_persist_local_bucket(
                 scan_mode,
             )
             .await
-            .ok(),
+            .map_err(|err| {
+                RemoteScannerServerError::retry_bucket(format!(
+                    "remote namespace scanner checkpoint identity unavailable during admission: {err}"
+                ))
+            })?,
         },
     );
     match scan_state {
@@ -909,8 +913,11 @@ async fn scan_and_persist_local_bucket(
             scan_mode,
         )
         .await
-        .ok()
-            != Some(expected)
+        .map_err(|err| {
+            RemoteScannerServerError::retry_bucket(format!(
+                "remote namespace scanner checkpoint identity unavailable after scanning: {err}"
+            ))
+        })? != Some(expected)
     {
         return Err(RemoteScannerServerError::retry_bucket(
             "remote scanner checkpoint identity changed during scanning",

crates/scanner/src/scanner/tests/cycle_budget_publication.rs

@@ -169,24 +169,30 @@ async fn progress_proof_is_bounded_without_a_runtime_budget() {
 #[tokio::test]
 #[serial]
 async fn coordinator_persists_budget_progress_without_usage_or_dirty_acknowledgement() {
-    assert_coordinator_partial_progress(false, None).await;
+    assert_coordinator_partial_progress(false, None, false).await;
+}
+
+#[tokio::test]
+#[serial]
+async fn coordinator_preserves_checkpoint_when_bucket_identity_is_unavailable() {
+    assert_coordinator_partial_progress(false, None, true).await;
 }
 
 #[tokio::test]
 #[serial]
 async fn coordinator_persists_mixed_plan_tail_without_usage_or_dirty_acknowledgement() {
-    assert_coordinator_partial_progress(true, None).await;
+    assert_coordinator_partial_progress(true, None, false).await;
 }
 
 #[tokio::test]
 #[serial]
 async fn coordinator_partial_progress_requires_successful_cycle_state_save() {
     for after_commit in [false, true] {
-        assert_coordinator_partial_progress(true, Some(after_commit)).await;
+        assert_coordinator_partial_progress(true, Some(after_commit), false).await;
     }
 }
 
-async fn assert_coordinator_partial_progress(resume_mixed_plan: bool, fail_cycle_save: Option<bool>) {
+async fn assert_coordinator_partial_progress(resume_mixed_plan: bool, fail_cycle_save: Option<bool>, block_identity_read: bool) {
     crate::scanner_io::clear_dirty_usage_buckets_for_tests();
     let (_temp_dir, store) = setup_scanner_cycle_store().await;
     let bucket = "scanner-budget-progress";
@@ -265,6 +271,100 @@ async fn assert_coordinator_partial_progress(resume_mixed_plan: bool, fail_cycle
     );
     assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before);
 
+    if block_identity_read {
+        let set = store.pools[0].disk_set[0].clone();
+        let cache_name = path_join_buf(&[bucket, DATA_USAGE_CACHE_NAME]);
+        let mut checkpoint = DataUsageCache::default();
+        let revisions = checkpoint
+            .load_with_revisions(set.clone(), &cache_name)
+            .await
+            .expect("load the saved checkpoint before blocking its identity read");
+        assert!(checkpoint.info.scan_identity.is_some());
+        assert!(checkpoint.has_scan_position());
+        let before = serde_json::to_value(&checkpoint).expect("serialize the existing checkpoint");
+        let identity_key = path_join_buf(&[crate::BUCKET_META_PREFIX, bucket, ".bucket-incarnation"]);
+        let lock = store
+            .new_ns_lock(RUSTFS_META_BUCKET, &identity_key)
+            .await
+            .expect("bucket identity object lock");
+        let guard = lock
+            .get_write_lock(Duration::from_secs(30))
+            .await
+            .expect("hold the identity sidecar unavailable to scanner reads");
+        let blocked_budget = ScannerCycleBudget::new_with_progress_tracking(
+            &ctx,
+            ScannerCycleBudgetConfig {
+                max_objects: Some(1),
+                ..Default::default()
+            },
+        );
+        let blocked_outcome = tokio::time::timeout(
+            Duration::from_secs(60),
+            run_data_scanner_cycle_with_budget(
+                &ctx,
+                &store,
+                &mut cycle,
+                &mut revision,
+                1,
+                Arc::clone(&blocked_budget),
+                ScannerCycleScheduling {
+                    requires_full_scan: true,
+                    service_cohort: None,
+                },
+            ),
+        )
+        .await
+        .expect("blocked identity reads must finish within the existing acquisition bounds");
+        assert_eq!(
+            blocked_outcome,
+            ScannerCycleOutcome::Failed,
+            "no verified bucket work must not publish a cycle"
+        );
+        assert_eq!(blocked_budget.progress().0, 0, "unverified identity must not start a legacy rebuild");
+        let after_revisions = checkpoint
+            .load_with_revisions(set, &cache_name)
+            .await
+            .expect("reload the retained checkpoint while only the identity object is locked");
+        assert_eq!(after_revisions, revisions, "unavailable identity must not overwrite saved progress");
+        assert_eq!(serde_json::to_value(&checkpoint).expect("serialize retained checkpoint"), before);
+        assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before);
+        drop(guard);
+
+        let recovery_budget = ScannerCycleBudget::new_with_progress_tracking(
+            &ctx,
+            ScannerCycleBudgetConfig {
+                max_objects: Some(1),
+                ..Default::default()
+            },
+        );
+        let recovered = run_data_scanner_cycle_with_budget(
+            &ctx,
+            &store,
+            &mut cycle,
+            &mut revision,
+            1,
+            Arc::clone(&recovery_budget),
+            ScannerCycleScheduling {
+                requires_full_scan: true,
+                service_cohort: None,
+            },
+        )
+        .await;
+        assert_eq!(
+            recovered,
+            ScannerCycleOutcome::Partial,
+            "the same checkpoint must resume after identity reads recover"
+        );
+        assert_eq!(recovery_budget.progress().0, 1);
+        checkpoint
+            .load(store.pools[0].disk_set[0].clone(), &cache_name)
+            .await
+            .expect("reload resumed progress");
+        assert!(checkpoint.info.scan_identity.is_some());
+        assert!(checkpoint.has_scan_position());
+        assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before);
+    }
+
     if resume_mixed_plan {
         let set = store.pools[0].disk_set[0].clone();
         let cache_name = path_join_buf(&[bucket, DATA_USAGE_CACHE_NAME]);

crates/scanner/src/scanner_io.rs

@@ -1502,8 +1502,10 @@ pub(crate) async fn scanner_bucket_checkpoint_identity(
     publication_epoch: u64,
     tier_registry_generation: u64,
     scan_mode: HealScanMode,
-) -> Result<crate::DataUsageScanIdentity> {
-    let bucket_incarnation = set.bucket_incarnation_id_from_disk(bucket).await?;
+) -> Result<Option<crate::DataUsageScanIdentity>> {
+    let Some(bucket_incarnation) = set.find_bucket_incarnation_id_from_disk(bucket).await? else {
+        return Ok(None);
+    };
     let disks = set
         .format
         .erasure
@@ -1519,14 +1521,14 @@ pub(crate) async fn scanner_bucket_checkpoint_identity(
     for disk in disks {
         digest.update(disk.as_bytes());
     }
-    Ok(crate::DataUsageScanIdentity {
+    Ok(Some(crate::DataUsageScanIdentity {
         version: 1,
         bucket_incarnation,
         set_layout: crate::DataUsageScanPlanDigest(digest.finalize().into()),
         publication_epoch,
         tier_registry_generation,
         scan_mode,
-    })
+    }))
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]

crates/scanner/src/scanner_io/io_cache.rs

@@ -894,10 +894,10 @@ impl ScannerIOCache for SetDisks {
                                         event = EVENT_SCANNER_DISK_BUCKET_STATE,
                                         component = LOG_COMPONENT_SCANNER,
                                         subsystem = LOG_SUBSYSTEM_IO,
-                                        bucket = %bucket.name,
                                         state = "remote_zero_progress_requeued",
+                                        bucket = %bucket.name,
                                         error = %e,
-                                        "Remote scanner requeued bucket rejected before execution"
+                                        "Remote scanner requeued bucket work"
                                     );
                                     break;
                                 }
@@ -1031,7 +1031,7 @@ impl ScannerIOCache for SetDisks {
                     };
                     // Lack of an authoritative legacy identity disables the new
                     // checkpoint protocol; the existing full rebuild remains available.
-                    let checkpoint_identity = scanner_bucket_checkpoint_identity(
+                    let checkpoint_identity = match scanner_bucket_checkpoint_identity(
                         &store_clone_clone,
                         &bucket.name,
                         expected_publication_epoch_clone,
@@ -1039,7 +1039,28 @@ impl ScannerIOCache for SetDisks {
                         scan_mode,
                     )
                     .await
-                    .ok();
+                    {
+                        Ok(identity) => identity,
+                        Err(error) => {
+                            record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
+                            warn!(
+                                target: "rustfs::scanner::io",
+                                event = EVENT_SCANNER_DISK_BUCKET_STATE,
+                                component = LOG_COMPONENT_SCANNER,
+                                subsystem = LOG_SUBSYSTEM_IO,
+                                state = "checkpoint_identity_unavailable",
+                                stage = "admission",
+                                bucket = %bucket.name,
+                                pool = source.pool_index,
+                                set = source.set_index,
+                                cycle = want_cycle,
+                                leader_epoch,
+                                error = %error,
+                                "Scanner retained the bucket checkpoint after an identity lookup failure"
+                            );
+                            continue;
+                        }
+                    };
                     let scan_state = current_cache_root_or_prepare_with_generation(
                         &mut cache,
                         &bucket.name,
@@ -1260,8 +1281,8 @@ impl ScannerIOCache for SetDisks {
                             }
                         }
                     };
-                    if let Some(expected) = checkpoint_identity
-                        && scanner_bucket_checkpoint_identity(
+                    if let Some(expected) = checkpoint_identity {
+                        let current_identity = match scanner_bucket_checkpoint_identity(
                             &store_clone_clone,
                             &bucket.name,
                             expected_publication_epoch_clone,
@@ -1269,11 +1290,46 @@ impl ScannerIOCache for SetDisks {
                             scan_mode,
                         )
                         .await
-                        .ok()
-                            != Some(expected)
-                    {
-                        record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
-                        continue;
+                        {
+                            Ok(identity) => identity,
+                            Err(error) => {
+                                record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
+                                warn!(
+                                    target: "rustfs::scanner::io",
+                                    event = EVENT_SCANNER_DISK_BUCKET_STATE,
+                                    component = LOG_COMPONENT_SCANNER,
+                                    subsystem = LOG_SUBSYSTEM_IO,
+                                    state = "checkpoint_identity_unavailable",
+                                    stage = "post_scan",
+                                    bucket = %bucket.name,
+                                    pool = source.pool_index,
+                                    set = source.set_index,
+                                    cycle = want_cycle,
+                                    leader_epoch,
+                                    error = %error,
+                                    "Scanner could not verify the bucket identity before persistence"
+                                );
+                                continue;
+                            }
+                        };
+                        if current_identity != Some(expected) {
+                            record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
+                            warn!(
+                                target: "rustfs::scanner::io",
+                                event = EVENT_SCANNER_DISK_BUCKET_STATE,
+                                component = LOG_COMPONENT_SCANNER,
+                                subsystem = LOG_SUBSYSTEM_IO,
+                                state = "checkpoint_identity_changed",
+                                stage = "post_scan",
+                                bucket = %bucket.name,
+                                pool = source.pool_index,
+                                set = source.set_index,
+                                cycle = want_cycle,
+                                leader_epoch,
+                                "Scanner rejected the bucket result after an identity change"
+                            );
+                            continue;
+                        }
                     }
                     let scan_outcome = match scan_result {
                         Ok(scan_outcome) => scan_outcome,

crates/scanner/src/scanner_io/tests.rs

@@ -1060,6 +1060,7 @@ async fn checkpoint_fixture_bucket_identity_uses_its_set_instance_owner() {
     let first_identity =
         scanner_bucket_checkpoint_identity(&first.pools[0].disk_set[0], "checkpoint-identity", 0, 7, HealScanMode::Normal)
             .await
+            .expect("first durable identity lookup")
             .expect("first durable identity");
     let (_second_dir, second) = setup_two_pool_scanner_store().await;
     second
@@ -1069,18 +1070,20 @@ async fn checkpoint_fixture_bucket_identity_uses_its_set_instance_owner() {
     let second_identity =
         scanner_bucket_checkpoint_identity(&second.pools[0].disk_set[0], "checkpoint-identity", 0, 7, HealScanMode::Normal)
             .await
+            .expect("second durable identity lookup")
             .expect("second durable identity");
     assert_ne!(first_identity.bucket_incarnation, second_identity.bucket_incarnation);
     assert_eq!(
         scanner_bucket_checkpoint_identity(&first.pools[0].disk_set[0], "checkpoint-identity", 0, 7, HealScanMode::Normal)
             .await
             .expect("first owner remains bound"),
-        first_identity
+        Some(first_identity)
     );
     assert!(
         scanner_bucket_checkpoint_identity(&first.pools[0].disk_set[0], "missing-checkpoint-bucket", 0, 7, HealScanMode::Normal)
             .await
-            .is_err()
+            .expect("missing identity must be distinguished from a read failure")
+            .is_none()
     );
 }
 

docs/architecture/scanner-usage-publication.md

@@ -287,6 +287,8 @@ worker finishing does not certify its namespace coverage.
 | `bucket_checkpoint_prepared` | Checkpoint admission identifies the bucket, pool/set, cycle/leader epoch, and previous cycle/epoch. `resume_position` distinguishes retained traversal position from a fresh sweep; `mixed_coverage` reports whether that retained prefix began under another plan. |
 | `bucket_sweep_incomplete` | A bucket reached the end of a sweep but needs another verification pass. `unresolved_objects` reports the complete coverage/health gate, including failed child-cache entries that can be absent from the root counter and diagnostic failure map. Inspect it with `coverage_gap`, `root_failed_objects`, `failed_objects`, `size_reconciliation_entries`, and `mixed_coverage`, using bucket, pool/set, and cycle/leader epoch. `resume_position_on_entry` identifies a sweep entered with traversal position; `resume_position` reports the returned cache position after terminal cleanup. Multiple blockers can coexist. |
 | `partial_saved_not_published` | The local or remote worker successfully saved a terminal Partial cache. Correlate its bucket, pool/set, cycle/leader epoch and `resume_position` with admission and sweep events. This confirms persistence, not complete coverage or authoritative publication. |
+| `checkpoint_identity_unavailable` | A local worker could not read the persisted bucket identity. `stage=admission` preserves the existing checkpoint without preparing a legacy rebuild; `stage=post_scan` rejects the unverified candidate before persistence. Correlate bucket, pool/set, cycle/epoch and the underlying read error. Remote terminal errors distinguish the same two stages. |
+| `checkpoint_identity_changed` | A successful post-scan read no longer matches the admitted identity, including a now-absent sidecar. Rejecting the result protects against bucket replacement; this differs from an unavailable read. |
 | `remote_scan_partial` | The coordinator received a Partial terminal response for the named bucket and disk. Correlate the worker's bucket-sweep or checkpoint events; Partial is not a complete bucket root. |
 | `bucket_result_rejected` | A bucket result did not match the cycle's frozen tier-registry generation and was not applied. |
 | `set_cache_publish_skipped` | `completed_buckets` counts distinct bucket roots successfully applied to this set cache. Use `first_incomplete_bucket` with pool/set/cycle to locate one actual gap in this set. Inspect `unprocessed_buckets` and budget/cancellation state, then correlate that bucket's worker events. |
@@ -311,6 +313,19 @@ pool/set across consecutive cycles. Samples from different sets cannot establish
 a repeated mixed loop. Check save failures, remote interruption, and leadership
 changes if only periodic progress survives.
 
+Identity lookup has three outcomes: a verified UUID, a confirmed absent sidecar,
+or a read failure. Only confirmed absence may use the existing full-rebuild path
+without checkpoint identity. A lock, transport, or corrupt-metadata error must
+not become absence and reset a valid frontier. A successful read that proves a
+different identity still invalidates the old result.
+
+The `.bucket-incarnation@latest` resource belongs to a read of the persisted
+identity sidecar. Scanner identity checks occur before a walk and again before
+saving its result; set inventory preflight also reads identities. A single
+remote lock-client warning can coexist with a successful quorum, so it does not
+prove that checkpoint persistence failed. Use the scanner's identity stage and
+the terminal save result to establish that connection.
+
 An unknown bucket cannot be quarantined as zero in an authoritative total.
 Existing pending-bucket semantics still require complete structural coverage.
 A separately labelled subtotal would need proof for each included bucket across