chore: merge latest main into improve/base (#8332)

* refactor(storage): bind rename rollback to its commit context (#8324) * refactor(storage): separate metadata scheduling from quorum decisions (#8327) * test(storage): verify rejected rollback inspection on real disks (#8331) * test(storage): make commit fault checkpoints independent (#8326) * test(storage): make commit fault checkpoints independent * test(storage): simplify commit checkpoint lookup * fix(listing): count logical prefixes before pagination (#8328) * fix(listing): count logical prefixes before pagination * test(listing): normalize the fixed corpus seed --------- Co-authored-by: Chris <anzhengchao@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-13 09:17:41 +08:00 · 2026-10-04 02:15:19 +08:00
parent 06782925ff
commit 64cafb46f0
17 changed files with 3551 additions and 1558 deletions
@@ -111,46 +111,107 @@ async fn wait_for_heal_commit_test_barrier(root: &Path, bucket: &str, object: &s
    }
 }

-fn rollback_committed_rename_std(
-    dst_file_path: &Path,
-    new_data_path: Option<&Path>,
-    rollback_data_dir: Option<Uuid>,
-) -> std::io::Result<()> {
-    if let Some(old_data_dir) = rollback_data_dir {
-        let Some(dst_parent) = dst_file_path.parent() else {
-            return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing object metadata parent"));
-        };
-        let backup_path = dst_parent.join(old_data_dir.to_string()).join(STORAGE_FORMAT_FILE_BACKUP);
-        std::fs::rename(backup_path, dst_file_path)?;
-    } else {
-        remove_file_if_exists(dst_file_path)?;
-    }
-
-    if let Some(new_data_path) = new_data_path {
-        remove_dir_all_if_exists(new_data_path)?;
-    }
-
-    Ok(())
+struct RenameCommitContext<'a> {
+    // Drop the directory identity guard before releasing mutation ownership.
+    directory_guard: os::RenameCommitGuard,
+    publication_root: &'a os::PublicationRoot,
+    mutation_lease: Arc<os::NamespaceMutationLease>,
+    source_metadata: &'a Path,
+    destination_metadata: &'a Path,
+    source_volume_dir: &'a Path,
+    data_paths: Option<&'a (PathBuf, PathBuf)>,
 }

-fn rollback_inline_metadata_commit_std(
-    dst_file_path: &Path,
-    rollback_data_dir: Option<Uuid>,
-    local_rollback_path: Option<&Path>,
-) -> std::io::Result<()> {
-    #[cfg(all(test, not(windows)))]
-    os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Rollback, dst_file_path);
-    if let Some(backup_path) = local_rollback_path {
-        // The commit immediately before this rollback renamed the staged
-        // xl.meta from the same directory as `backup_path` onto
-        // `dst_file_path`, proving both paths are on the same filesystem.
-        // Unix rename atomically replaces the committed destination; never
-        // unlink it first or an interrupted rollback could lose xl.meta.
-        std::fs::rename(backup_path, dst_file_path)?;
-    } else {
-        rollback_committed_rename_std(dst_file_path, None, rollback_data_dir)?;
+struct StagedRollbackBackup(PathBuf);
+
+impl StagedRollbackBackup {
+    fn path(&self) -> &Path {
+        &self.0
+    }
+}
+
+impl RenameCommitContext<'_> {
+    fn rollback_metadata_std(&self, rollback_data_dir: Option<Uuid>) -> std::io::Result<()> {
+        if let Some(old_data_dir) = rollback_data_dir {
+            let Some(dst_parent) = self.destination_metadata.parent() else {
+                return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing object metadata parent"));
+            };
+            let backup_path = dst_parent.join(old_data_dir.to_string()).join(STORAGE_FORMAT_FILE_BACKUP);
+            std::fs::rename(backup_path, self.destination_metadata)?;
+        } else {
+            remove_file_if_exists(self.destination_metadata)?;
+        }
+        Ok(())
+    }
+
+    fn rollback_non_inline_std(&self, rollback_data_dir: Option<Uuid>) -> std::io::Result<()> {
+        self.rollback_metadata_std(rollback_data_dir)?;
+        if let Some((_, new_data_path)) = self.data_paths {
+            remove_dir_all_if_exists(new_data_path)?;
+        }
+        Ok(())
+    }
+
+    fn rollback_inline_std(
+        &self,
+        rollback_data_dir: Option<Uuid>,
+        local_backup: Option<&StagedRollbackBackup>,
+    ) -> std::io::Result<()> {
+        #[cfg(all(test, not(windows)))]
+        os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Rollback, self.destination_metadata);
+        if let Some(backup) = local_backup {
+            // The preceding commit used this staged directory on the same
+            // filesystem. Never unlink xl.meta before its atomic replacement.
+            std::fs::rename(backup.path(), self.destination_metadata)?;
+        } else {
+            self.rollback_metadata_std(rollback_data_dir)?;
+        }
+        Ok(())
+    }
+
+    async fn restore_published_data_source(&self) -> Result<()> {
+        let Some((src_data_path, dst_data_path)) = self.data_paths else {
+            return Ok(());
+        };
+        let src_volume_dir = self.source_volume_dir;
+        let publication_root = self.publication_root;
+        let mutation_lease = self.mutation_lease.clone();
+        if fs::symlink_metadata(src_data_path).await.is_ok() {
+            return Ok(());
+        }
+        let result =
+            match os::rename_all_with_lease(dst_data_path, src_data_path, src_volume_dir, publication_root, mutation_lease).await
+            {
+                Ok(()) => Ok(()),
+                Err(DiskError::FileNotFound) => {
+                    let source_exists = fs::symlink_metadata(src_data_path).await.is_ok();
+                    let destination_missing = matches!(
+                        fs::symlink_metadata(dst_data_path).await,
+                        Err(err) if err.kind() == ErrorKind::NotFound
+                    );
+                    if source_exists && destination_missing {
+                        Ok(())
+                    } else {
+                        Err(DiskError::FileNotFound)
+                    }
+                }
+                Err(err) => Err(err),
+            };
+        if let Err(err) = &result {
+            warn!(
+                target: "rustfs_ecstore::disk::local",
+                event = EVENT_DISK_LOCAL_RENAME_REJECTED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
+                reason = "restore_staged_data_source_failed",
+                src_path = ?src_data_path,
+                dst_path = ?dst_data_path,
+                error = ?err,
+                "Failed to restore staged data after a metadata commit was rejected"
+            );
+        }
+        result
    }
-    Ok(())
 }

 #[cfg(any(not(windows), test))]
@@ -249,61 +310,6 @@ async fn prepare_rename_destination_metadata(
    .map_err(DiskError::from)
 }

-async fn restore_renamed_data_source(
-    src_volume_dir: &Path,
-    src_data_path: &Path,
-    dst_data_path: &Path,
-    publication_root: &os::PublicationRoot,
-    mutation_lease: Arc<os::NamespaceMutationLease>,
-) -> Result<()> {
-    if fs::symlink_metadata(src_data_path).await.is_ok() {
-        return Ok(());
-    }
-    let result =
-        match os::rename_all_with_lease(dst_data_path, src_data_path, src_volume_dir, publication_root, mutation_lease).await {
-            Ok(()) => Ok(()),
-            Err(DiskError::FileNotFound) => {
-                let source_exists = fs::symlink_metadata(src_data_path).await.is_ok();
-                let destination_missing = matches!(
-                    fs::symlink_metadata(dst_data_path).await,
-                    Err(err) if err.kind() == ErrorKind::NotFound
-                );
-                if source_exists && destination_missing {
-                    Ok(())
-                } else {
-                    Err(DiskError::FileNotFound)
-                }
-            }
-            Err(err) => Err(err),
-        };
-    if let Err(err) = &result {
-        warn!(
-            target: "rustfs_ecstore::disk::local",
-            event = EVENT_DISK_LOCAL_RENAME_REJECTED,
-            component = LOG_COMPONENT_ECSTORE,
-            subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
-            reason = "restore_staged_data_source_failed",
-            src_path = ?src_data_path,
-            dst_path = ?dst_data_path,
-            error = ?err,
-            "Failed to restore staged data after a metadata commit was rejected"
-        );
-    }
-    result
-}
-
-async fn restore_published_data_source(
-    data_paths: Option<&(PathBuf, PathBuf)>,
-    src_volume_dir: &Path,
-    publication_root: &os::PublicationRoot,
-    mutation_lease: Arc<os::NamespaceMutationLease>,
-) -> Result<()> {
-    let Some((src_data_path, dst_data_path)) = data_paths else {
-        return Ok(());
-    };
-    restore_renamed_data_source(src_volume_dir, src_data_path, dst_data_path, publication_root, mutation_lease).await
-}
-
 /// Proof produced only when the local rename returns at an existing access
 /// preflight, before metadata, backups, or object data can be published.
 #[derive(Debug)]
@@ -493,6 +499,15 @@ impl LocalDisk {
            mutation_lease.clone(),
        )
        .await?;
+        let mut commit = RenameCommitContext {
+            directory_guard: rename_commit_guard,
+            publication_root: &self.publication_root,
+            mutation_lease,
+            source_metadata: &src_file_path,
+            destination_metadata: &dst_file_path,
+            source_volume_dir: &src_volume_dir,
+            data_paths: has_data_dir_path.as_ref(),
+        };

        if no_inline {
            // Non-inline: read xl.meta, parse, write, rename data dir, rename xl.meta
@@ -576,18 +591,18 @@ impl LocalDisk {
            // rename below to report through the existing rollback path. Payload
            // durability is kept by both strict and relaxed.
            let tmp_meta_write = {
-                let src_file_path = src_file_path.clone();
+                let src_file_path = commit.source_metadata.to_path_buf();
                let dst_file_path = dst_file_path.clone();
-                let rename_commit_guard = rename_commit_guard.clone();
-                let mutation_lease = mutation_lease.clone();
+                let owned_directory_guard = commit.directory_guard.clone();
+                let owned_mutation_lease = commit.mutation_lease.clone();
                async move {
-                    os::run_blocking_namespace_operation(mutation_lease, move || {
+                    os::run_blocking_namespace_operation(owned_mutation_lease, move || {
                        #[cfg(test)]
                        run_owned_file_write_before_open(&src_file_path);
                        let mut prepared_metadata_source = os::create_prepared_rename_source_with_commit_guard(
                            &src_file_path,
                            &dst_file_path,
-                            &rename_commit_guard,
+                            &owned_directory_guard,
                        )?;
                        prepared_metadata_source.write_all(&new_dst_buf, tmp_meta_sync != SyncMode::None)?;
                        Ok(prepared_metadata_source)
@@ -613,14 +628,14 @@ impl LocalDisk {
            // sequential version did.
            let prepared_metadata_source = tmp_meta_res?;
            shard_sync_res?;
-            let rename_commit_guard = remove_dst_base_before_commit(
+            commit.directory_guard = remove_dst_base_before_commit(
                dst_path,
-                rename_commit_guard,
+                commit.directory_guard,
                src_file_parent,
                dst_file_parent,
                &dst_volume_dir,
-                &self.publication_root,
-                mutation_lease.clone(),
+                commit.publication_root,
+                commit.mutation_lease.clone(),
            )
            .await?;
            if should_remove_staged_meta_before_commit(dst_path) {
@@ -638,7 +653,7 @@ impl LocalDisk {
            if fi_healing
                && let Some((_, dst_data_path)) = has_data_dir_path.as_ref()
                && let Err(err) = self
-                    .move_to_trash_with_namespace_owner(dst_data_path, true, false, Some(mutation_lease.clone()))
+                    .move_to_trash_with_namespace_owner(dst_data_path, true, false, Some(commit.mutation_lease.clone()))
                    .await
            {
                warn!(
@@ -656,9 +671,9 @@ impl LocalDisk {
                    src_data_path,
                    dst_data_path,
                    &skip_parent,
-                    &self.publication_root,
-                    &rename_commit_guard,
-                    mutation_lease.clone(),
+                    commit.publication_root,
+                    &commit.directory_guard,
+                    commit.mutation_lease.clone(),
                )
                .await
            {
@@ -673,13 +688,7 @@ impl LocalDisk {
                    error = ?err,
                    "Disk local rename flow failed"
                );
-                restore_published_data_source(
-                    has_data_dir_path.as_ref(),
-                    &src_volume_dir,
-                    &self.publication_root,
-                    mutation_lease.clone(),
-                )
-                .await?;
+                commit.restore_published_data_source().await?;
                return Err(err);
            }
            #[cfg(test)]
@@ -704,13 +713,7 @@ impl LocalDisk {
                    reason = "test_fail_before_old_metadata_backup",
                    "Disk local rename flow failed before metadata commit"
                );
-                restore_published_data_source(
-                    has_data_dir_path.as_ref(),
-                    &src_volume_dir,
-                    &self.publication_root,
-                    mutation_lease.clone(),
-                )
-                .await?;
+                commit.restore_published_data_source().await?;
                return Err(DiskError::Unexpected);
            }

@@ -728,25 +731,16 @@ impl LocalDisk {
                let backup_parent = dst_file_parent.join(old_data_dir.to_string());
                #[cfg(not(windows))]
                if let Err(err) = os::make_dir_all(&backup_parent, &skip_parent).await {
-                    restore_published_data_source(
-                        has_data_dir_path.as_ref(),
-                        &src_volume_dir,
-                        &self.publication_root,
-                        mutation_lease.clone(),
-                    )
-                    .await?;
+                    commit.restore_published_data_source().await?;
                    return Err(err);
                }
-                let backup_path_guard = match rename_commit_guard.create_destination_directory_for_path_access(&backup_parent) {
+                let backup_path_guard = match commit
+                    .directory_guard
+                    .create_destination_directory_for_path_access(&backup_parent)
+                {
                    Ok(guard) => guard,
                    Err(err) => {
-                        restore_published_data_source(
-                            has_data_dir_path.as_ref(),
-                            &src_volume_dir,
-                            &self.publication_root,
-                            mutation_lease.clone(),
-                        )
-                        .await?;
+                        commit.restore_published_data_source().await?;
                        return Err(DiskError::from(to_file_error(err)));
                    }
                };
@@ -754,13 +748,7 @@ impl LocalDisk {
                if let Err(err) = check_path_length(backup_path.to_string_lossy().as_ref()) {
                    #[cfg(windows)]
                    drop(backup_path_guard);
-                    restore_published_data_source(
-                        has_data_dir_path.as_ref(),
-                        &src_volume_dir,
-                        &self.publication_root,
-                        mutation_lease.clone(),
-                    )
-                    .await?;
+                    commit.restore_published_data_source().await?;
                    return Err(err);
                }
                let backup_bytes = dst_buf.clone();
@@ -769,7 +757,7 @@ impl LocalDisk {
                // sync finish. A detached spawn_blocking writer could survive
                // cancellation and later truncate a newer transaction's
                // deterministic rollback backup.
-                let write_result = os::run_blocking_namespace_operation(mutation_lease.clone(), move || {
+                let write_result = os::run_blocking_namespace_operation(commit.mutation_lease.clone(), move || {
                    #[cfg(test)]
                    run_owned_file_write_before_open(&backup_path);
                    backup_path_guard.write_file_for_path_access(
@@ -792,13 +780,7 @@ impl LocalDisk {
                        error = ?err,
                        "Disk local rename flow failed"
                    );
-                    restore_published_data_source(
-                        has_data_dir_path.as_ref(),
-                        &src_volume_dir,
-                        &self.publication_root,
-                        mutation_lease.clone(),
-                    )
-                    .await?;
+                    commit.restore_published_data_source().await?;
                    return Err(err);
                }
            }
@@ -816,9 +798,9 @@ impl LocalDisk {
                &src_file_path,
                &dst_file_path,
                &skip_parent,
-                &self.publication_root,
-                &rename_commit_guard,
-                mutation_lease.clone(),
+                commit.publication_root,
+                &commit.directory_guard,
+                commit.mutation_lease.clone(),
            )
            .await
            {
@@ -833,20 +815,11 @@ impl LocalDisk {
                    error = ?err,
                    "Disk local rename flow failed"
                );
-                restore_published_data_source(
-                    has_data_dir_path.as_ref(),
-                    &src_volume_dir,
-                    &self.publication_root,
-                    mutation_lease.clone(),
-                )
-                .await?;
+                commit.restore_published_data_source().await?;
                return Err(err);
            }
-
-            let committed_new_data_path = has_data_dir_path.as_ref().map(|(_, dst_data_path)| dst_data_path.as_path());
            if should_fail_after_metadata_commit(dst_path) {
-                rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir)
-                    .map_err(to_file_error)?;
+                commit.rollback_non_inline_std(rollback_data_dir).map_err(to_file_error)?;
                return Err(DiskError::Unexpected);
            }

@@ -866,13 +839,14 @@ impl LocalDisk {
                && let Some(parent) = dst_file_path.parent()
            {
                let fsync_started = rustfs_io_metrics::put_stage_timer();
-                if let Err(err) = os::fsync_commit_directory(parent, &rename_commit_guard, mutation_lease.clone(), None).await {
+                if let Err(err) =
+                    os::fsync_commit_directory(parent, &commit.directory_guard, commit.mutation_lease.clone(), None).await
+                {
                    rustfs_io_metrics::record_put_object_stage_duration_from(
                        rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC,
                        fsync_started,
                    );
-                    rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir)
-                        .map_err(to_file_error)?;
+                    commit.rollback_non_inline_std(rollback_data_dir).map_err(to_file_error)?;
                    // The commit rename changed the dst part inodes before this fsync
                    // failed and rolled them back; drop any fd cached during that
                    // window so readers re-open the restored inode (rustfs/backlog#1177).
@@ -904,13 +878,12 @@ impl LocalDisk {
                        break;
                    }
                    let fsync_started = rustfs_io_metrics::put_stage_timer();
-                    if let Err(err) = os::fsync_dir_with_owner(dir, Some(mutation_lease.clone())).await {
+                    if let Err(err) = os::fsync_dir_with_owner(dir, Some(commit.mutation_lease.clone())).await {
                        rustfs_io_metrics::record_put_object_stage_duration_from(
                            rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC,
                            fsync_started,
                        );
-                        rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir)
-                            .map_err(to_file_error)?;
+                        commit.rollback_non_inline_std(rollback_data_dir).map_err(to_file_error)?;
                        // Same post-commit rollback window as above — drop cached
                        // dst part fds so readers re-open the restored inode
                        // (rustfs/backlog#1177).
@@ -934,7 +907,7 @@ impl LocalDisk {
            // complete. Do not retain the Windows object identity guard while
            // cleaning staging paths or invalidating cached descriptors.
            #[cfg(windows)]
-            drop(rename_commit_guard);
+            drop(commit.directory_guard);

            if let Some(src_file_path_parent) = src_file_path.parent() {
                if src_volume != super::super::RUSTFS_META_MULTIPART_BUCKET {
@@ -986,7 +959,7 @@ impl LocalDisk {
            let dst_path_for_source = dst_path.to_string();
            #[cfg(windows)]
            let source_parent = src_file_parent.to_path_buf();
-            let rename_commit_guard_for_preparation = rename_commit_guard.clone();
+            let rename_commit_guard_for_preparation = commit.directory_guard.clone();
            let sync = durability.syncs_commit_metadata();
            #[cfg(test)]
            run_inline_before_file_sync_admission(dst_path);
@@ -1077,7 +1050,7 @@ impl LocalDisk {
                    if sync {
                        std::fs::File::open(&backup_path)?.sync_data()?;
                    }
-                    staged_rollback_path = Some(backup_path);
+                    staged_rollback_path = Some(StagedRollbackBackup(backup_path));
                }

                Ok::<_, std::io::Error>((
@@ -1093,9 +1066,10 @@ impl LocalDisk {
                ))
            };
            let inline_preparation = if let Some(admission) = file_sync_admission.as_ref() {
-                os::run_blocking_namespace_file_sync_operation(mutation_lease.clone(), admission, prepare_inline_metadata).await
+                os::run_blocking_namespace_file_sync_operation(commit.mutation_lease.clone(), admission, prepare_inline_metadata)
+                    .await
            } else {
-                os::run_blocking_namespace_operation(mutation_lease.clone(), prepare_inline_metadata).await
+                os::run_blocking_namespace_operation(commit.mutation_lease.clone(), prepare_inline_metadata).await
            }
            .map_err(to_file_error)
            .map_err(DiskError::from);
@@ -1120,28 +1094,30 @@ impl LocalDisk {
                }
            };

-            let rename_commit_guard = remove_dst_base_before_commit(
+            commit.directory_guard = remove_dst_base_before_commit(
                dst_path,
-                rename_commit_guard,
+                commit.directory_guard,
                src_file_parent,
                dst_file_parent,
                &dst_volume_dir,
-                &self.publication_root,
-                mutation_lease.clone(),
+                commit.publication_root,
+                commit.mutation_lease.clone(),
            )
            .await?;

            if should_remove_staged_meta_before_commit(dst_path) {
                drop(prepared_metadata_source);
                let remove_result = std::fs::remove_file(&src_file_path);
-                if let Some(backup_path) = local_rollback_path.as_deref() {
+                if let Some(backup_path) = local_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                    let _ = remove_file_if_exists(backup_path);
                }
                remove_result.map_err(to_file_error)?;
                return Err(DiskError::FileNotFound);
            }

-            if let (Some(rollback_data_dir), Some(staged_backup)) = (rollback_data_dir, local_rollback_path.as_deref()) {
+            if let (Some(rollback_data_dir), Some(staged_backup)) =
+                (rollback_data_dir, local_rollback_path.as_ref().map(StagedRollbackBackup::path))
+            {
                let Some(dst_parent) = dst_file_path.parent() else {
                    return Err(DiskError::other("missing object metadata parent"));
                };
@@ -1155,8 +1131,8 @@ impl LocalDisk {
                    staged_backup,
                    &backup_path,
                    &dst_volume_dir,
-                    &self.publication_root,
-                    Some(mutation_lease.clone()),
+                    commit.publication_root,
+                    Some(commit.mutation_lease.clone()),
                )
                .await
                {
@@ -1178,7 +1154,8 @@ impl LocalDisk {
                {
                    let fsync_started = rustfs_io_metrics::put_stage_timer();
                    if let Err(err) =
-                        os::fsync_dir_with_namespace_file_sync_limit(backup_parent, mutation_lease.clone(), admission).await
+                        os::fsync_dir_with_namespace_file_sync_limit(backup_parent, commit.mutation_lease.clone(), admission)
+                            .await
                    {
                        rustfs_io_metrics::record_put_object_stage_duration_from(
                            rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC,
@@ -1202,14 +1179,14 @@ impl LocalDisk {
                    &src_file_path,
                    &dst_file_path,
                    &dst_volume_dir,
-                    &self.publication_root,
-                    &rename_commit_guard,
-                    mutation_lease.clone(),
+                    commit.publication_root,
+                    &commit.directory_guard,
+                    commit.mutation_lease.clone(),
                )
                .await
            };
            if let Err(err) = commit_result {
-                if let Some(backup_path) = local_rollback_path.as_deref() {
+                if let Some(backup_path) = local_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                    let _ = remove_file_if_exists(backup_path);
                }
                for part_path in &invalidate_part_paths {
@@ -1220,7 +1197,7 @@ impl LocalDisk {

            let post_commit = async {
                if should_fail_after_metadata_commit(dst_path) {
-                    rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?;
+                    commit.rollback_inline_std(rollback_data_dir, local_rollback_path.as_ref())?;
                    return Err(std::io::Error::other("test fail after metadata commit"));
                }

@@ -1229,15 +1206,19 @@ impl LocalDisk {
                    && let Some(dst_parent) = dst_file_path.parent()
                {
                    let fsync_started = rustfs_io_metrics::put_stage_timer();
-                    if let Err(err) =
-                        os::fsync_commit_directory(dst_parent, &rename_commit_guard, mutation_lease.clone(), Some(admission))
-                            .await
+                    if let Err(err) = os::fsync_commit_directory(
+                        dst_parent,
+                        &commit.directory_guard,
+                        commit.mutation_lease.clone(),
+                        Some(admission),
+                    )
+                    .await
                    {
                        rustfs_io_metrics::record_put_object_stage_duration_from(
                            rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC,
                            fsync_started,
                        );
-                        rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?;
+                        commit.rollback_inline_std(rollback_data_dir, local_rollback_path.as_ref())?;
                        return Err(err);
                    }
                    rustfs_io_metrics::record_put_object_stage_duration_from(
@@ -1263,17 +1244,14 @@ impl LocalDisk {
                        }
                        let fsync_started = rustfs_io_metrics::put_stage_timer();
                        if let Err(err) =
-                            os::fsync_dir_with_namespace_file_sync_limit(ancestor_dir, mutation_lease.clone(), admission).await
+                            os::fsync_dir_with_namespace_file_sync_limit(ancestor_dir, commit.mutation_lease.clone(), admission)
+                                .await
                        {
                            rustfs_io_metrics::record_put_object_stage_duration_from(
                                rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC,
                                fsync_started,
                            );
-                            rollback_inline_metadata_commit_std(
-                                &dst_file_path,
-                                rollback_data_dir,
-                                local_rollback_path.as_deref(),
-                            )?;
+                            commit.rollback_inline_std(rollback_data_dir, local_rollback_path.as_ref())?;
                            return Err(err);
                        }
                        rustfs_io_metrics::record_put_object_stage_duration_from(
@@ -1310,9 +1288,9 @@ impl LocalDisk {
            // The commit no longer has a rollback path. Release the Windows
            // object identity guard before best-effort staging cleanup.
            #[cfg(windows)]
-            drop(rename_commit_guard);
+            drop(commit.directory_guard);

-            if let Some(backup_path) = local_rollback_path.as_deref() {
+            if let Some(backup_path) = local_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                let _ = remove_file_if_exists(backup_path);
            }

@@ -29,6 +29,27 @@ use crate::disk::error_reduce::OBJECT_OP_IGNORED_ERRS;
 use crate::set_disk::file_info_is_valid_for_metadata;
 use rustfs_filemeta::FileInfo;

+/// One disk's observation. Pending is neither a successful vote nor an offline
+/// disk: the scheduler may not have issued this slot or may still be waiting.
+// Option<Result<...>> keeps FileInfo inline without a per-response Box.
+// None is pending; Some(Ok(_)) is success; Some(Err(_)) retains a disk error.
+pub(in crate::set_disk) type MetadataDiskResult = Option<crate::disk::error::Result<FileInfo>>;
+
+#[derive(Debug)]
+pub(in crate::set_disk) struct MetadataDiskObservation {
+    pub(in crate::set_disk) disk_index: usize,
+    pub(in crate::set_disk) result: MetadataDiskResult,
+}
+
+impl MetadataDiskObservation {
+    pub(in crate::set_disk) fn file_info(&self) -> Option<&FileInfo> {
+        match &self.result {
+            Some(Ok(metadata)) => Some(metadata),
+            None | Some(Err(_)) => None,
+        }
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(in crate::set_disk) struct MetadataEarlyStopDecision {
    pub(in crate::set_disk) reason: &'static str,
@@ -59,6 +80,14 @@ pub(in crate::set_disk) struct MetadataQuorumAccumulator {
 }

 impl MetadataQuorumAccumulator {
+    pub(in crate::set_disk) fn observe(&mut self, observation: &MetadataDiskObservation) {
+        match &observation.result {
+            None => {}
+            Some(Ok(metadata)) => self.observe_file_info_at(observation.disk_index, metadata),
+            Some(Err(error)) => self.observe_error(error),
+        }
+    }
+
    pub(in crate::set_disk) fn new(total_disks: usize, default_parity_count: usize, allow_early_stop: bool) -> Self {
        Self {
            total_disks,
@@ -86,6 +115,7 @@ impl MetadataQuorumAccumulator {
        self
    }

+    #[cfg(test)]
    pub(in crate::set_disk) fn observe_file_info(&mut self, file_info: &FileInfo) {
        self.observe_file_info_with_index(None, file_info);
    }
@@ -383,3 +413,186 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
 pub(in crate::set_disk) fn is_metadata_fanout_ignored_error(err: &DiskError) -> bool {
    OBJECT_OP_IGNORED_ERRS.iter().any(|ignored| ignored == err)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::set_disk::SetDisks;
+    use time::OffsetDateTime;
+    use uuid::Uuid;
+
+    fn payload(disk_index: usize) -> FileInfo {
+        let mut metadata = FileInfo::new("object", 2, 2);
+        metadata.volume = "bucket".to_string();
+        metadata.name = "object".to_string();
+        metadata.size = 1;
+        metadata.mod_time = Some(OffsetDateTime::UNIX_EPOCH);
+        metadata.data_dir = Some(Uuid::from_u128(1));
+        metadata.erasure.index = metadata.erasure.distribution[disk_index];
+        metadata.metadata.insert("etag".to_string(), "etag".to_string());
+        metadata.add_object_part(1, "etag".to_string(), 1, None, 1, None, None);
+        assert!(file_info_is_valid_for_metadata(&metadata), "the observation corpus needs a valid payload");
+        metadata
+    }
+
+    fn observation(disk_index: usize, state: usize) -> MetadataDiskObservation {
+        MetadataDiskObservation {
+            disk_index,
+            result: match state {
+                0 => Some(Ok(payload(disk_index))),
+                1 => Some(Err(DiskError::FileNotFound)),
+                2 => Some(Err(DiskError::FileCorrupt)),
+                3 => Some(Err(DiskError::DiskNotFound)),
+                4 => None,
+                _ => panic!("unexpected test state"),
+            },
+        }
+    }
+
+    fn assert_same_reduction(typed: &MetadataQuorumAccumulator, legacy: &MetadataQuorumAccumulator) {
+        assert_eq!(typed.early_stop_decision(), legacy.early_stop_decision());
+        assert_eq!(typed.version_early_stop_decision(), legacy.version_early_stop_decision());
+        assert_eq!(typed.final_miss_reason(), legacy.final_miss_reason());
+        assert_eq!(typed.valid_responses, legacy.valid_responses);
+        assert_eq!(typed.not_found_responses, legacy.not_found_responses);
+        assert_eq!(typed.version_not_found_responses, legacy.version_not_found_responses);
+        assert_eq!(typed.ignored_errors, legacy.ignored_errors);
+        assert_eq!(typed.hard_errors, legacy.hard_errors);
+        assert_eq!(typed.candidate_votes, legacy.candidate_votes);
+        assert_eq!(typed.candidate_shard_mask, legacy.candidate_shard_mask);
+        assert_eq!(typed.matching_version_votes, legacy.matching_version_votes);
+        assert_eq!(typed.delete_marker_votes, legacy.delete_marker_votes);
+        assert_eq!(typed.conflicting_metadata, legacy.conflicting_metadata);
+        assert_eq!(
+            typed.candidate.as_ref().map(SetDisks::file_info_quorum_hash),
+            legacy.candidate.as_ref().map(SetDisks::file_info_quorum_hash)
+        );
+    }
+
+    fn observe_legacy(accumulator: &mut MetadataQuorumAccumulator, observation: &MetadataDiskObservation) {
+        match &observation.result {
+            None => {}
+            Some(Ok(metadata)) => accumulator.observe_file_info_at(observation.disk_index, metadata),
+            Some(Err(error)) => accumulator.observe_error(error),
+        }
+    }
+
+    #[test]
+    fn metadata_observation_all_four_slot_states_and_arrival_orders_preserve_reduction() {
+        let permutations = (0..4)
+            .flat_map(|a| (0..4).filter(move |b| *b != a).map(move |b| (a, b)))
+            .flat_map(|(a, b)| (0..4).filter(move |c| *c != a && *c != b).map(move |c| (a, b, c)))
+            .map(|(a, b, c)| [a, b, c, 6 - a - b - c])
+            .collect::<Vec<_>>();
+        assert_eq!(permutations.len(), 24);
+        // 5^4 states and all 4! arrivals, both with early-stop enabled and disabled.
+        for encoded in 0..625usize {
+            let states = [encoded % 5, encoded / 5 % 5, encoded / 25 % 5, encoded / 125 % 5];
+            let inputs = std::array::from_fn::<_, 4, _>(|index| observation(index, states[index]));
+            for enabled in [false, true] {
+                for order in &permutations {
+                    let mut typed = MetadataQuorumAccumulator::new(4, 2, enabled);
+                    let mut legacy = MetadataQuorumAccumulator::new(4, 2, enabled);
+                    for &index in order {
+                        typed.observe(&inputs[index]);
+                        observe_legacy(&mut legacy, &inputs[index]);
+                        assert_same_reduction(&typed, &legacy);
+                    }
+                    assert_eq!(typed.valid_responses, states.iter().filter(|&&state| state == 0).count());
+                    assert_eq!(typed.not_found_responses, states.iter().filter(|&&state| state == 1).count());
+                    assert_eq!(typed.hard_errors, states.iter().filter(|&&state| state == 2).count());
+                    assert_eq!(typed.ignored_errors, states.iter().filter(|&&state| state == 3).count());
+                    let expected_early_stop =
+                        enabled && typed.valid_responses >= 3 && !states.iter().any(|&state| state == 1 || state == 2);
+                    assert_eq!(
+                        typed.early_stop_decision().is_some(),
+                        expected_early_stop,
+                        "states={states:?}, order={order:?}"
+                    );
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn metadata_observation_pending_newer_version_and_same_time_different_directory_force_full_wait() {
+        let mut accumulator = MetadataQuorumAccumulator::new(4, 2, true);
+        for index in 0..2 {
+            accumulator.observe(&observation(index, 0));
+        }
+        accumulator.observe(&observation(2, 4));
+        assert_eq!(accumulator.candidate_votes, 2, "an outstanding response cannot supply the deciding vote");
+        assert_eq!(accumulator.early_stop_decision(), None);
+
+        let mut newer = payload(2);
+        newer.mod_time = Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1));
+        accumulator.observe(&MetadataDiskObservation {
+            disk_index: 2,
+            result: Some(Ok(newer)),
+        });
+        assert!(accumulator.conflicting_metadata);
+        assert_eq!(accumulator.early_stop_decision(), None);
+
+        let mut same_time_different_directory = payload(3);
+        same_time_different_directory.data_dir = Some(Uuid::from_u128(2));
+        accumulator.observe(&MetadataDiskObservation {
+            disk_index: 3,
+            result: Some(Ok(same_time_different_directory)),
+        });
+        assert_eq!(accumulator.early_stop_decision(), None);
+    }
+
+    #[test]
+    fn metadata_observation_duplicate_shard_cannot_supply_a_read_reserve() {
+        let mut accumulator = MetadataQuorumAccumulator::new(4, 2, true);
+        let first = payload(0);
+        for disk_index in 0..3 {
+            accumulator.observe(&MetadataDiskObservation {
+                disk_index,
+                result: Some(Ok(first.clone())),
+            });
+        }
+        assert_eq!(accumulator.candidate_votes, 3);
+        assert!(
+            !accumulator.candidate_has_read_reserve(),
+            "copied erasure indexes cannot supply independent data shards"
+        );
+    }
+
+    #[test]
+    fn metadata_observation_null_versions_markers_and_invalid_success_keep_their_meaning() {
+        let mut null_version = MetadataQuorumAccumulator::new(4, 2, true);
+        for index in 0..3 {
+            null_version.observe(&observation(index, 0));
+        }
+        assert!(null_version.early_stop_decision().is_some());
+        assert_eq!(null_version.matching_version_votes, 0);
+
+        let mut marker = MetadataQuorumAccumulator::new(4, 2, true);
+        for disk_index in 0..3 {
+            let metadata = FileInfo {
+                volume: "bucket".to_string(),
+                name: "object".to_string(),
+                deleted: true,
+                mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)),
+                ..Default::default()
+            };
+            assert!(metadata.is_canonical_delete_marker(), "the null marker fixture must reach marker voting");
+            marker.observe(&MetadataDiskObservation {
+                disk_index,
+                result: Some(Ok(metadata)),
+            });
+        }
+        assert_eq!(marker.delete_marker_votes, 3);
+        assert!(marker.early_stop_decision().is_some());
+
+        let mut invalid = MetadataQuorumAccumulator::new(4, 2, true);
+        invalid.observe(&MetadataDiskObservation {
+            disk_index: 0,
+            result: Some(Ok(FileInfo::default())),
+        });
+        assert_eq!(invalid.valid_responses, 0);
+        assert_eq!(invalid.hard_errors, 1);
+        assert_eq!(invalid.early_stop_decision(), None);
+    }
+}
@@ -18,4 +18,5 @@
 //! duplicating read/write/erasure logic.

 pub(crate) mod io_primitives;
-mod metadata_quorum;
+pub(in crate::set_disk) mod metadata_quorum;
+pub(in crate::set_disk) mod metadata_read;
@@ -5886,12 +5886,29 @@ fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
    parts_metadata: &'a [FileInfo],
    fi: &FileInfo,
    data_shards: usize,
+    disk_is_online: impl FnMut(usize) -> bool,
+) -> std::result::Result<Vec<&'a FileInfo>, &'static str> {
+    collect_inline_data_shard_fileinfos_from_observations(
+        parts_metadata
+            .iter()
+            .enumerate()
+            .map(|(index, metadata)| (index, Some(metadata))),
+        fi,
+        data_shards,
+        disk_is_online,
+    )
+}
+
+fn collect_inline_data_shard_fileinfos_from_observations<'a>(
+    observations: impl IntoIterator<Item = (usize, Option<&'a FileInfo>)>,
+    fi: &FileInfo,
+    data_shards: usize,
    mut disk_is_online: impl FnMut(usize) -> bool,
 ) -> std::result::Result<Vec<&'a FileInfo>, &'static str> {
    let distribution = &fi.erasure.distribution;
    let mut data_files = vec![None; data_shards];

-    for (disk_index, file_info) in parts_metadata.iter().enumerate() {
+    for (disk_index, file_info) in observations {
        if !disk_is_online(disk_index) {
            continue;
        }
@@ -5901,9 +5918,9 @@ fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
        if block_index == 0 || block_index > data_shards {
            continue;
        }
-        if file_info.name.is_empty() {
+        let Some(file_info) = file_info.filter(|metadata| !metadata.name.is_empty()) else {
            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD);
-        }
+        };
        if file_info.erasure.index != block_index {
            return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
        }
@@ -602,17 +602,18 @@ impl SetDisks {
        }

        scope.check()?;
-        let (_, errors) =
-            Self::read_all_fileinfo(disks.as_slice(), "", bucket, object, &version.to_string(), false, false, false).await?;
+        let metadata_read =
+            Self::read_metadata(disks.as_slice(), "", bucket, object, &version.to_string(), false, false, false).await?;
+        let all_absent = metadata_read
+            .slots
+            .iter()
+            .all(|slot| matches!(&slot.result, Some(Err(DiskError::FileNotFound | DiskError::FileVersionNotFound))));
+
        let selected = self.get_disks_internal().await;
        let same_targets = selected.len() == disks.len() && selected.iter().zip(&disks).all(|(current, original)| {
            matches!((current, original), (Some(current), Some(original)) if std::sync::Arc::ptr_eq(current, original))
        });
-        if !same_targets
-            || !errors
-                .iter()
-                .all(|error| matches!(error, Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)))
-        {
+        if !same_targets || !all_absent {
            return Err(StorageError::SlowDown);
        }
        scope.check()?;
@@ -741,7 +742,7 @@ impl SetDisks {
        object: &str,
        version_id: &str,
    ) -> disk::error::Result<ReadRepairCommitFingerprint> {
-        let (parts_metadata, errs, _) = Self::read_all_fileinfo_observed(
+        let metadata_read = Self::read_metadata_observed(
            disks,
            "",
            bucket,
@@ -754,6 +755,7 @@ impl SetDisks {
            self.default_parity_count,
        )
        .await?;
+        let (parts_metadata, errs, _) = metadata_read.into_legacy();
        let (read_quorum, _) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)?;
        let read_quorum = usize::try_from(read_quorum).map_err(|_| DiskError::ErasureReadQuorum)?;
        let (_, quorum_mod_time, quorum_etag) = Self::list_online_disks(disks, &parts_metadata, &errs, read_quorum);
@@ -938,8 +940,9 @@ impl SetDisks {
            }
        };

-        let (mut parts_metadata, errs) =
-            Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true, false).await?;
+        let (mut parts_metadata, errs, _) = Self::read_metadata(&disks, "", bucket, object, version_id, true, true, false)
+            .await?
+            .into_legacy();

        trace!(
            event = EVENT_SET_DISK_HEAL,
@@ -1291,8 +1294,10 @@ impl SetDisks {
                            // in `parts_metadata` to defaults. Re-read only before
                            // destructive cleanup so the guard sees every original
                            // identity.
-                            let (delete_guard_metadata, delete_guard_errs) =
-                                Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true, false).await?;
+                            let (delete_guard_metadata, delete_guard_errs, _) =
+                                Self::read_metadata(&disks, "", bucket, object, version_id, true, true, false)
+                                    .await?
+                                    .into_legacy();
                            if self
                                .dangling_delete_safety(bucket, object, &delete_guard_metadata, &delete_guard_errs, &disks)
                                .await?
@@ -2327,16 +2332,16 @@ impl SetDisks {
                return Err(DiskError::retired_marker_deferred(format!("conditional marker deletion failed: {error}")));
            }
        }
-        let (_, errors) = Self::read_all_fileinfo(disks, "", bucket, object, &version.to_string(), false, false, false).await?;
+        let metadata_read = Self::read_metadata(disks, "", bucket, object, &version.to_string(), false, false, false).await?;
+        let all_absent = metadata_read
+            .slots
+            .iter()
+            .all(|slot| matches!(&slot.result, Some(Err(DiskError::FileNotFound | DiskError::FileVersionNotFound))));
        let selected = self.get_disks_internal().await;
        let same_targets = selected.len() == disks.len() && selected.iter().zip(disks).all(|(current, original)| {
            matches!((current, original), (Some(current), Some(original)) if std::sync::Arc::ptr_eq(current, original))
        });
-        if !same_targets
-            || !errors
-                .iter()
-                .all(|error| matches!(error, Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)))
-        {
+        if !same_targets || !all_absent {
            return Err(DiskError::retired_marker_deferred(
                "complete marker absence could not be verified under the original target set",
            ));
@@ -3421,9 +3426,10 @@ impl SetDisks {
        // The inner heal and missing-object report read the registry again;
        // release this snapshot guard before a topology writer can queue between reads.
        let disks = self.get_disks_internal().await;
-        let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false)
+        let (_, errs, _) = Self::read_metadata(&disks, "", bucket, object, version_id, false, false, false)
            .await
-            .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
+            .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?
+            .into_legacy();
        if DiskError::is_all_not_found(&errs) {
            debug!(
                event = EVENT_SET_DISK_HEAL,
@@ -115,7 +115,28 @@ use super::GetCodecStreamingObjectClass;
 use super::GetCodecStreamingRollout;
 #[cfg(test)]
 use super::classify_get_codec_streaming_object_class;
-use super::core::io_primitives::*;
+use super::core::io_primitives::{
+    BitrotReaderSetup, BitrotReaderSetupAttribution, BitrotReaderSetupMode, DeferredReaderReopener, EVENT_SET_DISK_READ,
+    GetCodecStreamingReaderBuildOutcome, MetadataCacheLookup, ObjectBitrotReader, ReadRepairAdmissionSubmitter,
+    ReadRepairHealSubmission, SLOW_OBJECT_READ_LOG_THRESHOLD, codec_streaming_reader_setup_fallback_reason,
+    create_bitrot_readers_until_quorum_with_preference, create_data_block_bitrot_readers, resolved_read_repair_version_id,
+    send_read_repair_heal_request, shard_read_costs_for_disks, submit_read_repair_heal, submit_read_repair_heal_with_submitter,
+};
+#[cfg(test)]
+use super::core::io_primitives::{
+    ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_READER_SETUP, ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP,
+    MultipartCodecStreamingReader, ReadRepairAdmissionFuture, ReadRepairAdmissionOutcome, collect_read_multiple_results,
+    collect_read_parts_results, create_bitrot_readers_until_quorum, create_bitrot_readers_until_quorum_all_shards,
+    release_read_repair_heal_reservation, reserve_read_repair_heal, resolve_read_part_from_responses, shard_read_cost_for_disk,
+    shard_read_cost_for_endpoint,
+};
+#[cfg(test)]
+use super::core::metadata_quorum::{MetadataEarlyStopDecision, MetadataQuorumAccumulator};
+#[cfg(test)]
+use super::core::metadata_read::{
+    MetadataFanoutDiagnostics, MetadataFanoutObservation, metadata_early_stop_permitted, should_allow_metadata_early_stop,
+};
+use super::core::metadata_read::{late_materialization_candidate_is_safe, non_inline_data_read_early_stop_allowed};
 #[cfg(test)]
 use super::get_codec_streaming_config_cached_core;
 #[cfg(test)]
@@ -672,11 +693,11 @@ impl SetDisks {
        };

        // Early-stop for safe metadata reads is handled inside
-        // read_all_fileinfo_observed (see read_all_fileinfo_early_stop in
-        // core/io_primitives.rs); unsafe requests and callers that opt out
+        // read_metadata_observed (see read_all_fileinfo_early_stop in
+        // core/metadata_read.rs); unsafe requests and callers that opt out
        // (allow_early_stop=false) fall back to full-wait.
-        let (mut parts_metadata, errs, metadata_fanout_diagnostics) = if allow_read_version_coalescing {
-            Self::read_all_fileinfo_observed_for_get_object(
+        let metadata_read = if allow_read_version_coalescing {
+            Self::read_metadata_for_get_object(
                &disks,
                "",
                bucket,
@@ -689,7 +710,7 @@ impl SetDisks {
            )
            .await?
        } else {
-            Self::read_all_fileinfo_observed(
+            Self::read_metadata_observed(
                &disks,
                "",
                bucket,
@@ -703,13 +724,14 @@ impl SetDisks {
            )
            .await?
        };
+        let metadata_fanout_complete = metadata_read.is_complete();
+        let (mut parts_metadata, errs, metadata_fanout_diagnostics) = metadata_read.into_legacy();
        let metadata_metrics_path = if crate::bucket::utils::is_meta_bucketname(bucket) {
            GET_OBJECT_PATH_INTERNAL_META
        } else {
            GET_OBJECT_PATH_LEGACY_DUPLEX
        };
        metadata_fanout_diagnostics.record(metadata_metrics_path);
-        let metadata_fanout_complete = metadata_fanout_diagnostics.total_responses() >= disks.len();
        // warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata);
        // warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs);

@@ -2190,7 +2212,7 @@ impl SetDisks {
        expected: &LateMetadataIdentity,
        metrics_path: &'static str,
    ) -> Result<(FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>)> {
-        let (mut parts_metadata, errs, diagnostics) = SetDisks::read_all_fileinfo_observed(
+        let metadata_read = SetDisks::read_metadata_observed(
            fallback_disks,
            "",
            bucket,
@@ -2203,6 +2225,7 @@ impl SetDisks {
            expected.parity_blocks,
        )
        .await?;
+        let (mut parts_metadata, errs, diagnostics) = metadata_read.into_legacy();
        diagnostics.record(metrics_path);

        let (read_quorum, write_quorum) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, expected.parity_blocks)
@@ -2631,6 +2654,7 @@ fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOpti
 #[cfg(test)]
 mod metadata_cache_tests {
    use super::*;
+    use crate::set_disk::core::io_primitives::disk_call_counters;
    use rustfs_heal_contracts::heal_channel::HealAdmissionDropReason;
    use serial_test::serial;
    use std::sync::atomic::{AtomicUsize, Ordering};
@@ -354,6 +354,15 @@ fn ensure_producer_limit_state(options: &mut ListPathOptions) -> Arc<AtomicBool>
    state
 }

+fn list_path_folds_common_prefixes(options: &ListPathOptions) -> bool {
+    !options.versioned
+        && (options.include_directories
+            || options
+                .separator
+                .as_deref()
+                .is_some_and(|separator| !separator.is_empty() && separator != SLASH_SEPARATOR))
+}
+
 async fn can_skip_hidden_prefix_check(options: &ListPathOptions) -> bool {
    if options.recursive || options.incl_deleted || options.versioned {
        return false;
@@ -2440,6 +2449,12 @@ where
            let suffix = key.trim_start_matches(prefix);
            if let Some((common_prefix, _)) = suffix.split_once(separator) {
                let common_prefix = format!("{prefix}{common_prefix}{separator}");
+                if marker.is_some_and(|marker| common_prefix.as_str() <= marker) {
+                    if collect_stats {
+                        stats.skipped_keys += 1;
+                    }
+                    continue;
+                }
                if prefix_set.insert(common_prefix.clone()) {
                    if collect_stats {
                        stats.common_prefixes += 1;
@@ -2542,6 +2557,10 @@ fn list_objects_from_metadata_snapshot_candidates(
            let suffix = object.name.trim_start_matches(prefix);
            if let Some((common_prefix, _)) = suffix.split_once(separator) {
                let common_prefix = format!("{prefix}{common_prefix}{separator}");
+                if marker.is_some_and(|marker| common_prefix.as_str() <= marker) {
+                    stats.skipped_keys += 1;
+                    continue;
+                }
                if prefix_set.insert(common_prefix.clone()) {
                    stats.common_prefixes += 1;
                    visible_entries.push(VerifiedIndexVisibleEntry::Prefix(common_prefix));
@@ -4962,6 +4981,8 @@ async fn gather_results(
    let gather_started = list_metrics_enabled.then(std::time::Instant::now);
    let mut scanned_entries = 0usize;
    let mut candidate_entries = 0usize;
+    let mut previous_prefix = String::new();
+    let folds_prefixes = list_path_folds_common_prefixes(&opts);

    while let Some(mut entry) = recv.recv().await {
        scanned_entries += 1;
@@ -5001,6 +5022,34 @@ async fn gather_results(
            continue;
        }

+        // Version LIST projections omit cleanup-only free versions. They must
+        // not consume the lookahead budget before a later visible version.
+        // Cleanup walks consume the raw set stream without this collector.
+        if opts.versioned
+            && is_object
+            && entry
+                .xl_meta()
+                .is_ok_and(|meta| !meta.versions.is_empty() && meta.versions.iter().all(|version| version.header.free_version()))
+        {
+            continue;
+        }
+
+        if folds_prefixes
+            && let Some(separator) = opts.separator.as_deref().filter(|separator| !separator.is_empty())
+            && let Some(offset) = entry.name[opts.prefix.len()..].find(separator)
+        {
+            let common_prefix = &entry.name[..opts.prefix.len() + offset + separator.len()];
+            if common_prefix == previous_prefix
+                || opts.marker.as_deref().is_some_and(|marker| {
+                    (!opts.include_marker && common_prefix <= marker) || (opts.include_marker && common_prefix < marker)
+                })
+            {
+                continue;
+            }
+            previous_prefix.clear();
+            previous_prefix.push_str(common_prefix);
+        }
+
        // TODO(backlog): integrate lifecycle evaluation during object listing

        entries.push(Some(entry));
@@ -7023,7 +7072,9 @@ impl SetDisks {
        );

        let limit = {
-            if opts.limit > 0 && opts.stop_disk_at_limit {
+            // A raw scan budget can be consumed entirely by keys folding into
+            // one common prefix. Let the logical collector stop these walks.
+            if opts.limit > 0 && opts.stop_disk_at_limit && !list_path_folds_common_prefixes(&opts) {
                opts.limit + 4 + (opts.limit / 16)
            } else {
                0
@@ -7248,6 +7299,9 @@ fn calc_common_counter(infos: &[DiskInfo], read_quorum: usize) -> u64 {

 // list_path_raw

+#[cfg(test)]
+mod differential_tests;
+
 #[cfg(test)]
 mod test {
    use super::{
@@ -7926,6 +7980,117 @@ mod test {
        assert!(cancel.is_cancelled());
    }

+    #[tokio::test]
+    async fn list_path_gather_results_counts_common_prefixes_before_page_limit() {
+        for separator in ["/", "-"] {
+            for extra_object in [false, true] {
+                let prefix_name = format!("a{separator}");
+                let first = if separator == "/" {
+                    prefix_name.clone()
+                } else {
+                    format!("{prefix_name}first")
+                };
+                let mut input = vec![test_object_meta_entry(&first)];
+                if separator == "/" {
+                    input.push(test_dir_meta_entry(&prefix_name));
+                } else {
+                    input.push(test_object_meta_entry(&format!("{prefix_name}second")));
+                }
+                input.push(test_object_meta_entry("b"));
+                if extra_object {
+                    input.push(test_object_meta_entry("c"));
+                }
+                let (entry_tx, entry_rx) = mpsc::channel(input.len());
+                let (result_tx, mut result_rx) = mpsc::channel(1);
+                for entry in input {
+                    entry_tx.send(entry).await.expect("delimiter candidates should queue");
+                }
+                drop(entry_tx);
+                let state = gather_results(
+                    CancellationToken::new(),
+                    ListPathOptions {
+                        bucket: "bucket".to_owned(),
+                        separator: Some(separator.to_owned()),
+                        recursive: separator != "/",
+                        include_directories: separator == "/",
+                        limit: 3,
+                        ..Default::default()
+                    },
+                    entry_rx,
+                    result_tx,
+                )
+                .await
+                .expect("delimiter collection should succeed");
+                let result = result_rx.recv().await.expect("delimiter page should arrive");
+                let mut expected = vec![first, "b".to_owned()];
+                if extra_object {
+                    expected.push("c".to_owned());
+                }
+                assert_eq!(
+                    result
+                        .entries
+                        .expect("delimiter entries should exist")
+                        .entries()
+                        .into_iter()
+                        .map(|entry| entry.name.clone())
+                        .collect::<Vec<_>>(),
+                    expected,
+                    "delimiter={separator}, extra={extra_object}"
+                );
+                assert_eq!(result.err.is_none(), extra_object, "n=max must reach EOF; max+1 must retain lookahead");
+                assert_eq!(
+                    state,
+                    if extra_object {
+                        GatherResultsState::LimitReached
+                    } else {
+                        GatherResultsState::InputClosed
+                    }
+                );
+            }
+        }
+        for include_marker in [false, true] {
+            let (entry_tx, entry_rx) = mpsc::channel(2);
+            let (result_tx, mut result_rx) = mpsc::channel(1);
+            for name in ["a/child", "b"] {
+                entry_tx
+                    .send(test_object_meta_entry(name))
+                    .await
+                    .expect("marker candidates should queue");
+            }
+            drop(entry_tx);
+            gather_results(
+                CancellationToken::new(),
+                ListPathOptions {
+                    bucket: "bucket".to_owned(),
+                    separator: Some("/".to_owned()),
+                    recursive: true,
+                    include_directories: true,
+                    marker: Some("a/".to_owned()),
+                    include_marker,
+                    limit: 3,
+                    ..Default::default()
+                },
+                entry_rx,
+                result_tx,
+            )
+            .await
+            .expect("logical marker collection should succeed");
+            let entries = result_rx
+                .recv()
+                .await
+                .expect("logical marker page should arrive")
+                .entries
+                .expect("logical marker entries should exist");
+            let projected = ObjectInfo::from_meta_cache_entries_sorted_infos(&entries, "bucket", "", Some("/".to_owned())).await;
+            let names: Vec<_> = projected.into_iter().map(|object| object.name).collect();
+            assert_eq!(
+                names,
+                if include_marker { vec!["a/", "b"] } else { vec!["b"] },
+                "a folded prefix equal to the marker must follow include_marker={include_marker}"
+            );
+        }
+    }
+
    #[test]
    fn list_versions_pagination_scan_limit_boundaries() {
        for has_version_marker in [false, true] {
@@ -8977,6 +9142,29 @@ mod test {
        assert_eq!(result.info.objects[0].etag.as_deref(), Some("etag-z"));
        assert_eq!(result.info.prefixes, vec!["photos/2026/nested/".to_string()]);
        assert!(!result.info.is_truncated);
+
+        let resumed = list_objects_from_metadata_snapshot_candidates(
+            "bucket",
+            "photos/2026/",
+            Some("photos/2026/nested/"),
+            &Some("/".to_string()),
+            1,
+            &objects,
+        );
+        assert!(
+            resumed.info.prefixes.is_empty(),
+            "metadata-fast cursors must not replay an emitted prefix"
+        );
+        assert_eq!(
+            resumed
+                .info
+                .objects
+                .iter()
+                .map(|object| object.name.as_str())
+                .collect::<Vec<_>>(),
+            ["photos/2026/z.jpg"]
+        );
+        assert!(!resumed.info.is_truncated);
    }

    #[test]
@@ -10089,6 +10277,23 @@ mod test {
        assert_eq!(result.prefixes, vec!["photos/2026/archive/".to_string()]);
        assert!(result.is_truncated);
        assert_eq!(result.next_marker.as_deref(), Some("photos/2026/archive/"));
+
+        let resumed = list_objects_from_verified_index_candidates(
+            "photos/2026/",
+            result.next_marker.as_deref(),
+            &Some("/".to_string()),
+            1,
+            &candidates,
+            |key| async move { Ok(Some(test_live_object_info(&key, "live-etag"))) },
+        )
+        .await
+        .expect("verified prefix continuation should succeed");
+        assert!(resumed.prefixes.is_empty(), "verified cursors must not replay an emitted prefix");
+        assert_eq!(
+            resumed.objects.iter().map(|object| object.name.as_str()).collect::<Vec<_>>(),
+            ["photos/2026/d.jpg"]
+        );
+        assert!(!resumed.is_truncated);
    }

    #[tokio::test]
@@ -0,0 +1,14 @@
+[
+  "a",
+  "a/",
+  "a/b",
+  "a/b/child",
+  "a0",
+  "b",
+  "b/leaf",
+  "quote\"&<",
+  "space key",
+  "z",
+  "é",
+  "中/文"
+]
@@ -618,6 +618,7 @@ impl MetaCacheEntries {
        }

        let mut dir_exists = 0;
+        let mut directory = None;
        let mut selected = None;

        params.candidates.clear();
@@ -633,7 +634,7 @@ impl MetaCacheEntries {
            }
            if entry.is_dir() {
                dir_exists += 1;
-                selected = Some(entry.clone());
+                directory = Some(entry.clone());
                debug!(entry = %entry.name, "metacache resolve observed directory candidate");
                continue;
            }
@@ -664,6 +665,10 @@ impl MetaCacheEntries {
            }
        }

+        // A prefix and an explicit directory-marker object can share a name.
+        // Keep directory evidence separate so its arrival cannot displace an
+        // object that meets the metadata quorum.
+        let directory = directory.filter(|_| dir_exists >= params.dir_quorum);
        let Some(selected) = selected else {
            debug!(
                bucket = %params.bucket,
@@ -671,19 +676,9 @@ impl MetaCacheEntries {
                obj_quorum = params.obj_quorum,
                "metacache resolve could not select any candidate entry"
            );
-            return None;
+            return directory;
        };

-        if selected.is_dir() && dir_exists >= params.dir_quorum {
-            debug!(
-                entry = %selected.name,
-                dir_exists,
-                dir_quorum = params.dir_quorum,
-                "metacache resolve selected directory candidate after quorum reconciliation"
-            );
-            return Some(selected);
-        }
-
        // If we would never be able to reach the required object quorum.
        if objs_valid < params.obj_quorum {
            debug!(
@@ -691,7 +686,7 @@ impl MetaCacheEntries {
                obj_quorum = params.obj_quorum,
                "metacache resolve did not have enough valid object candidates to satisfy quorum"
            );
-            return None;
+            return directory;
        }

        if objs_agree == objs_valid {
@@ -724,7 +719,7 @@ impl MetaCacheEntries {

        let Some(cached) = selected.cached else {
            debug!(selected = %selected.name, "metacache resolve could not merge because selected candidate had no cached metadata");
-            return None;
+            return directory;
        };

        let versions = if enforce_write_quorum {
@@ -744,7 +739,7 @@ impl MetaCacheEntries {
                requested_versions = params.requested_versions,
                "metacache resolve produced no merged versions after reconciliation"
            );
-            return None;
+            return directory;
        }

        let merged_cached = FileMeta {
@@ -761,7 +756,7 @@ impl MetaCacheEntries {
                    error = ?e,
                    "metacache resolve failed to marshal merged metadata entry"
                );
-                return None;
+                return directory;
            }
        };

@@ -2534,6 +2529,57 @@ mod tests {
        assert!(resolved.is_none());
    }

+    #[test]
+    fn resolve_directory_marker_and_prefix_preserve_quorum_in_both_orders() {
+        let modified = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let mut object = metacache_entry_with_erasure(modified, "marker-etag", 2, 2);
+        object.name = "folder/".to_string();
+        let directory = metacache_dir_entry("folder/");
+        for enforce_write_quorum in [false, true] {
+            for (object_copies, directory_copies) in [(3, 1), (3, 3), (2, 3), (0, 3), (2, 2)] {
+                for reverse in [false, true] {
+                    let mut entries = vec![Some(object.clone()); object_copies];
+                    entries.extend(vec![Some(directory.clone()); directory_copies]);
+                    if reverse {
+                        entries.reverse();
+                    }
+                    let entries = MetaCacheEntries(entries);
+                    let params = MetadataResolutionParams {
+                        dir_quorum: 3,
+                        obj_quorum: 3,
+                        requested_versions: 1,
+                        bucket: "bucket".to_string(),
+                        strict: true,
+                        ..Default::default()
+                    };
+                    let resolved = if enforce_write_quorum {
+                        entries.resolve_with_write_quorum(params)
+                    } else {
+                        entries.resolve(params)
+                    };
+                    let context = format!(
+                        "objects={object_copies}, directories={directory_copies}, reverse={reverse}, write={enforce_write_quorum}"
+                    );
+                    if object_copies >= 3 {
+                        let resolved = resolved.expect("metadata at quorum must shadow a same-named prefix");
+                        assert!(resolved.is_object(), "{context}");
+                        assert_eq!(
+                            resolved
+                                .to_fileinfo("bucket")
+                                .expect("marker metadata should decode")
+                                .mod_time,
+                            Some(modified)
+                        );
+                    } else if directory_copies >= 3 {
+                        assert!(resolved.expect("prefix at quorum must remain a fallback").is_dir(), "{context}");
+                    } else {
+                        assert!(resolved.is_none(), "{context}");
+                    }
+                }
+            }
+        }
+    }
+
    fn build_hashmap_cache(update_size: usize) -> Arc<Cache<HashMap<usize, usize>>> {
        let generation = Arc::new(AtomicUsize::new(0));
        Arc::new(Cache::new(
@@ -11,6 +11,8 @@

 ## Open Items

+- `backlog-2253-metadata-observation-adapter` metadata observations: preserve aligned metadata/error slices for existing layout, shard and exact-version PUT, MPU and delete consumers while read/heal scheduling uses typed disk observations. Pending slots acquire empty placeholders only in this adapter. Remove after those remaining consumers use typed observations; delete the adapter and old facade in a separate cleanup.
+
 - `connect-894` pending Connect heartbeats: replay the exact preceding producer-capability list, with or without jobs, when upgrading to the memory-service capability. Preserve request ID, sequence, and persisted body rather than inserting the new capability into a retry. Remove after upgrades from the pre-service-memory capability set are unsupported.

 - `backlog-2539` administrator erasure-set scope: decode historical pending intents whose empty bucket list omitted the all-buckets marker. Normalize that representation to the explicit scope before replay or checkpoint comparison. Remove after all pre-marker administrator ErasureSet intents are retired; deployment verification must confirm that no such pending records remain on coordinator disks.
@@ -83,3 +83,39 @@ A candidate is ready for code movement only when all of these hold:
 - old facade names have compatibility tests or explicit deprecation coverage;
 - focused tests cover the changed owner path before any full gate is attempted;
 - rollback preserves object IO, quorum, lifecycle/replication queues, scanner repair, notification/audit events, and metadata compatibility.
+
+## Metadata read ownership and identity
+
+`set_disk/core/metadata_read.rs` owns disk/RPC scheduling, coalescing, timeouts,
+hedging and cancellation. Its result retains each disk slot as pending, a
+successful `FileInfo`, or a typed `DiskError`. Pending includes both an
+unscheduled disk and a response not yet observed; it contributes no vote and
+does not establish absence. Task IDs retain the disk slot when a task panics.
+The existing full-wait and early-stop gates, coalescer admission, cancellation
+drain and fallback policy remain at this scheduling boundary.
+
+`set_disk/core/metadata_quorum.rs` consumes those observations without IO.
+`SetDisks` remains the sole owner of topology, locks and caches. Read and heal
+callers use the typed result; `MetadataReadResult::into_legacy` is the explicit
+adapter for existing quorum/layout and shard-reader helpers. It moves successful
+metadata without cloning and creates empty placeholders only at that adapter.
+Old full-wait facade callers retain the same aligned metadata/error slices.
+
+These identities have different purposes and must not become one generic hash:
+
+| Decision | Equal fields | Deliberately different or separately checked |
+| --- | --- | --- |
+| Quorum grouping (`file_info_quorum_hash`) | Size, deletion/restore/transition state, timestamps, version and data-directory UUIDs, checksums, mode/writer version, normalized metadata, target delete-marker versions and part identities. Payloads also include data/parity counts and distribution. | Object name and volume come from the request; per-disk erasure index and shard bytes are not an object identity. Canonical delete markers and zero-size objects do not add payload geometry. |
+| Early-stop strict match (`metadata_early_stop_candidate_matches`) | Request name/volume, version/latest/deletion state, transition fields, size/time, mode/writer version, complete metadata/replication state/parts/checksum, version-history fields, data directory, algorithm/block size/data/parity/distribution. | Per-disk erasure index and data bytes differ. A non-inline read reserve additionally proves distinct indexes mapped to their disk slots; inline reads verify shard identity and bitrot/body content before stopping. |
+| Late materialization (`LateMetadataIdentity`) | Request name/volume, algorithm, block size, legacy-checksum mode and quorum hash. | Each late shard must match the original distribution at its disk slot. A changed identity or insufficient matching shards fails read quorum; it cannot mix generations into a body. |
+
+`metadata_observation_all_four_slot_states_and_arrival_orders_preserve_reduction`
+checks all 625 four-disk success/missing/corrupt/offline/pending combinations and
+24 arrival orders with early-stop on and off. The observation regressions also
+cover a newer pending response, same-time different directories, duplicated
+shard indexes, null versions, canonical delete markers and invalid success
+payloads. Existing bounded-fanout, legacy inline bitrot, cache, late-metadata and
+GET/Range regressions continue to exercise the real scheduling and reader paths;
+the pure matrix does not replace those checks. Their call counters bound actual
+disk reads, including canceled tails, without adding a scheduling policy or a
+per-request metadata clone.
@@ -82,7 +82,7 @@ A single filesystem rename does not atomically commit a sidecar plus `xl.meta`.

 A laggard need not contain the predecessor. Recovery fetches the chosen decision and its verified metadata, reconstructs or validates its shards under [erasure-coding.md](erasure-coding.md), and installs that state. An empty replacement uses a fresh disk incarnation and cannot reuse old preparation receipts. A disk whose durable state claims a later decision than the supplied one rejects the operation; a conflicting same-revision digest is quarantined. Never erase a divergent disk merely because it is in the minority.

-`rollback_committed_rename_std` and `rollback_inline_metadata_commit_std` in `crates/ecstore/src/disk/local/commit.rs`, and `restore_metadata_backup` in `crates/ecstore/src/disk/local.rs`, must become decision-aware before strict mode includes them. The permitted rollback is limited to a transaction's unchosen private preparation, or restoration proven by recovery to be necessary before any newer local effect. A chosen operation is repaired forward. Neither a client timeout nor a rename-tail error permits reverting an acknowledged decision.
+`RenameCommitContext::rollback_non_inline_std` and `RenameCommitContext::rollback_inline_std` in `crates/ecstore/src/disk/local/commit.rs`, and `restore_metadata_backup` in `crates/ecstore/src/disk/local.rs`, must become decision-aware before strict mode includes them. The permitted rollback is limited to a transaction's unchosen private preparation, or restoration proven by recovery to be necessary before any newer local effect. A chosen operation is repaired forward. Neither a client timeout nor a rename-tail error permits reverting an acknowledged decision.

 `RenameConvergence` remains a post-publication repair signal. `PartialCommit`, `SignatureDivergent`, and `Unknown` do not decide which transaction won. Keep their diagnostics and quorum accounting; resolve authority first. Early ACK may still precede minority-tail completion after the two quorum conditions hold. Tests that inspect all disks must synchronize the tail or assert the permitted minority residue separately.

@@ -58,7 +58,7 @@ Fault injection is explicit and deterministic: local disk mocks for unit tests,

 `ObjectOptions::default()` uses `WriteCompletion::Quorum`: a namespace-lock-owning PUT may acknowledge write quorum while its rename tail retains the lock. A fixture that immediately inspects every disk or primes a metadata generation must set `write_completion: WriteCompletion::TailDrained` and keep normal locking. TailDrained waits for the existing rename fan-out; it does not require every disk to succeed or change fsync policy. Codec-only `no_lock` fixtures do not cover namespace locking.

-The object tests reuse `rename_fanout_barrier::arm(object, disk_slot, phase)` and `observe_tasks(object)`. Wait for the barrier with a deadline, observe actual metadata quorum with `wait_for_paused_tail_metadata_quorum`, then release or cancel. The metadata check distinguishes a real quorum from disk tasks that have not started. Assert zero remaining rename tasks after the owned coordinator releases its lock; cancellation tests also wait for staging cleanup.
+The object tests reuse `rename_fanout_barrier::arm(object, disk_slot, phase)` and `observe_tasks(object)`. The fixture uses typed rename, rollback, cleanup, read-version and hedge-timer phases; multiple disk/phase checkpoints can be armed for one object. A duplicate live checkpoint is rejected instead of replacing its owner. `wait_until_paused_before(deadline)` reports the exact unreached checkpoint on timeout. Wait for the barrier with a deadline, observe actual metadata quorum with `wait_for_paused_tail_metadata_quorum`, then release or cancel. The metadata check distinguishes a real quorum from disk tasks that have not started. Assert zero remaining rename tasks after the owned coordinator releases its lock; cancellation tests also wait for staging cleanup.

 | Fixture | Completion boundary |
 |---|---|
@@ -1113,22 +1113,11 @@ mod tests {

    #[test]
    fn list_through_static_namespace_boundary_matrix() {
-        let corpus = [
-            "a",
-            "a/",
-            "a/b",
-            "a/b/child",
-            "a0",
-            "b",
-            "b/leaf",
-            "quote\"&<",
-            "space key",
-            "z",
-            "é",
-            "中/文",
-        ];
+        let corpus: Vec<String> =
+            serde_json::from_str(include_str!("../../../crates/ecstore/tests/fixtures/list_namespace_keys.json"))
+                .expect("shared LIST namespace corpus should decode");
        for count in [0, 1, 3, 4, corpus.len()] {
-            let keys: Vec<String> = corpus[..count].iter().map(|key| (*key).to_string()).collect();
+            let keys = corpus[..count].to_vec();
            for placement in 0..3 {
                let (local, source): (Vec<_>, Vec<_>) =
                    keys.iter()