feat(scanner): observe cache read origins

fix(ecstore): resolve dangling latest versions in metadata updates (#9009 )
* fix(ecstore): select the latest version at read quorum like MinIO GET and HEAD escalated latest-version selection to write quorum when enough metadata answered and skipped a newer version below write quorum on degraded reads. LIST and heal already pick the newer side of a split write at read quorum, so such keys listed but answered 503 until heal. Select with MinIO's getObjectFileInfo rule: the most common mod time that reaches the object's data blocks wins, ties go to the newer time, and one content identity at that time must reach the same quorum. Two identities tied at the selected time fail instead of resolving by hash order, and the selected identity's version-stack shape follows disk order. * fix(ecstore): resolve selection ties by disk order and guard writes Review of the read-quorum selection found three gaps: - Rejecting a same-time half split made explicit-version reads of a partially applied tagging update fail and blocked the retry that repairs it. Resolve ties to the identity on the lowest-index disk, for reads and heal alike, instead of rejecting or using map order. - Delete markers and empty versions can lower the common read quorum below the selected version's data blocks, so HEAD could report a version GET cannot decode. Require that many copies, as MinIO's getLatestFileInfo does. - A selection can now return exactly read-quorum disks, and partial metadata writes are not rolled back. Refuse an update before writing when its target disks cannot reach write quorum. * fix(ecstore): count identity copies in the data-blocks guard The first-candidate shortcut counts byte-identical copies, so copies that differ only in identity-neutral fields such as replication status were missing from the data-blocks check. When the shortcut's count falls short, count every copy of the selected content identity before refusing the version. * fix(ecstore): keep commit-proof selection for system-bucket reads Pool metadata's PREPARE/COMMIT recovery and identity epochs treat an ordinary read of pool.bin as proof that the record committed. Under the read-quorum rule, a COMMIT that reached only half of a 2+2 set was served as committed, and the pool_meta_native_physical_* integration tests failed. Route reads through select_fileinfo_for_read: S3 buckets keep MinIO's read-quorum rule, while is_meta_bucketname reads use the previous algorithm unchanged as select_committed_fileinfo, so a split stays unreadable until native recovery resolves it. * fix(ecstore): serve the latest readable version past dangling versions A failed versioned PUT whose rollback also failed can leave the new version on top of fewer drives than its data blocks. Reading only each drive's top version then left the committed version below it short of read quorum, so GET and HEAD of the latest version answered 503 although that version is intact in every stack. MinIO resolves the stacks in pickLatestQuorumFilesInfo. When every drive answered and an S3 latest read still cannot select a version, walk the newest version headers of each stack through exact-version reads: serve the readable version with the newest selected modification time, skip versions whose exact read proves them dangling (the NoSuchVersion an explicit GET reports) and queue heal for them, and keep the read-quorum error on any other outcome. Candidates are ordered by each version's newest modification time across drives, and at most eight exact reads are spent. Test PUTs now wait for every drive so tests that edit one drive's metadata no longer race the PUT's background tail writes. * fix(ecstore): resolve dangling latest versions in metadata updates A failed versioned PUT whose rollback also failed can leave the new version on fewer drives than its data blocks. GET and HEAD of the latest version already look past such versions, but read-before-write callers without a version id still answered 503: object tagging read each drive's top version with the quorum-error read, and put_object_metadata and the CopyObject source read selected from read_all_xl, whose merged latest version is missing on the other drives, so object_quorum_from_meta failed before selection. Object tagging now reads through a variant that runs the same walk past dangling versions as a latest GET while keeping the quorum error for a dangling key. put_object_metadata and the CopyObject source try the walk when their quorum or selection step fails with read quorum, before the existing dangling cleanup could report a key with a committed version absent. A version-only copy takes the served version's own copies and write quorum. The walk still applies only to S3 buckets, latest reads, and sets where every drive answered, and the update still targets only the served version's disks, so a version below write quorum is refused before writing. * fix(ecstore): refuse version-only copies below write quorum up front Review of the dangling-latest metadata updates found three gaps: - A version-only CopyObject wrote the new version and reverted on a write-quorum miss; refuse it before writing when the served version's disks cannot reach its write quorum, like update_object_meta_with_opts. - Replication status write-back, including the zero-write ValidateOnly mode, now keeps its read-quorum error and never walks past dangling versions, since the walk queues heal for the versions it skips. - The erasure-coding doc overstated that these callers never report a dangling key as absent; put_object_metadata and CopyObject keep their existing dangling cleanup when the walk finds no readable version.
2026-10-29 23:37:04 +00:00 · 2026-10-28 11:01:05 +08:00 · 2026-10-28 08:59:37 +08:00 · 2026-10-28 08:40:55 +08:00
9 changed files with 1566 additions and 987 deletions
@@ -8839,7 +8839,7 @@ impl SetDisks {
        // quorum, failing write quorum on update_object_meta (backlog#872).
        let read_opts = opts.as_read_opts();
        let (mut fi, _, disks) = self
-            .get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
+            .get_object_fileinfo_for_metadata_update(bucket, object, &read_opts)
            .await?
            .into_owned();

@@ -9294,36 +9294,56 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
            }
        };

-        let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) {
-            Ok((r, w)) => (
-                usize::try_from(r)
-                    .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![src_bucket, src_object]))?,
-                usize::try_from(w)
-                    .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![src_bucket, src_object]))?,
-            ),
+        let src_version_id = src_opts.version_id.as_deref().unwrap_or_default();
+        let quorum = Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count);
+        let selection = match &quorum {
+            Ok((read_quorum, write_quorum)) => {
+                let read_quorum = usize::try_from(*read_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![src_bucket, src_object]))?;
+                let write_quorum = usize::try_from(*write_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![src_bucket, src_object]))?;
+                Self::select_fileinfo_for_read(src_bucket, &disks, &metas, &errs, src_version_id, read_quorum, write_quorum)
+                    .map(|(online_disks, fi)| (online_disks, fi, write_quorum))
+            }
+            Err(err) => Err(err.clone()),
+        };
+        let (online_disks, mut fi, write_quorum) = match selection {
+            Ok(selection) => selection,
            Err(mut err) => {
-                if err == DiskError::ErasureReadQuorum
-                    && !src_bucket.starts_with(RUSTFS_META_BUCKET)
-                    && self
-                        .delete_if_dangling(src_bucket, src_object, &metas, &errs, &HashMap::new(), src_opts.clone())
-                        .await
-                        .is_ok()
+                // The merged latest version can be one a failed PUT left below
+                // quorum; resolve it like a latest GET before the dangling
+                // cleanup below can report a key with a committed version absent.
+                if let Some(snapshot) = self
+                    .select_latest_past_dangling_versions(src_bucket, src_object, src_opts, true, &disks, &err)
+                    .await
                {
-                    if src_opts.version_id.is_some() {
-                        err = DiskError::FileVersionNotFound
-                    } else {
-                        err = DiskError::FileNotFound
+                    // A version-only copy writes the served version's own
+                    // copies, so they and its write quorum replace the latest
+                    // read's.
+                    let (fi, parts_metadata, online_disks) = snapshot.into_owned();
+                    metas = parts_metadata;
+                    let write_quorum = fi.write_quorum(self.default_write_quorum());
+                    (online_disks, fi, write_quorum)
+                } else {
+                    if quorum.is_err()
+                        && err == DiskError::ErasureReadQuorum
+                        && !src_bucket.starts_with(RUSTFS_META_BUCKET)
+                        && self
+                            .delete_if_dangling(src_bucket, src_object, &metas, &errs, &HashMap::new(), src_opts.clone())
+                            .await
+                            .is_ok()
+                    {
+                        if src_opts.version_id.is_some() {
+                            err = DiskError::FileVersionNotFound
+                        } else {
+                            err = DiskError::FileNotFound
+                        }
                    }
+                    return Err(to_object_err(err.into(), vec![src_bucket, src_object]));
                }
-                return Err(to_object_err(err.into(), vec![src_bucket, src_object]));
            }
        };

-        let src_version_id = src_opts.version_id.as_deref().unwrap_or_default();
-        let (online_disks, mut fi) =
-            Self::select_fileinfo_for_read(src_bucket, &disks, &metas, &errs, src_version_id, read_quorum, write_quorum)
-                .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
-
        if fi.deleted {
            if src_opts.version_id.is_none() {
                return Err(to_object_err(Error::FileNotFound, vec![src_bucket, src_object]));
@@ -9461,6 +9481,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                }
            }

+            // A version-only copy writes and then reverts on a quorum miss;
+            // refuse it before writing when the served version's disks cannot
+            // reach its write quorum even if every write succeeds.
+            let reachable = online_disks
+                .iter()
+                .map(|disk| disk.is_none().then_some(DiskError::DiskNotFound))
+                .collect::<Vec<_>>();
+            if let Some(err) = reduce_write_quorum_errs(&reachable, OBJECT_OP_IGNORED_ERRS, write_quorum) {
+                return Err(to_object_err(err.into(), vec![src_bucket, src_object]));
+            }
            Self::write_unique_file_info(&online_disks, "", src_bucket, src_object, &metas, write_quorum)
                .await
                .map_err(|e| to_object_err(e.into(), vec![src_bucket, src_object]))?;
@@ -11147,45 +11177,63 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
            }
        };

-        let (read_quorum, write_quorum) = match Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count) {
-            Ok((read_quorum, write_quorum)) => (read_quorum, write_quorum),
+        let version_id = opts.version_id.as_deref().unwrap_or_default();
+        let quorum = Self::object_quorum_from_meta(&metas, &errs, self.default_parity_count);
+        let selection = match &quorum {
+            Ok((read_quorum, write_quorum)) => {
+                let read_quorum = usize::try_from(*read_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
+                let write_quorum = usize::try_from(*write_quorum)
+                    .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
+                Self::select_fileinfo_for_read(bucket, &disks, &metas, &errs, version_id, read_quorum, write_quorum)
+            }
+            Err(err) => Err(err.clone()),
+        };
+        let (online_disks, mut fi) = match selection {
+            Ok(selection) => selection,
            Err(mut err) => {
-                // A replication terminal CAS is a validation/write-back
-                // attempt, never a heal operation. In particular,
-                // ValidateOnly must remain zero-write, and a worker whose
-                // outer replication lease is already lost must not delete the
-                // surviving xl.meta copies before the namespace fence below
-                // can reject it. Leave dangling cleanup to heal/scanner.
-                if opts.replication_status_writeback.is_some() {
+                // The merged latest version can be one a failed PUT left below
+                // quorum; resolve it like a latest GET before the dangling
+                // cleanup below can report a key with a committed version absent.
+                // A replication terminal CAS stays side-effect free: the walk
+                // queues heal for the versions it skips, so it does not run.
+                let resolved = if opts.replication_status_writeback.is_none() {
+                    self.select_latest_past_dangling_versions(bucket, object, opts, false, &disks, &err)
+                        .await
+                } else {
+                    None
+                };
+                if let Some(snapshot) = resolved {
+                    let (fi, _, online_disks) = snapshot.into_owned();
+                    (online_disks, fi)
+                } else {
+                    // A replication terminal CAS is a validation/write-back
+                    // attempt, never a heal operation. In particular,
+                    // ValidateOnly must remain zero-write, and a worker whose
+                    // outer replication lease is already lost must not delete
+                    // the surviving xl.meta copies before the namespace fence
+                    // below can reject it. Leave dangling cleanup to
+                    // heal/scanner.
+                    if quorum.is_err()
+                        && opts.replication_status_writeback.is_none()
+                        && err == DiskError::ErasureReadQuorum
+                        && !bucket.starts_with(RUSTFS_META_BUCKET)
+                        && self
+                            .delete_if_dangling(bucket, object, &metas, &errs, &HashMap::new(), opts.clone())
+                            .await
+                            .is_ok()
+                    {
+                        if opts.version_id.is_some() {
+                            err = DiskError::FileVersionNotFound
+                        } else {
+                            err = DiskError::FileNotFound
+                        }
+                    }
                    return Err(to_object_err(err.into(), vec![bucket, object]));
                }
-                if err == DiskError::ErasureReadQuorum
-                    && !bucket.starts_with(RUSTFS_META_BUCKET)
-                    && self
-                        .delete_if_dangling(bucket, object, &metas, &errs, &HashMap::new(), opts.clone())
-                        .await
-                        .is_ok()
-                {
-                    if opts.version_id.is_some() {
-                        err = DiskError::FileVersionNotFound
-                    } else {
-                        err = DiskError::FileNotFound
-                    }
-                }
-                return Err(to_object_err(err.into(), vec![bucket, object]));
            }
        };

-        let read_quorum =
-            usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
-        let write_quorum = usize::try_from(write_quorum)
-            .map_err(|_| to_object_err(DiskError::ErasureWriteQuorum.into(), vec![bucket, object]))?;
-
-        let version_id = opts.version_id.as_deref().unwrap_or_default();
-        let (online_disks, mut fi) =
-            Self::select_fileinfo_for_read(bucket, &disks, &metas, &errs, version_id, read_quorum, write_quorum)
-                .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
-
        if fi.deleted {
            return Err(to_object_err(Error::MethodNotAllowed, vec![bucket, object]));
        }
@@ -278,6 +278,10 @@ const MAX_DANGLING_LATEST_VERSIONS: usize = 8;
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum DanglingRead {
    KeepQuorumError,
+    /// Metadata updates of the latest version keep the quorum error but, like
+    /// object reads, look past newer versions proven dangling
+    /// (`read_latest_past_dangling_versions`).
+    SkipDanglingVersions,
    ReportNotFound,
 }

@@ -729,9 +733,10 @@ impl SetDisks {
    }

    /// Like `get_object_fileinfo`, but `allow_early_stop=false` forces the full
-    /// quorum fanout. Read-before-write callers (object tagging) must use this:
-    /// the returned online-disk set is the write target, and the early-stop
-    /// subset would fail write quorum (backlog#872 regression fix).
+    /// quorum fanout. Read-before-write callers must use this or
+    /// `get_object_fileinfo_for_metadata_update`: the returned online-disk set
+    /// is the write target, and the early-stop subset would fail write quorum
+    /// (backlog#872 regression fix).
    #[tracing::instrument(level = "debug", skip(self))]
    pub(in crate::set_disk) async fn get_object_fileinfo_gated(
        &self,
@@ -753,6 +758,64 @@ impl SetDisks {
        .await
    }

+    /// The full-fanout read behind a metadata update such as object tagging.
+    /// A latest read whose newest versions never committed resolves to the
+    /// version a latest GET serves, while a dangling key keeps the quorum
+    /// error. The returned online disks are the write target.
+    #[tracing::instrument(level = "debug", skip(self))]
+    pub(in crate::set_disk) async fn get_object_fileinfo_for_metadata_update(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+    ) -> Result<GetObjectFileInfo> {
+        self.get_object_fileinfo_gated_inner(
+            bucket,
+            object,
+            opts,
+            false,
+            false,
+            MetadataReadPurpose::FullLayout,
+            DanglingRead::SkipDanglingVersions,
+        )
+        .await
+    }
+
+    /// Recovers a failed latest selection in read-before-write callers that
+    /// read and select metadata themselves (`put_object_metadata` and the
+    /// CopyObject source) the way a latest GET does. `None` keeps the caller's
+    /// error: a failure other than read quorum, an explicit version, the
+    /// system bucket's commit-proof rule, or a walk that finds no readable
+    /// version or cannot read every drive. The selected metadata keeps its
+    /// part checksums because a version-only copy carries them into the new
+    /// version.
+    pub(in crate::set_disk) async fn select_latest_past_dangling_versions(
+        &self,
+        bucket: &str,
+        object: &str,
+        opts: &ObjectOptions,
+        read_data: bool,
+        disks: &[Option<DiskStore>],
+        err: &DiskError,
+    ) -> Option<GetObjectFileInfo> {
+        if *err != DiskError::ErasureReadQuorum || opts.version_id.is_some() || crate::bucket::utils::is_meta_bucketname(bucket) {
+            return None;
+        }
+        let opts = ObjectOptions {
+            include_part_checksums: true,
+            ..opts.clone()
+        };
+        Box::pin(self.read_latest_past_dangling_versions(
+            bucket,
+            object,
+            &opts,
+            read_data,
+            MetadataReadPurpose::FullLayout,
+            disks,
+        ))
+        .await
+    }
+
    #[allow(clippy::too_many_arguments)]
    async fn get_object_fileinfo_gated_inner(
        &self,
@@ -978,7 +1041,7 @@ impl SetDisks {
                Err(err) => {
                    if err == DiskError::ErasureReadQuorum
                        && vid.is_empty()
-                        && dangling_read == DanglingRead::ReportNotFound
+                        && dangling_read != DanglingRead::KeepQuorumError
                        && metadata_fanout_complete
                        && !crate::bucket::utils::is_meta_bucketname(bucket)
                        && let Some(snapshot) =
@@ -9930,6 +9993,7 @@ mod dangling_read_tests {
    use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
    use crate::storage_api_contracts::object::ObjectOperations;
    use http::HeaderMap;
+    use std::collections::HashMap;
    use std::path::PathBuf;
    use tempfile::TempDir;
    use tokio::io::AsyncReadExt;
@@ -10451,6 +10515,353 @@ mod dangling_read_tests {
        assert_ne!(latest.version_id, Some(v2));
    }

+    /// Version one on every drive under a version two that a failed PUT left
+    /// on drives `0..3`, below the read quorum of four.
+    async fn put_committed_below_dangling(set: &SetDisks, bucket: &str, object: &str) -> (Uuid, Uuid) {
+        let v1 = put(set, bucket, object, b"version one", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        let v2 = put(set, bucket, object, b"version two", true)
+            .await
+            .version_id
+            .expect("versioned PUT should assign a version id");
+        drop_version_from(set, bucket, object, v2, 3..DRIVES).await;
+        (v1, v2)
+    }
+
+    /// Each drive's copy of `version_id`, or `None` where the drive lacks it.
+    async fn version_metadata(
+        set: &SetDisks,
+        bucket: &str,
+        object: &str,
+        version_id: Uuid,
+    ) -> Vec<Option<HashMap<String, String>>> {
+        let disks = set.disks.read().await.clone();
+        let mut copies = Vec::with_capacity(disks.len());
+        for disk in &disks {
+            let disk = disk.as_ref().expect("every drive should be online");
+            copies.push(
+                disk.read_version("", bucket, object, &version_id.to_string(), &ReadOptions::default())
+                    .await
+                    .ok()
+                    .map(|fi| fi.metadata),
+            );
+        }
+        copies
+    }
+
+    #[tokio::test]
+    async fn latest_tagging_skips_dangling_versions_above_a_committed_version() {
+        let bucket = "dangling-latest-tagging";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let (v1, v2) = put_committed_below_dangling(&set, bucket, object).await;
+        let tag_of = |metadata: &Option<HashMap<String, String>>| {
+            metadata
+                .as_ref()
+                .and_then(|metadata| metadata.get(rustfs_utils::http::headers::AMZ_OBJECT_TAGGING).cloned())
+        };
+
+        let tagged = set
+            .put_object_tags(bucket, object, "stage=committed", &ObjectOptions::default())
+            .await
+            .expect("PutObjectTagging must update the version a latest GET serves");
+        assert_eq!(tagged.version_id, Some(v1));
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            assert_eq!(
+                tag_of(metadata).as_deref(),
+                Some("stage=committed"),
+                "drive {disk} must carry the tag on version one"
+            );
+        }
+        for (disk, metadata) in version_metadata(&set, bucket, object, v2).await.iter().enumerate() {
+            assert_eq!(tag_of(metadata), None, "drive {disk} must leave the dangling version untouched");
+        }
+        assert_eq!(*head(&set, bucket, object, None).await.expect("latest HEAD").user_tags, "stage=committed");
+
+        let untagged = set
+            .delete_object_tags(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("DeleteObjectTagging must update the version a latest GET serves");
+        assert_eq!(untagged.version_id, Some(v1));
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            assert_eq!(tag_of(metadata).as_deref(), Some(""), "drive {disk} must clear the tag on version one");
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_metadata_update_skips_dangling_versions_above_a_committed_version() {
+        let bucket = "dangling-latest-metadata";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let (v1, v2) = put_committed_below_dangling(&set, bucket, object).await;
+
+        let updated = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "updated".to_string())])),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a latest metadata update must target the version a latest GET serves");
+        assert_eq!(updated.version_id, Some(v1));
+        assert!(updated.is_latest, "the updated version is the latest one");
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            let metadata = metadata.as_ref().expect("version one is on every drive");
+            assert_eq!(metadata.get("x-amz-meta-probe").map(String::as_str), Some("updated"), "drive {disk}");
+        }
+        for (disk, metadata) in version_metadata(&set, bucket, object, v2).await.iter().enumerate() {
+            assert!(
+                metadata
+                    .as_ref()
+                    .is_none_or(|metadata| !metadata.contains_key("x-amz-meta-probe")),
+                "drive {disk} must leave the dangling version untouched"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_copy_source_skips_dangling_versions_above_a_committed_version() {
+        let bucket = "dangling-latest-copy";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+
+        // A metadata-only self-copy rewrites the latest version in place.
+        let object = "torn/in-place";
+        let (v1, _) = put_committed_below_dangling(&set, bucket, object).await;
+        let mut source = head(&set, bucket, object, None).await.expect("latest HEAD");
+        assert_eq!(source.version_id, Some(v1));
+        source.metadata_only = true;
+        source.user_defined = Arc::new(HashMap::from([("x-amz-meta-copied".to_string(), "in-place".to_string())]));
+        let copied = set
+            .copy_object(
+                bucket,
+                object,
+                bucket,
+                object,
+                &mut source,
+                &ObjectOptions::default(),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("a CopyObject source without a version id must resolve like a latest GET");
+        assert_eq!(copied.version_id, Some(v1));
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            let metadata = metadata.as_ref().expect("version one is on every drive");
+            assert_eq!(metadata.get("x-amz-meta-copied").map(String::as_str), Some("in-place"), "drive {disk}");
+        }
+
+        // A version-only self-copy commits a new version that shares the
+        // served version's data.
+        let object = "torn/new-version";
+        let (_, v2) = put_committed_below_dangling(&set, bucket, object).await;
+        let v3 = Uuid::new_v4();
+        let mut source = head(&set, bucket, object, None).await.expect("latest HEAD");
+        source.version_only = true;
+        let copied = set
+            .copy_object(
+                bucket,
+                object,
+                bucket,
+                object,
+                &mut source,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+                &ObjectOptions {
+                    version_id: Some(v3.to_string()),
+                    versioned: true,
+                    object_lock_config_snapshot: Some(Arc::new(crate::object_api::ObjectLockConfigSnapshot::new(
+                        crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+                    ))),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a version-only CopyObject source without a version id must resolve like a latest GET");
+        assert_eq!(copied.version_id, Some(v3));
+        let latest = head(&set, bucket, object, None).await.expect("latest HEAD");
+        assert_eq!(latest.version_id, Some(v3));
+        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"version one");
+        assert!(
+            version_metadata(&set, bucket, object, v2).await[..3]
+                .iter()
+                .all(Option::is_some),
+            "the copy must not remove the dangling version; heal owns its cleanup"
+        );
+    }
+
+    #[tokio::test]
+    async fn latest_metadata_update_past_dangling_versions_keeps_write_quorum() {
+        // EC 3+3: read quorum three, write quorum four. Version one reached
+        // three drives, so it is readable but below write quorum; versions two
+        // and three are dangling on two drives each. A latest read serves
+        // version one, and a metadata update of it is refused before writing.
+        let bucket = "dangling-latest-write-quorum";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set_with(bucket, 6, 3).await;
+        let mut versions = Vec::new();
+        for body in [b"version zero".as_slice(), b"version one", b"version two", b"version three"] {
+            versions.push(
+                put(&set, bucket, object, body, true)
+                    .await
+                    .version_id
+                    .expect("versioned PUT should assign a version id"),
+            );
+        }
+        let [_, v1, v2, v3] = versions[..] else {
+            unreachable!("four versions were written");
+        };
+        drop_version_from(&set, bucket, object, v1, 3..6).await;
+        drop_version_from(&set, bucket, object, v2, 2..6).await;
+        drop_version_from(&set, bucket, object, v3, 0..3).await;
+        drop_version_from(&set, bucket, object, v3, 5..6).await;
+
+        let latest = head(&set, bucket, object, None)
+            .await
+            .expect("version one is readable at read quorum");
+        assert_eq!(latest.version_id, Some(v1));
+
+        let is_write_quorum_error =
+            |err: &StorageError| matches!(err, StorageError::ErasureWriteQuorum | StorageError::InsufficientWriteQuorum(_, _));
+        let Err(err) = set
+            .put_object_tags(bucket, object, "stage=refused", &ObjectOptions::default())
+            .await
+        else {
+            panic!("tagging a version below write quorum must be refused");
+        };
+        assert!(is_write_quorum_error(&err), "tagging keeps the write-quorum error: {err:?}");
+        let Err(err) = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "refused".to_string())])),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a metadata update of a version below write quorum must be refused");
+        };
+        assert!(is_write_quorum_error(&err), "the metadata update keeps the write-quorum error: {err:?}");
+        let v4 = Uuid::new_v4();
+        let mut source = head(&set, bucket, object, None).await.expect("latest HEAD");
+        source.version_only = true;
+        let Err(err) = set
+            .copy_object(
+                bucket,
+                object,
+                bucket,
+                object,
+                &mut source,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+                &ObjectOptions {
+                    version_id: Some(v4.to_string()),
+                    versioned: true,
+                    object_lock_config_snapshot: Some(Arc::new(crate::object_api::ObjectLockConfigSnapshot::new(
+                        crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+                    ))),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a version-only copy of a version below write quorum must be refused");
+        };
+        assert!(is_write_quorum_error(&err), "the version-only copy keeps the write-quorum error: {err:?}");
+        assert!(
+            version_metadata(&set, bucket, object, v4).await.iter().all(Option::is_none),
+            "a refused version-only copy must not write the new version to any drive"
+        );
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            assert!(
+                metadata.as_ref().is_none_or(|metadata| {
+                    !metadata.contains_key(rustfs_utils::http::headers::AMZ_OBJECT_TAGGING)
+                        && !metadata.contains_key("x-amz-meta-probe")
+                }),
+                "drive {disk} must not be written by a refused update"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn replication_status_writeback_never_walks_past_dangling_versions() {
+        // A replication terminal CAS stays side-effect free: it keeps the
+        // read-quorum error instead of resolving the latest version, so it
+        // neither queues heal for the dangling version nor writes version one.
+        let bucket = "dangling-latest-replication-cas";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        let (v1, _) = put_committed_below_dangling(&set, bucket, object).await;
+
+        let Err(err) = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "written".to_string())])),
+                    replication_status_writeback: Some(Box::new(crate::object_api::ReplicationStatusWritebackCondition {
+                        mode: crate::object_api::ReplicationStatusWritebackMode::ValidateOnly,
+                        ..Default::default()
+                    })),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a replication terminal CAS must not resolve past a dangling version");
+        };
+        assert!(is_read_quorum_error(&err), "the CAS keeps the read-quorum error: {err:?}");
+        for (disk, metadata) in version_metadata(&set, bucket, object, v1).await.iter().enumerate() {
+            let metadata = metadata.as_ref().expect("version one is on every drive");
+            assert!(!metadata.contains_key("x-amz-meta-probe"), "drive {disk} must not be written");
+        }
+    }
+
+    #[tokio::test]
+    async fn latest_metadata_updates_past_dangling_versions_require_every_drive_to_answer() {
+        let bucket = "dangling-latest-update-offline";
+        let object = "torn/latest";
+        let (_dirs, set) = dangling_test_set(bucket).await;
+        put_committed_below_dangling(&set, bucket, object).await;
+        let offline = set.disks.write().await[5].take().expect("drive five should start online");
+
+        let Err(err) = set
+            .put_object_tags(bucket, object, "stage=uncertain", &ObjectOptions::default())
+            .await
+        else {
+            panic!("tagging must not resolve past a version an offline drive may commit");
+        };
+        assert!(is_read_quorum_error(&err), "tagging keeps the read-quorum error: {err:?}");
+        let Err(err) = set
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    eval_metadata: Some(HashMap::from([("x-amz-meta-probe".to_string(), "uncertain".to_string())])),
+                    ..Default::default()
+                },
+            )
+            .await
+        else {
+            panic!("a metadata update must not resolve past a version an offline drive may commit");
+        };
+        assert!(is_read_quorum_error(&err), "the metadata update keeps the read-quorum error: {err:?}");
+        set.disks.write().await[5] = Some(offline);
+    }
+
    #[tokio::test]
    async fn dangling_classification_keeps_quorum_error_for_corrupt_metadata() {
        let bucket = "dangling-read-corrupt";
@@ -67,11 +67,12 @@ const METRIC_CACHE_SAVE_RETRY_TOTAL: &str = "rustfs_scanner_cache_save_retry_tot
 const METRIC_CACHE_SAVE_PAYLOAD_BYTES_TOTAL: &str = "rustfs_scanner_cache_save_payload_bytes_total";
 const METRIC_CACHE_SAVE_DURATION_SECONDS: &str = "rustfs_scanner_cache_save_duration_seconds";
 const METRIC_CACHE_BACKUP_REVISION_FAILURE_TOTAL: &str = "rustfs_scanner_cache_backup_revision_failure_total";
+const METRIC_CACHE_READ_ORIGIN_TOTAL: &str = "rustfs_scanner_cache_read_origin_total";
 const LOG_COMPONENT_SCANNER: &str = "scanner";
 const LOG_SUBSYSTEM_CACHE: &str = "cache";
 const EVENT_SCANNER_CACHE_LOAD_STATE: &str = "scanner_cache_load_state";
 const EVENT_SCANNER_CACHE_SAVE_STATE: &str = "scanner_cache_save_state";
-static CACHE_SAVE_METRICS_ONCE: Once = Once::new();
+static CACHE_METRICS_ONCE: Once = Once::new();

 pub const DATA_USAGE_SCAN_CHECKPOINT_VERSION: u16 = 1;
 pub const DATA_USAGE_RAW_ENUMERATION_CURSOR_VERSION: u16 = 1;
@@ -204,6 +205,54 @@ pub(crate) struct DataUsageCacheRevisions {
    backup: Option<DataUsageCacheRevision>,
 }

+/// The observed result of trying the logical primary cache slot.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum DataUsageCachePrimaryRead {
+    Missing,
+    Corrupt,
+}
+
+/// Process-local read provenance, not evidence of freshness or authority.
+/// Legacy objects may come from the old data-usage bucket. Empty may mean
+/// missing or corrupt objects; neither state certifies a complete snapshot.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum DataUsageCacheReadOrigin {
+    Primary {
+        legacy: bool,
+    },
+    Backup {
+        primary: DataUsageCachePrimaryRead,
+        legacy: bool,
+    },
+    Empty,
+}
+
+impl DataUsageCacheReadOrigin {
+    fn metric_label(self) -> &'static str {
+        match self {
+            Self::Primary { legacy: false } => "primary",
+            Self::Primary { legacy: true } => "legacy_primary",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: false,
+            } => "backup_primary_missing",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: true,
+            } => "legacy_backup_primary_missing",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: false,
+            } => "backup_primary_corrupt",
+            Self::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: true,
+            } => "legacy_backup_primary_corrupt",
+            Self::Empty => "empty",
+        }
+    }
+}
+
 pub static DATA_USAGE_BUCKET: LazyLock<String> =
    LazyLock::new(|| format!("{RUSTFS_META_BUCKET}{SLASH_SEPARATOR}{BUCKET_META_PREFIX}"));

@@ -1271,8 +1320,12 @@ impl DataUsageCache {
        Ok(())
    }

-    fn ensure_cache_save_metrics_registered() {
-        CACHE_SAVE_METRICS_ONCE.call_once(|| {
+    fn ensure_cache_metrics_registered() {
+        CACHE_METRICS_ONCE.call_once(|| {
+            describe_counter!(
+                METRIC_CACHE_READ_ORIGIN_TOTAL,
+                "Successful scanner cache loads by observed logical origin; does not prove cache freshness."
+            );
            describe_counter!(
                METRIC_CACHE_SAVE_ATTEMPT_TOTAL,
                "Total scanner data usage cache save attempts by result and cache type."
@@ -70,6 +70,7 @@ struct DataUsageCacheLoadResult {
    cache: DataUsageCache,
    main_revision: DataUsageCacheRevision,
    backup_revision: Option<DataUsageCacheRevision>,
+    origin: DataUsageCacheReadOrigin,
 }

 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -97,6 +98,7 @@ impl DataUsageCache {
    /// If the object is not found, a nil error with empty data usage cache is returned.
    pub async fn load<S: ScannerObjectIO>(&mut self, store: Arc<S>, name: &str) -> StorageResult<()> {
        let loaded = Self::load_cache(store, name).await?;
+        Self::record_cache_read_origin(loaded.origin);
        *self = loaded.cache;
        Ok(())
    }
@@ -106,6 +108,17 @@ impl DataUsageCache {
        store: Arc<S>,
        name: &str,
    ) -> StorageResult<DataUsageCacheRevisions> {
+        self.load_with_revisions_and_origin(store, name)
+            .await
+            .map(|(revisions, _)| revisions)
+    }
+
+    /// Origin is an observation of this load, not proof of current coverage.
+    pub(crate) async fn load_with_revisions_and_origin<S: ScannerObjectIO>(
+        &mut self,
+        store: Arc<S>,
+        name: &str,
+    ) -> StorageResult<(DataUsageCacheRevisions, DataUsageCacheReadOrigin)> {
        let backup_name = format!("{name}.bkp");
        let backup_path = path_join_buf(&[BUCKET_META_PREFIX, &backup_name]);
        let loaded = Self::load_cache(store.clone(), name).await?;
@@ -131,9 +144,16 @@ impl DataUsageCache {
            },
        };
        let main = loaded.main_revision;
+        let origin = loaded.origin;
+        Self::record_cache_read_origin(origin);
        *self = loaded.cache;

-        Ok(DataUsageCacheRevisions { main, backup })
+        Ok((DataUsageCacheRevisions { main, backup }, origin))
+    }
+
+    fn record_cache_read_origin(origin: DataUsageCacheReadOrigin) {
+        Self::ensure_cache_metrics_registered();
+        counter!(METRIC_CACHE_READ_ORIGIN_TOTAL, "origin" => origin.metric_label()).increment(1);
    }

    pub(crate) async fn read_revisions<S: ScannerObjectIO>(store: Arc<S>, name: &str) -> StorageResult<DataUsageCacheRevisions> {
@@ -156,10 +176,12 @@ impl DataUsageCache {
                    cache,
                    revision: Some(main_revision),
                } => {
+                    let legacy = matches!(&main_revision, DataUsageCacheRevision::Missing);
                    return Ok(DataUsageCacheLoadResult {
                        cache: *cache,
                        main_revision,
                        backup_revision: None,
+                        origin: DataUsageCacheReadOrigin::Primary { legacy },
                    });
                }
                DataUsageCacheLoadAttempt::Loaded { revision: None, .. } => {
@@ -193,10 +215,17 @@ impl DataUsageCache {
                        let main_revision = main_revision.ok_or_else(|| {
                            Error::other(format!("scanner cache main revision is unavailable while loading backup: {name}"))
                        })?;
+                        let primary = if main_state == DataUsageCacheLoadState::Corrupt {
+                            DataUsageCachePrimaryRead::Corrupt
+                        } else {
+                            DataUsageCachePrimaryRead::Missing
+                        };
+                        let legacy = matches!(&backup_revision, DataUsageCacheRevision::Missing);
                        return Ok(DataUsageCacheLoadResult {
                            cache: *cache,
                            main_revision,
                            backup_revision: Some(backup_revision),
+                            origin: DataUsageCacheReadOrigin::Backup { primary, legacy },
                        });
                    }
                    DataUsageCacheLoadState::Loaded
@@ -220,6 +249,7 @@ impl DataUsageCache {
                        main_revision: main_revision
                            .ok_or_else(|| Error::other(format!("scanner cache missing state has no revision: {name}")))?,
                        backup_revision,
+                        origin: DataUsageCacheReadOrigin::Empty,
                    });
                }
                (DataUsageCacheLoadState::Corrupt, DataUsageCacheLoadState::Missing)
@@ -239,6 +269,7 @@ impl DataUsageCache {
                        main_revision: main_revision
                            .ok_or_else(|| Error::other(format!("scanner cache corrupt state has no revision: {name}")))?,
                        backup_revision,
+                        origin: DataUsageCacheReadOrigin::Empty,
                    });
                }
                _ => {}
@@ -518,7 +549,7 @@ impl DataUsageCache {
        revision: Option<DataUsageCacheRevision>,
        expected_epoch: Option<u64>,
    ) -> StorageResult<()> {
-        Self::ensure_cache_save_metrics_registered();
+        Self::ensure_cache_metrics_registered();
        let path_type = Self::cache_path_type(path);
        let path = path.to_string();
        let sha256hex = (revision.is_some() && !buf.is_empty())
@@ -168,8 +168,10 @@ impl AsyncRead for PrefixThenErrorReader {
 #[derive(Debug)]
 struct CacheReadStore {
    main: CacheReadBody,
+    main_missing: bool,
    backup: Option<Vec<u8>>,
    legacy: Option<Vec<u8>>,
+    legacy_backup: Option<Vec<u8>>,
    puts: AtomicUsize,
    #[cfg(feature = "diagnostic-read-version-caller")]
    entry_sources: Mutex<Vec<Option<crate::storage_api::ecstore_diagnostic::ReadVersionEntrySource>>>,
@@ -185,8 +187,10 @@ impl CacheReadStore {
    fn new(main: CacheReadBody, backup: Option<Vec<u8>>) -> Self {
        Self {
            main,
+            main_missing: false,
            backup,
            legacy: None,
+            legacy_backup: None,
            puts: AtomicUsize::new(0),
            #[cfg(feature = "diagnostic-read-version-caller")]
            entry_sources: Mutex::new(Vec::new()),
@@ -239,8 +243,12 @@ impl ObjectIO for CacheReadStore {
            .await
            .push(crate::storage_api::ecstore_diagnostic::current_read_version_entry_source());
        if bucket == DATA_USAGE_BUCKET.as_str() {
-            return self
-                .legacy
+            let legacy = if object.ends_with(".bkp") {
+                &self.legacy_backup
+            } else {
+                &self.legacy
+            };
+            return legacy
                .clone()
                .map(|data| Self::reader(CacheReadBody::Bytes(data), "legacy-etag"))
                .ok_or(Error::FileNotFound);
@@ -252,6 +260,9 @@ impl ObjectIO for CacheReadStore {
        let main_path = path_join_buf(&[BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME]);
        let backup_path = format!("{main_path}.bkp");
        if object == main_path {
+            if self.main_missing {
+                return Err(Error::FileNotFound);
+            }
            return Ok(Self::reader(self.main.clone(), "main-etag"));
        }
        if object == backup_path {
@@ -712,6 +723,143 @@ async fn corrupt_primary_cache_loads_valid_backup() {
    assert_eq!(store.puts.load(Ordering::SeqCst), 0);
 }

+#[tokio::test]
+async fn cache_read_origin_reports_logical_slot_and_legacy_location() {
+    let mut primary = DataUsageCache::default();
+    primary.info.name = "primary".to_owned();
+    let mut backup = DataUsageCache::default();
+    backup.info.name = "backup".to_owned();
+    backup.info.snapshot_complete = true;
+    let primary_bytes = primary.marshal_msg().expect("serialize primary");
+    let backup_bytes = backup.marshal_msg().expect("serialize backup");
+    let empty_bytes = DataUsageCache::default().marshal_msg().expect("serialize empty cache");
+    let corrupt = b"not-msgpack".to_vec();
+
+    let cases = [
+        (
+            "primary",
+            Some(primary_bytes.clone()),
+            Some(backup_bytes.clone()),
+            None,
+            None,
+            DataUsageCacheReadOrigin::Primary { legacy: false },
+            primary_bytes.clone(),
+        ),
+        (
+            "legacy primary",
+            None,
+            None,
+            Some(primary_bytes.clone()),
+            None,
+            DataUsageCacheReadOrigin::Primary { legacy: true },
+            primary_bytes,
+        ),
+        (
+            "backup after missing",
+            None,
+            Some(backup_bytes.clone()),
+            None,
+            None,
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: false,
+            },
+            backup_bytes.clone(),
+        ),
+        (
+            "legacy backup after missing",
+            None,
+            None,
+            None,
+            Some(backup_bytes.clone()),
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Missing,
+                legacy: true,
+            },
+            backup_bytes.clone(),
+        ),
+        (
+            "backup after corrupt",
+            Some(corrupt.clone()),
+            Some(backup_bytes.clone()),
+            None,
+            None,
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: false,
+            },
+            backup_bytes.clone(),
+        ),
+        (
+            "legacy backup after corrupt",
+            Some(corrupt.clone()),
+            None,
+            None,
+            Some(backup_bytes.clone()),
+            DataUsageCacheReadOrigin::Backup {
+                primary: DataUsageCachePrimaryRead::Corrupt,
+                legacy: true,
+            },
+            backup_bytes,
+        ),
+        (
+            "missing objects",
+            None,
+            None,
+            None,
+            None,
+            DataUsageCacheReadOrigin::Empty,
+            empty_bytes.clone(),
+        ),
+        (
+            "corrupt primary without backup",
+            Some(corrupt),
+            None,
+            None,
+            None,
+            DataUsageCacheReadOrigin::Empty,
+            empty_bytes,
+        ),
+    ];
+
+    for (name, main, current_backup, legacy_primary, legacy_backup, expected_origin, expected_bytes) in cases {
+        let has_main = main.is_some();
+        let has_backup = current_backup.is_some();
+        let mut fixture = CacheReadStore::new(CacheReadBody::Bytes(main.unwrap_or_default()), current_backup);
+        fixture.main_missing = !has_main;
+        fixture.legacy = legacy_primary;
+        fixture.legacy_backup = legacy_backup;
+        let store = Arc::new(fixture);
+        let mut loaded = DataUsageCache::default();
+        let (revisions, origin) = loaded
+            .load_with_revisions_and_origin(store.clone(), DATA_USAGE_CACHE_NAME)
+            .await
+            .expect("fixture cache load");
+
+        assert_eq!(origin, expected_origin, "origin for {name}");
+        assert_eq!(loaded.marshal_msg().expect("re-encode cache"), expected_bytes, "cache body for {name}");
+        assert_eq!(
+            revisions.main,
+            if has_main {
+                DataUsageCacheRevision::Etag("main-etag".to_owned())
+            } else {
+                DataUsageCacheRevision::Missing
+            },
+            "primary CAS revision for {name}"
+        );
+        assert_eq!(
+            revisions.backup,
+            Some(if has_backup {
+                DataUsageCacheRevision::Etag("backup-etag".to_owned())
+            } else {
+                DataUsageCacheRevision::Missing
+            }),
+            "backup CAS revision for {name}"
+        );
+        assert_eq!(store.puts.load(Ordering::SeqCst), 0, "reading must not write for {name}");
+    }
+}
+
 #[tokio::test]
 async fn corrupt_cache_without_valid_backup_rebuilds_with_cas_revisions() {
    for backup in [None, Some(b"also-not-msgpack".to_vec())] {
@@ -7,17 +7,18 @@ use std::sync::atomic::AtomicU64;

 const PAYLOAD_BYTES: &str = "rustfs_scanner_cache_save_payload_bytes_total";
 const ENCODED_BYTES: &str = "rustfs_scanner_cache_serialized_bytes_total";
+const READ_ORIGIN: &str = "rustfs_scanner_cache_read_origin_total";

 #[derive(Default)]
-struct SaveRecorder(StdMutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
+struct CacheRecorder(StdMutex<HashMap<metrics::Key, Arc<AtomicU64>>>);

-impl metrics::Recorder for SaveRecorder {
+impl metrics::Recorder for CacheRecorder {
    fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
    fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
    fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}

    fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
-        if ![PAYLOAD_BYTES, ENCODED_BYTES].contains(&key.name()) {
+        if ![PAYLOAD_BYTES, ENCODED_BYTES, READ_ORIGIN].contains(&key.name()) {
            return metrics::Counter::noop();
        }
        metrics::Counter::from_arc(self.0.lock().expect("save recorder").entry(key.clone()).or_default().clone())
@@ -36,7 +37,7 @@ async fn check_save(store: BackupFallbackStore, revisions: Option<DataUsageCache
    let store = Arc::new(store);
    let cache = DataUsageCache::default();
    let encoded = rmp_serde::to_vec(&cache).expect("independent encoded body");
-    let recorder = SaveRecorder::default();
+    let recorder = CacheRecorder::default();
    let save = async {
        match revisions {
            Some(revisions) => {
@@ -85,6 +86,49 @@ async fn check_save(store: BackupFallbackStore, revisions: Option<DataUsageCache
    );
 }

+#[tokio::test(start_paused = true)]
+async fn read_origin_counts_successful_loads_with_fixed_labels() {
+    let bytes = DataUsageCache::default().marshal_msg().expect("serialize backup");
+    let corrupt_primary = CacheReadStore::new(CacheReadBody::Bytes(b"not-msgpack".to_vec()), Some(bytes.clone()));
+    let mut legacy_backup = CacheReadStore::new(CacheReadBody::Bytes(Vec::new()), None);
+    legacy_backup.main_missing = true;
+    legacy_backup.legacy_backup = Some(bytes);
+    let mut empty = CacheReadStore::new(CacheReadBody::Bytes(Vec::new()), None);
+    empty.main_missing = true;
+    let recorder = CacheRecorder::default();
+
+    for (store, with_revisions) in [(corrupt_primary, false), (legacy_backup, true), (empty, true)] {
+        let store = Arc::new(store);
+        let mut loaded = DataUsageCache::default();
+        let read = async {
+            if with_revisions {
+                loaded.load_with_revisions(store, DATA_USAGE_CACHE_NAME).await.map(|_| ())
+            } else {
+                loaded.load(store, DATA_USAGE_CACHE_NAME).await
+            }
+        };
+        let mut read = std::pin::pin!(read);
+        std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || read.as_mut().poll(cx)))
+            .await
+            .expect("fixture load");
+    }
+
+    let unavailable = Arc::new(BackupFallbackStore::new(Vec::new(), false));
+    let mut loaded = DataUsageCache::default();
+    let mut failed_read = std::pin::pin!(loaded.load_with_revisions(unavailable, DATA_USAGE_CACHE_NAME));
+    let error = std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || failed_read.as_mut().poll(cx)))
+        .await
+        .expect_err("an unreadable primary must not count as a fallback");
+    assert!(matches!(error, Error::ErasureReadQuorum));
+
+    let rows = recorder.0.lock().expect("cache recorder");
+    assert_eq!(rows.len(), 3, "one fixed origin label per successful load");
+    for origin in ["backup_primary_corrupt", "legacy_backup_primary_missing", "empty"] {
+        let key = metrics::Key::from_parts(READ_ORIGIN, vec![metrics::Label::new("origin", origin)]);
+        assert_eq!(rows[&key].load(Ordering::Relaxed), 1, "one observation for {origin}");
+    }
+}
+
 fn revisions(backup: bool) -> DataUsageCacheRevisions {
    DataUsageCacheRevisions {
        main: DataUsageCacheRevision::Missing,
@@ -313,7 +313,7 @@ Small objects store their payload inline after the container CRC ([filemeta_inli
 ## 8. Read path and read quorum

 - **INVARIANT — read quorum = `data_blocks`.** `object_quorum_from_meta` returns `(read_quorum = data_blocks, write_quorum)` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)); `parity_blocks = common_parity(...)` is the parity value held by the most disks that still reaches its own read quorum. When `default_parity_count == 0`, read = write = all shards.
- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. S3 reads select at read quorum through `select_valid_fileinfo`, the same rule heal uses and MinIO's `getObjectFileInfo`: the most common mod_time wins and a tie goes to the newer one, so a write that failed after reaching exactly `data_blocks` copies is served as listing shows it, instead of `ErasureReadQuorum` until heal runs. Two identities tied at the selected mod_time (a data = parity set split in half, such as a metadata update that failed after reaching half the drives) resolve to the one on the lowest-index disk, in reads and heal alike, so a retried metadata write can repair the split. A selected version with local shards also needs as many copies as its own `data_blocks`, since delete markers and empty versions can lower the common parity's read quorum. A metadata update whose target disks cannot reach write quorum is refused before any write. A version replaced by a committed write keeps fewer than `data_blocks` copies and can never be selected again; the newer side of a failed write can be served and then lose to the older side if one of its disks fails before heal converges the set. Reads in the system bucket (`is_meta_bucketname`) keep the commit-proof rule in `select_committed_fileinfo`: when enough copies answer to prove write quorum, the latest version must hold it, and a degraded read never serves a newer version below write quorum. Pool metadata's PREPARE/COMMIT recovery and identity epochs rely on an ordinary read implying a committed record, so a split stays unreadable until native recovery resolves it. When every drive answered and an S3 latest read still cannot select a version, `read_latest_past_dangling_versions` walks the version stack newest first through exact-version reads, as MinIO's `pickLatestQuorumFilesInfo` resolves stacks: it serves the readable version with the newest selected mod_time, skips versions whose exact read proves them dangling (the NoSuchVersion an explicit GET reports) and queues heal for them, and keeps the read-quorum error on any other outcome. A failed versioned PUT that left the new version on fewer drives than its data blocks therefore no longer hides the committed version below it; ListObjectVersions can still show the dangling version as latest until heal removes it.
+- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. S3 reads select at read quorum through `select_valid_fileinfo`, the same rule heal uses and MinIO's `getObjectFileInfo`: the most common mod_time wins and a tie goes to the newer one, so a write that failed after reaching exactly `data_blocks` copies is served as listing shows it, instead of `ErasureReadQuorum` until heal runs. Two identities tied at the selected mod_time (a data = parity set split in half, such as a metadata update that failed after reaching half the drives) resolve to the one on the lowest-index disk, in reads and heal alike, so a retried metadata write can repair the split. A selected version with local shards also needs as many copies as its own `data_blocks`, since delete markers and empty versions can lower the common parity's read quorum. A metadata update whose target disks cannot reach write quorum is refused before any write. A version replaced by a committed write keeps fewer than `data_blocks` copies and can never be selected again; the newer side of a failed write can be served and then lose to the older side if one of its disks fails before heal converges the set. Reads in the system bucket (`is_meta_bucketname`) keep the commit-proof rule in `select_committed_fileinfo`: when enough copies answer to prove write quorum, the latest version must hold it, and a degraded read never serves a newer version below write quorum. Pool metadata's PREPARE/COMMIT recovery and identity epochs rely on an ordinary read implying a committed record, so a split stays unreadable until native recovery resolves it. When every drive answered and an S3 latest read still cannot select a version, `read_latest_past_dangling_versions` walks the version stack newest first through exact-version reads, as MinIO's `pickLatestQuorumFilesInfo` resolves stacks: it serves the readable version with the newest selected mod_time, skips versions whose exact read proves them dangling (the NoSuchVersion an explicit GET reports) and queues heal for them, and keeps the read-quorum error on any other outcome. A failed versioned PUT that left the new version on fewer drives than its data blocks therefore no longer hides the committed version below it; ListObjectVersions can still show the dangling version as latest until heal removes it. Metadata updates of the latest version (object tagging, `put_object_metadata`) and the CopyObject source read resolve a failed selection through the same walk; when the walk finds no readable version, tagging keeps the read-quorum error while `put_object_metadata` and CopyObject keep their existing `delete_if_dangling` cleanup. Replication status write-back never walks. The update targets only the served version's disks and is refused before writing when they cannot reach its write quorum, including a version-only copy.
 - **INVARIANT — decode needs ≥ `data_blocks` shards.** The stripe reader requires `available_shards ≥ data_shards`; below that the read fails closed with a read-quorum error (never silent truncation) ([set_disk/read.rs](../../crates/ecstore/src/set_disk/read.rs), [set_disk/shard_source.rs](../../crates/ecstore/src/set_disk/shard_source.rs)). Codec geometry is validated at construction: `Erasure::try_new` / `try_new_with_options` ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)) return `ErasureConstructionError` for `data_shards == 0`, `block_size == 0`, shard-count overflow, or an unsupported shard configuration, so a read never reaches a `block_size` / `data_shards` division with invalid geometry (§13).
 - If `available ≥ data_blocks` but some shards are missing, the read is served **and** a background read-repair heal is enqueued.
 - **INVARIANT — cross-stripe read verification.** When a data shard is missing and `available > data_blocks`, reconstruction regenerates parity and compares it to the surviving parity; a mismatch is `InvalidData "inconsistent read source shards"` (backlog#832), catching corruption that passed per-shard bitrot but disagrees across the stripe ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)).
@@ -19,6 +19,18 @@ rustfs-security-governance = { path = "../crates/security-governance" }
 rustfs-utils = { path = "../crates/utils", features = ["compress", "path"] }
 serde_json = "1"

+[patch.crates-io]
+# Standalone workspaces must repeat the root workspace's fixed SigV4a backend.
+aws-sigv4 = { git = "https://github.com/smithy-lang/smithy-rs", rev = "5c1c4bca2ea7af97496e077c9f1a718eaac64bf8" }
+
+[patch."https://github.com/smithy-lang/smithy-rs"]
+# Keep the signer's path dependencies on the SDK's existing registry types.
+aws-credential-types = { version = "=1.3.1" }
+aws-smithy-eventstream = { version = "=0.61.6" }
+aws-smithy-http = { version = "=0.64.3" }
+aws-smithy-runtime-api = { version = "=1.19.2" }
+aws-smithy-types = { version = "=1.9.1" }
+
 [[bin]]
 name = "bucket_validation"
 path = "fuzz_targets/bucket_validation.rs"