test(ecstore): preserve inline budget semantics

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

crates/ecstore/src/config/storageclass.rs

@@ -101,6 +101,7 @@ const DEFAULT_RRS_STORAGE_CLASS: &str = "EC:1";
 const ZERO_SET_DRIVE_COUNT_ERROR: &str = "set drive count must be greater than zero";
 
 pub static DEFAULT_INLINE_BLOCK: usize = 128 * 1024;
+const DEFAULT_INLINE_OBJECT_BUDGET: usize = 2 * DEFAULT_INLINE_BLOCK;
 
 pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
     let kvs = vec![
@@ -150,6 +151,8 @@ pub struct Config {
     optimize: Option<String>,
     inline_block: usize,
     initialized: bool,
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    inline_block_explicit: bool,
     #[serde(skip)]
     standard_parities: Vec<PoolParity>,
     #[serde(skip)]
@@ -233,17 +236,19 @@ impl Config {
             .map(|(pool_index, pool)| (pool_index, pool.drives_per_set))
     }
 
-    pub fn should_inline(&self, shard_size: i64, versioned: bool) -> bool {
-        if shard_size < 0 {
+    pub fn should_inline(&self, shard_size: i64, data_shards: usize, versioned: bool) -> bool {
+        if shard_size < 0 || data_shards == 0 {
             return false;
         }
 
         let shard_size = shard_size as usize;
-
-        let mut inline_block = DEFAULT_INLINE_BLOCK;
-        if self.initialized {
-            inline_block = self.inline_block;
-        }
+        // Keep the historical two-data-shard object budget while preventing
+        // wider EC layouts from multiplying the maximum inline object size.
+        let inline_block = if self.initialized && self.inline_block_explicit {
+            self.inline_block
+        } else {
+            (DEFAULT_INLINE_OBJECT_BUDGET / data_shards).min(DEFAULT_INLINE_BLOCK)
+        };
 
         if versioned {
             shard_size <= inline_block / 8
@@ -392,6 +397,7 @@ fn lookup_config_for_pools_with_env(
     }
 
     let optimize = overrides.optimize;
+    let inline_block_explicit = overrides.inline_block.is_some();
     let inline_block = if let Some(value) = overrides.inline_block {
         let block = value
             .parse::<bytesize::ByteSize>()
@@ -424,6 +430,7 @@ fn lookup_config_for_pools_with_env(
         optimize,
         inline_block,
         initialized: true,
+        inline_block_explicit,
         standard_parities,
         rrs_parities,
     })
@@ -541,22 +548,26 @@ mod tests {
     }
 
     #[test]
-    fn should_inline_preserves_exact_default_shard_boundaries() {
-        let config = Config::default();
+    fn should_inline_scales_default_threshold_by_data_shards() {
+        let config = lookup_config_for_pools_with_env(&KVS::new(), &[3, 12], no_env_overrides())
+            .expect("default inline policy should resolve for EC2+1 and EC8+4");
 
-        for (case, shard_size, versioned, expected) in [
-            ("unversioned below", 128 * 1024 - 1, false, true),
-            ("unversioned exact", 128 * 1024, false, true),
-            ("unversioned above", 128 * 1024 + 1, false, false),
-            ("versioned below", 16 * 1024 - 1, true, true),
-            ("versioned exact", 16 * 1024, true, true),
-            ("versioned above", 16 * 1024 + 1, true, false),
-            ("negative", -1, false, false),
+        for (case, shard_size, data_shards, versioned, expected) in [
+            ("EC2+1 unversioned exact", 128 * 1024, 2, false, true),
+            ("EC2+1 unversioned above", 128 * 1024 + 1, 2, false, false),
+            ("EC2+1 versioned exact", 16 * 1024, 2, true, true),
+            ("EC2+1 versioned above", 16 * 1024 + 1, 2, true, false),
+            ("EC8+4 unversioned exact", 32 * 1024, 8, false, true),
+            ("EC8+4 unversioned above", 32 * 1024 + 1, 8, false, false),
+            ("EC8+4 versioned exact", 4 * 1024, 8, true, true),
+            ("EC8+4 versioned above", 4 * 1024 + 1, 8, true, false),
+            ("negative", -1, 2, false, false),
+            ("zero data shards", 0, 0, false, false),
         ] {
             assert_eq!(
-                config.should_inline(shard_size, versioned),
+                config.should_inline(shard_size, data_shards, versioned),
                 expected,
-                "{case}: shard_size={shard_size}, versioned={versioned}"
+                "{case}: shard_size={shard_size}, data_shards={data_shards}, versioned={versioned}"
             );
         }
     }
@@ -577,13 +588,28 @@ mod tests {
             let shard_size = erasure.shard_file_size(object_size);
             assert_eq!(shard_size, expected_shard_size, "{case}: object_size={object_size}");
             assert_eq!(
-                config.should_inline(shard_size, versioned),
+                config.should_inline(shard_size, erasure.data_shards, versioned),
                 expected,
                 "{case}: object_size={object_size}, shard_size={shard_size}, versioned={versioned}"
             );
         }
     }
 
+    #[test]
+    fn explicit_inline_block_preserves_fixed_per_shard_rollback() {
+        let overrides = StorageClassEnvOverrides {
+            inline_block: Some("128KiB".to_string()),
+            ..Default::default()
+        };
+        let config = lookup_config_for_pools_with_env(&KVS::new(), &[12], overrides)
+            .expect("explicit inline block should resolve for EC8+4");
+
+        assert!(config.should_inline(128 * 1024, 8, false));
+        assert!(!config.should_inline(128 * 1024 + 1, 8, false));
+        assert!(config.should_inline(16 * 1024, 8, true));
+        assert!(!config.should_inline(16 * 1024 + 1, 8, true));
+    }
+
     #[test]
     fn write_capability_contract_only_accepts_implemented_layouts() {
         assert_eq!(SUPPORTED_WRITE_CLASSES, [STANDARD, RRS]);
@@ -777,6 +803,7 @@ mod tests {
         let encoded = serde_json::to_string(&cfg).expect("config should serialize");
         assert!(!encoded.contains("standard_parities"));
         assert!(!encoded.contains("rrs_parities"));
+        assert!(!encoded.contains("inline_block_explicit"));
 
         let decoded: Config = serde_json::from_str(&encoded).expect("legacy scalar config should deserialize");
         assert_eq!(decoded.get_parity_for_sc(STANDARD), Some(2));
@@ -786,6 +813,25 @@ mod tests {
         assert!(validate_parity(0, 0).is_err());
     }
 
+    #[test]
+    fn explicit_inline_block_survives_config_round_trip() {
+        let cfg = lookup_config_for_pools_with_env(
+            &KVS::new(),
+            &[12],
+            StorageClassEnvOverrides {
+                inline_block: Some("128KiB".to_string()),
+                ..Default::default()
+            },
+        )
+        .expect("explicit inline block should resolve");
+        assert!(cfg.should_inline(100 * 1024, 8, false));
+
+        let encoded = serde_json::to_string(&cfg).expect("config should serialize");
+        assert!(encoded.contains("\"inline_block_explicit\":true"));
+        let decoded: Config = serde_json::from_str(&encoded).expect("explicit inline config should deserialize");
+        assert!(decoded.should_inline(100 * 1024, 8, false));
+    }
+
     #[test]
     fn lookup_config_reads_rrs_from_class_rrs_key() {
         // Regression: kvs.get(RRS) used RRS="REDUCED_REDUNDANCY" instead of

crates/ecstore/src/runtime/sources.rs

@@ -236,10 +236,6 @@ pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option<usize>
     get_global_storage_class_snapshot().get_parity_for_sc(storage_class.unwrap_or_default())
 }
 
-pub(crate) fn storage_class_should_inline(shard_size: i64, versioned: bool) -> bool {
-    get_global_storage_class_snapshot().should_inline(shard_size, versioned)
-}
-
 pub(crate) fn deployment_upload_id(upload_id: &str) -> String {
     base64_simd::URL_SAFE_NO_PAD
         .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_id).as_bytes())

crates/ecstore/src/set_disk/mod.rs

@@ -880,12 +880,44 @@ mod prepared_get_object_metadata_tests {
     use super::*;
     use crate::ecstore_validation_blackbox::make_local_set_disks;
     use crate::object_api::{BLOCK_SIZE_V2, PutObjReader};
-    use crate::set_disk::core::io_primitives::disk_call_counters;
+    use crate::set_disk::core::io_primitives::{bounded_metadata_fanout_order, disk_call_counters, rename_fanout_barrier};
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
+    use crate::test_metrics::CapturingRecorder;
     use http::HeaderMap;
     use tokio::io::AsyncReadExt;
 
+    const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
+
+    fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {
+        (0..1000)
+            .map(|index| format!("{prefix}-{index}.bin"))
+            .find(|name| {
+                let order = bounded_metadata_fanout_order(bucket, name, 4, 2);
+                let distribution = FileInfo::new(&[bucket, name].join("/"), 2, 2).erasure.distribution;
+                let mut seen = [false; 2];
+                for disk_index in order.into_iter().take(3) {
+                    if let Some(block_index @ 1..=2) = distribution.get(disk_index).copied() {
+                        seen[block_index - 1] = true;
+                    }
+                }
+                seen.into_iter().all(|seen| seen)
+            })
+            .expect("test should find an object whose initial fanout covers both data shards")
+    }
+
+    fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
+        *bounded_metadata_fanout_order(bucket, object, 4, 2)
+            .get(3)
+            .expect("4-disk test geometry should leave one bounded spare disk")
+    }
+
+    fn bounded_slow_initial_disk_index(bucket: &str, object: &str) -> usize {
+        *bounded_metadata_fanout_order(bucket, object, 4, 2)
+            .get(2)
+            .expect("4-disk test geometry should include a third initial metadata disk")
+    }
+
     #[tokio::test]
     async fn prepared_metadata_is_consumed_exactly_once() {
         let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new());
@@ -1002,6 +1034,307 @@ mod prepared_get_object_metadata_tests {
         );
     }
 
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn inline_data_read_early_stop_reader_returns_exact_body() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "inline-data-read-early-stop-reader";
+        let object = object_with_initial_data_shards(bucket, "inline-data-read-early-stop-reader-object");
+        let payload = b"inline early-stop reader payload".repeat(256);
+        let recorder = CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        let (restored, object_size, calls_total) = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                let opts = ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                };
+
+                set_disks
+                    .make_bucket(bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("bucket should be created");
+                let mut put_reader = PutObjReader::from_vec(payload.clone());
+                set_disks
+                    .put_object(bucket, &object, &mut put_reader, &opts)
+                    .await
+                    .expect("inline object should be written");
+
+                temp_env::async_with_vars(
+                    [
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                    ],
+                    async {
+                        let slow_initial_disk = bounded_slow_initial_disk_index(bucket, &object);
+                        let barrier =
+                            rename_fanout_barrier::arm(&object, slow_initial_disk, rename_fanout_barrier::PHASE_READ_VERSION);
+                        let calls = disk_call_counters::observe(&object);
+                        let set_disks_for_read = Arc::clone(&set_disks);
+                        let opts_for_read = opts.clone();
+                        let object_for_read = object.clone();
+                        let mut open_reader = tokio::spawn(async move {
+                            set_disks_for_read
+                                .get_object_reader(bucket, &object_for_read, None, HeaderMap::new(), &opts_for_read)
+                                .await
+                        });
+
+                        tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
+                            .await
+                            .expect("bounded inline GET should pause a slow initial metadata read");
+                        let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut open_reader)
+                            .await
+                            .expect("production inline GET should return before the paused metadata response")
+                            .expect("inline GET reader task should not panic")
+                            .expect("inline GET reader should open");
+                        let object_size = reader.object_info.size;
+                        let mut restored = Vec::new();
+                        reader
+                            .stream
+                            .read_to_end(&mut restored)
+                            .await
+                            .expect("inline GET body should stream");
+
+                        (restored, object_size, calls.total(disk_call_counters::KIND_READ_VERSION))
+                    },
+                )
+                .await
+            })
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        assert_eq!(object_size, payload.len() as i64);
+        assert_eq!(restored, payload);
+        assert_eq!(calls_total, 4, "bounded production GET should schedule the initial quorum plus one spare");
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_scheduled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "bounded production GET should record all scheduled metadata tasks"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_completed",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![3.0],
+            "bounded production GET should record only observed metadata responses as completed"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_cancelled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![1.0],
+            "bounded production GET should record the aborted slow metadata task"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn prepared_metadata_uses_full_fanout_even_when_data_read_early_stop_is_enabled() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "prepared-metadata-early-stop-enabled";
+        let object = object_with_initial_data_shards(bucket, "prepared-metadata-early-stop-enabled-object");
+        let payload = b"prepared metadata early-stop enabled payload".repeat(16);
+        let recorder = CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                let opts = ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                };
+
+                set_disks
+                    .make_bucket(bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("bucket should be created");
+                let mut put_reader = PutObjReader::from_vec(payload.clone());
+                set_disks
+                    .put_object(bucket, &object, &mut put_reader, &opts)
+                    .await
+                    .expect("object should be written");
+
+                temp_env::async_with_vars(
+                    [
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                    ],
+                    async {
+                        let calls = disk_call_counters::observe(&object);
+                        let metadata = set_disks
+                            .prepare_get_object_metadata(bucket, &object, &opts)
+                            .await
+                            .expect("prepared metadata should resolve");
+                        let calls_total = calls.total(disk_call_counters::KIND_READ_VERSION);
+
+                        let mut reader = set_disks
+                            .get_object_reader_with_prepared_metadata(bucket, &object, None, HeaderMap::new(), &opts, metadata)
+                            .await
+                            .expect("prepared body reader should open");
+                        let mut restored = Vec::new();
+                        reader
+                            .stream
+                            .read_to_end(&mut restored)
+                            .await
+                            .expect("prepared body should stream");
+                        (restored, calls_total)
+                    },
+                )
+                .await
+            })
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        assert_eq!(restored, payload);
+        assert_eq!(
+            calls_total, 4,
+            "prepared metadata must opt out of data-read early-stop until the read shape is known"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_scheduled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "prepared metadata should schedule the full metadata fanout"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_completed",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "prepared metadata must wait for every scheduled metadata response"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_cancelled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![0.0],
+            "prepared metadata must not cancel metadata responses"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn data_read_early_stop_request_shapes_full_wait_in_production_reader() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "data-read-early-stop-shape-reader";
+        let payload = b"shape-gated inline reader payload".repeat(256);
+
+        for (object_prefix, range, configure_opts, expected_body) in [
+            (
+                "data-read-early-stop-range-reader-object",
+                Some(HTTPRangeSpec {
+                    start: 0,
+                    end: 3,
+                    is_suffix_length: false,
+                }),
+                None,
+                payload[..4].to_vec(),
+            ),
+            ("data-read-early-stop-part-reader-object", None, Some(1), payload.clone()),
+        ] {
+            let recorder = CapturingRecorder::default();
+            let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+            rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+            let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
+                runtime.block_on(async {
+                    let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                    let object = object_with_initial_data_shards(bucket, object_prefix);
+                    let mut opts = ObjectOptions {
+                        no_lock: true,
+                        ..Default::default()
+                    };
+                    opts.part_number = configure_opts;
+
+                    set_disks
+                        .make_bucket(bucket, &MakeBucketOptions::default())
+                        .await
+                        .expect("bucket should be created");
+                    let mut put_reader = PutObjReader::from_vec(payload.clone());
+                    set_disks
+                        .put_object(bucket, &object, &mut put_reader, &opts)
+                        .await
+                        .expect("inline object should be written");
+
+                    temp_env::async_with_vars(
+                        [
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                            ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                        ],
+                        async {
+                            let calls = disk_call_counters::observe(&object);
+                            let mut reader = set_disks
+                                .get_object_reader(bucket, &object, range, HeaderMap::new(), &opts)
+                                .await
+                                .expect("shape-gated GET reader should open");
+                            let mut restored = Vec::new();
+                            reader
+                                .stream
+                                .read_to_end(&mut restored)
+                                .await
+                                .expect("shape-gated GET body should stream");
+                            (restored, calls.total(disk_call_counters::KIND_READ_VERSION))
+                        },
+                    )
+                    .await
+                })
+            });
+            rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+            assert_eq!(restored, expected_body);
+            assert_eq!(calls_total, 4, "shape-gated production GET should keep full metadata fanout");
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_scheduled",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![4.0],
+                "shape-gated production GET should schedule the full metadata fanout"
+            );
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_completed",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![4.0],
+                "shape-gated production GET must wait for every scheduled metadata response"
+            );
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_cancelled",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![0.0],
+                "shape-gated production GET must not cancel metadata responses"
+            );
+        }
+    }
+
     #[tokio::test]
     #[serial_test::serial(body_cache_hook)]
     async fn prepared_reader_rebuilds_object_info_when_precomputed_value_is_absent() {
@@ -1105,7 +1438,7 @@ impl SetDisks {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<PreparedGetObjectMetadata> {
-        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, true).await?;
+        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, false).await?;
         let object_info = build_get_object_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended);
         Ok(PreparedGetObjectMetadata {
             snapshot,
@@ -2408,12 +2741,12 @@ mod write_layout_tests {
 
         let held_layout = resolve_write_layout(&held, 0, 4, 2, None, false).expect("held snapshot should remain valid");
         assert_eq!(held_layout.parity_drives, 2);
-        assert!(held.should_inline(512, false));
+        assert!(held.should_inline(512, held_layout.data_drives, false));
 
         let current = published.load_full();
         let current_layout = resolve_write_layout(&current, 0, 4, 2, None, false).expect("new snapshot should resolve");
         assert_eq!(current_layout.parity_drives, 1);
-        assert!(!current.should_inline(512, false));
+        assert!(!current.should_inline(512, current_layout.data_drives, false));
     }
 }
 
@@ -3411,9 +3744,15 @@ fn collect_inline_data_shard_fileinfos_by_index<'a>(
         if block_index == 0 || block_index > data_shards {
             continue;
         }
+        if file_info.erasure.index != block_index {
+            continue;
+        }
         if !file_info.has_valid_erasure_geometry() {
             continue;
         }
+        if !core::io_primitives::metadata_early_stop_candidate_matches(file_info, fi) {
+            continue;
+        }
         if file_info.data.as_ref().is_none_or(|data| data.is_empty()) {
             continue;
         }
@@ -9221,6 +9560,9 @@ mod tests {
             HashAlgorithm::HighwayHash256S
         };
         let shards = erasure.encode_data(payload).expect("payload should encode");
+        let version_id = Some(Uuid::new_v4());
+        let data_dir = Some(Uuid::new_v4());
+        let mod_time = Some(OffsetDateTime::now_utc());
         let mut files = Vec::with_capacity(shards.len());
 
         for shard in shards {
@@ -9233,6 +9575,16 @@ mod tests {
             writer.shutdown().await.expect("inline writer should shutdown");
             let data = writer.into_inline_data().expect("inline data should be retained");
             let mut file = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
+            file.volume = "bucket".to_string();
+            file.name = "object".to_string();
+            file.size = i64::try_from(payload.len()).expect("test payload should fit i64");
+            file.is_latest = true;
+            file.version_id = version_id;
+            file.data_dir = data_dir;
+            file.mod_time = mod_time;
+            file.metadata.insert("etag".to_string(), "etag-inline".to_string());
+            file.add_object_part(1, "part-etag-inline".to_string(), payload.len(), file.mod_time, file.size, None, None);
+            file.set_inline_data();
             file.erasure.index = files.len() + 1;
             file.data = Some(Bytes::from(data));
             files.push(file);
@@ -9245,16 +9597,40 @@ mod tests {
         inline_bitrot_files_for_payload_with_mode(payload, false).await
     }
 
+    fn disk_ordered_fileinfos(files: &[FileInfo]) -> Vec<FileInfo> {
+        let distribution = &files
+            .first()
+            .expect("inline data shard fixture should include metadata")
+            .erasure
+            .distribution;
+        distribution
+            .iter()
+            .map(|block_index| {
+                files
+                    .get(block_index.checked_sub(1).expect("erasure block indexes are one-based"))
+                    .expect("inline data shard fixture should include every distributed shard")
+                    .clone()
+            })
+            .collect()
+    }
+
     fn inline_data_shard_fileinfo(
-        name: &str,
         data_blocks: usize,
         parity_blocks: usize,
         erasure_index: usize,
         distribution: &[usize],
         data: Option<&'static [u8]>,
     ) -> FileInfo {
-        let mut fi = FileInfo::new(name, data_blocks, parity_blocks);
-        fi.name = name.to_string();
+        let mut fi = FileInfo::new("object", data_blocks, parity_blocks);
+        fi.name = "object".to_string();
+        fi.volume = "bucket".to_string();
+        fi.size = 4;
+        fi.is_latest = true;
+        fi.data_dir = Some(Uuid::nil());
+        fi.mod_time = Some(OffsetDateTime::UNIX_EPOCH);
+        fi.metadata.insert("etag".to_string(), "etag-inline".to_string());
+        fi.add_object_part(1, "part-etag-inline".to_string(), 4, fi.mod_time, 4, None, None);
+        fi.set_inline_data();
         fi.erasure.index = erasure_index;
         fi.erasure.distribution = distribution.to_vec();
         fi.data = data.map(Bytes::from_static);
@@ -9264,36 +9640,41 @@ mod tests {
     #[test]
     fn collect_inline_data_shards_by_index_uses_distribution_order() {
         let distribution = vec![3, 1, 5, 2, 4, 6];
-        let mut fi = FileInfo::new("object", 4, 2);
+        let mut fi = inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"x"));
+        fi.erasure.index = 1;
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo("block-3", 4, 2, 3, &distribution, Some(b"c")),
-            inline_data_shard_fileinfo("block-1", 4, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo("parity-5", 4, 2, 5, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo("block-2", 4, 2, 2, &distribution, Some(b"b")),
-            inline_data_shard_fileinfo("block-4", 4, 2, 4, &distribution, Some(b"d")),
-            inline_data_shard_fileinfo("parity-6", 4, 2, 6, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo(4, 2, 3, &distribution, Some(b"c")),
+            inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo(4, 2, 5, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo(4, 2, 2, &distribution, Some(b"b")),
+            inline_data_shard_fileinfo(4, 2, 4, &distribution, Some(b"d")),
+            inline_data_shard_fileinfo(4, 2, 6, &distribution, Some(b"q")),
         ];
 
         let data_files =
             collect_inline_data_shard_fileinfos_by_index(&files, &fi, 4, |_| true).expect("all data shards should be collected");
 
         assert_eq!(
-            data_files.iter().map(|file| file.name.as_str()).collect::<Vec<_>>(),
-            ["block-1", "block-2", "block-3", "block-4"]
+            data_files
+                .iter()
+                .map(|file| file.data.as_deref().expect("fixture carries inline bytes"))
+                .collect::<Vec<_>>(),
+            [b"a".as_slice(), b"b".as_slice(), b"c".as_slice(), b"d".as_slice()]
         );
     }
 
     #[test]
     fn collect_inline_data_shards_by_index_rejects_missing_data_shard() {
         let distribution = vec![1, 2, 3, 4];
-        let mut fi = FileInfo::new("object", 2, 2);
+        let mut fi = inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"x"));
+        fi.erasure.index = 1;
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo("block-1", 2, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo("block-2", 2, 2, 2, &distribution, None),
-            inline_data_shard_fileinfo("parity-3", 2, 2, 3, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo("parity-4", 2, 2, 4, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo(2, 2, 2, &distribution, None),
+            inline_data_shard_fileinfo(2, 2, 3, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo(2, 2, 4, &distribution, Some(b"q")),
         ];
 
         assert!(collect_inline_data_shard_fileinfos_by_index(&files, &fi, 2, |_| true).is_none());
@@ -9426,10 +9807,8 @@ mod tests {
 
         let payload = vec![b'i'; 192 * 1024];
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
-        fi.size = payload.len() as i64;
-        fi.data = files[0].data.clone();
-        fi.add_object_part(1, String::new(), payload.len(), None, payload.len() as i64, None, None);
+        let fi = files[0].clone();
+        let disk_files = disk_ordered_fileinfos(&files);
 
         let disks = vec![Some(disk); erasure.total_shard_count()];
         let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(fi.size);
@@ -9438,7 +9817,7 @@ mod tests {
             "bucket",
             "object",
             &fi,
-            &files,
+            &disk_files,
             &disks,
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
@@ -9469,10 +9848,8 @@ mod tests {
         let payload = vec![b'v'; 64 * 1024];
         let payload_size = i64::try_from(payload.len()).expect("test payload size should fit i64");
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
-        fi.size = payload_size;
-        fi.data = files[0].data.clone();
-        fi.add_object_part(1, String::new(), payload.len(), None, payload_size, None, None);
+        let fi = files[0].clone();
+        let disk_files = disk_ordered_fileinfos(&files);
 
         let mut object_info = ObjectInfo {
             size: payload_size,
@@ -9503,7 +9880,7 @@ mod tests {
             "bucket",
             "object",
             &fi,
-            &files,
+            &disk_files,
             &vec![Some(disk); erasure.total_shard_count()],
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,

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

@@ -2296,140 +2296,157 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         }
         ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
-        let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
-
-        // Crash-consistency injection: hard power loss after the upload is fully
-        // staged and locked but before the authoritative rename_data commit. No
-        // disk has moved the staged data, so a crash here must leave any prior
-        // committed version byte-for-byte intact (rustfs/backlog#864) and the
-        // upload fully retryable. Compiles to a no-op outside `#[cfg(test)]`.
-        if crash_inject::should_crash_at(CrashPoint::MultipartBeforeCommitRename, object) {
-            return Err(StorageError::Unexpected);
-        }
-
-        // The trailing `_` drops the rename_data old-size backfill
-        // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
-        // `get_object_info` lookup, so the backfill has no consumer here yet.
-        let (online_disks, convergence, op_old_dir, cleanup_disks, _) = Self::rename_data(
-            &shuffle_disks,
-            RUSTFS_META_MULTIPART_BUCKET,
-            &upload_id_path,
-            &parts_metadatas,
-            bucket,
-            object,
-            write_quorum,
-        )
-        .await?;
-
-        // Detach admission before any post-commit await: client cancellation
-        // must not couple durable convergence repair to cleanup work.
-        if convergence.needs_heal() {
-            let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
-                bucket.to_string(),
-                Some(object.to_string()),
-                false,
-                Some(HealChannelPriority::Normal),
-                Some(self.pool_index),
-                Some(self.set_index),
-            );
-            request.object_version_id = fi
-                .version_id
-                .or_else(|| opts.version_suspended.then(Uuid::nil))
-                .map(|version_id| version_id.to_string());
-            tokio::spawn(async move {
-                let _ = rustfs_common::heal_channel::send_heal_request(request).await;
-            });
-        }
-
-        // Crash-consistency injection: hard power loss after the authoritative
-        // rename_data commit succeeded but before the stale part.N.meta cleanup.
-        // The new version is durably committed and visible, so a crash here must
-        // leave the object readable as the new version; the un-reclaimed staging
-        // parts are swept by a retried completion or upload GC (rustfs/backlog#946).
-        // Compiles to a no-op outside `#[cfg(test)]`.
-        if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object) {
-            return Err(StorageError::Unexpected);
-        }
-
-        // backlog#946: reclaim the stale per-part metadata (and any superfluous
-        // part.N data files no longer in the completed set) only *after* the
-        // authoritative rename_data commit above has succeeded. If rename_data
-        // fails write quorum and returns via `?`, the upload directory must keep
-        // its part.N.meta so a retried CompleteMultipartUpload can still read the
-        // parts; deleting them before the commit would strand the upload
-        // permanently. This mirrors the "clean up only after commit" pattern
-        // already used for the old data-dir GC and the upload-dir delete_all below.
-        self.cleanup_multipart_path(&parts).await;
-
-        if let Some(old_dir) = op_old_dir {
-            let committed_dir = fi.data_dir.unwrap_or_default().to_string();
-            // backlog#898: best-effort reclaim of the dereferenced old data dir.
-            // Returns a receipt (never `Err`); a failed GC must not turn an
-            // already-committed multipart completion into a 503.
-            let cleanup = self
-                .commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum)
-                .await;
-            self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup)
-                .await;
-        }
-
-        if let Some(stage_start) = complete_tail_stage_start {
-            rustfs_io_metrics::record_put_object_stage_duration(
-                "multipart_complete_tail",
-                stage_start.elapsed().as_secs_f64() * 1000.0,
-            );
-        }
-
-        #[cfg(test)]
-        pause_multipart_commit(bucket, object, MultipartCommitPause::AfterRename).await;
-
-        let cleanup_store = self.clone();
-        let cleanup_upload_id_path = upload_id_path.clone();
-        let cleanup_bucket = bucket.to_owned();
-        let cleanup_object = object.to_owned();
-        let cleanup_upload_id = upload_id.to_owned();
-        let cleanup_handle = tokio::spawn(async move {
+        let commit_set = self.clone();
+        let commit_bucket = bucket.to_owned();
+        let commit_object = object.to_owned();
+        let commit_upload_id = upload_id.to_owned();
+        let commit_upload_id_path = upload_id_path.clone();
+        let commit_version_suspended = opts.version_suspended;
+        let commit_is_versioned = opts.versioned || opts.version_suspended;
+        let commit_capacity_scope_token = opts.capacity_scope_token;
+        let commit_object_lock_guard = object_lock_guard.take();
+        let detach_commit_owner = commit_object_lock_guard.is_some() || upload_guard.is_some();
+        let commit = async move {
+            let _object_lock_guard = commit_object_lock_guard;
             let _upload_guard = upload_guard;
-            if let Err(err) = cleanup_store
-                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_id_path, write_quorum)
+            let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
+
+            // Crash-consistency injection: hard power loss after the upload is fully
+            // staged and locked but before the authoritative rename_data commit. No
+            // disk has moved the staged data, so a crash here must leave any prior
+            // committed version byte-for-byte intact (rustfs/backlog#864) and the
+            // upload fully retryable. Compiles to a no-op outside `#[cfg(test)]`.
+            if crash_inject::should_crash_at(CrashPoint::MultipartBeforeCommitRename, &commit_object) {
+                return Err(StorageError::Unexpected);
+            }
+
+            // The trailing `_` drops the rename_data old-size backfill
+            // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
+            // `get_object_info` lookup, so the backfill has no consumer here yet.
+            let (online_disks, convergence, op_old_dir, cleanup_disks, _) = SetDisks::rename_data(
+                &shuffle_disks,
+                RUSTFS_META_MULTIPART_BUCKET,
+                &commit_upload_id_path,
+                &parts_metadatas,
+                &commit_bucket,
+                &commit_object,
+                write_quorum,
+            )
+            .await?;
+
+            // Detach admission before any post-commit await: client cancellation
+            // must not couple durable convergence repair to cleanup work.
+            if convergence.needs_heal() {
+                let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
+                    commit_bucket.clone(),
+                    Some(commit_object.clone()),
+                    false,
+                    Some(HealChannelPriority::Normal),
+                    Some(commit_set.pool_index),
+                    Some(commit_set.set_index),
+                );
+                request.object_version_id = fi
+                    .version_id
+                    .or_else(|| commit_version_suspended.then(Uuid::nil))
+                    .map(|version_id| version_id.to_string());
+                tokio::spawn(async move {
+                    let _ = rustfs_common::heal_channel::send_heal_request(request).await;
+                });
+            }
+
+            // Crash-consistency injection: hard power loss after the authoritative
+            // rename_data commit succeeded but before the stale part.N.meta cleanup.
+            // The new version is durably committed and visible, so a crash here must
+            // leave the object readable as the new version; the un-reclaimed staging
+            // parts are swept by a retried completion or upload GC (rustfs/backlog#946).
+            // Compiles to a no-op outside `#[cfg(test)]`.
+            if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, &commit_object) {
+                return Err(StorageError::Unexpected);
+            }
+
+            // backlog#946: reclaim the stale per-part metadata (and any superfluous
+            // part.N data files no longer in the completed set) only *after* the
+            // authoritative rename_data commit above has succeeded. If rename_data
+            // fails write quorum and returns via `?`, the upload directory must keep
+            // its part.N.meta so a retried CompleteMultipartUpload can still read the
+            // parts; deleting them before the commit would strand the upload
+            // permanently. This mirrors the "clean up only after commit" pattern
+            // already used for the old data-dir GC and the upload-dir delete_all below.
+            commit_set.cleanup_multipart_path(&parts).await;
+
+            if let Some(old_dir) = op_old_dir {
+                let committed_dir = fi.data_dir.unwrap_or_default().to_string();
+                // backlog#898: best-effort reclaim of the dereferenced old data dir.
+                // Returns a receipt (never `Err`); a failed GC must not turn an
+                // already-committed multipart completion into a 503.
+                let cleanup = commit_set
+                    .commit_rename_data_dir(
+                        &cleanup_disks,
+                        &commit_bucket,
+                        &commit_object,
+                        &old_dir.to_string(),
+                        &committed_dir,
+                        write_quorum,
+                    )
+                    .await;
+                commit_set
+                    .report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
+                    .await;
+            }
+
+            if let Some(stage_start) = complete_tail_stage_start {
+                rustfs_io_metrics::record_put_object_stage_duration(
+                    "multipart_complete_tail",
+                    stage_start.elapsed().as_secs_f64() * 1000.0,
+                );
+            }
+
+            #[cfg(test)]
+            pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterRename).await;
+
+            if let Err(err) = commit_set
+                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum)
                 .await
             {
                 warn!(
-                    bucket = %cleanup_bucket,
-                    object = %cleanup_object,
-                    upload_id = %cleanup_upload_id,
+                    bucket = %commit_bucket,
+                    object = %commit_object,
+                    upload_id = %commit_upload_id,
                     error = ?err,
                     "completed multipart upload staging cleanup did not reach write quorum"
                 );
             }
-        });
-        if let Err(err) = cleanup_handle.await {
-            warn!(
-                bucket = %bucket,
-                object = %object,
-                upload_id = %upload_id,
-                error = ?err,
-                "completed multipart upload staging cleanup task failed"
-            );
-        }
-        drop(object_lock_guard); // drop object lock guard to release the lock
 
-        for (i, op_disk) in online_disks.iter().enumerate() {
-            if let Some(disk) = op_disk
-                && disk.is_online().await
-            {
-                fi = parts_metadatas[i].clone();
-                break;
+            for (i, op_disk) in online_disks.iter().enumerate() {
+                if let Some(disk) = op_disk
+                    && disk.is_online().await
+                {
+                    fi = parts_metadatas[i].clone();
+                    break;
+                }
             }
+
+            commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
+
+            fi.is_latest = true;
+
+            commit_set
+                .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
+                .await;
+
+            drop(_object_lock_guard); // drop object lock guard to release the lock
+            drop(_upload_guard);
+
+            Ok(ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned))
+        };
+
+        if detach_commit_owner {
+            tokio::spawn(commit)
+                .await
+                .map_err(|err| Error::other(format!("complete_multipart_upload commit task failed: {err}")))?
+        } else {
+            commit.await
         }
-
-        self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks);
-
-        fi.is_latest = true;
-
-        self.invalidate_get_object_metadata_cache(bucket, object).await;
-
-        Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
     }
 }
 
@@ -4883,6 +4900,74 @@ mod tests {
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn cancelled_complete_keeps_upload_lock_through_tail_cleanup() {
+        temp_env::async_with_vars(
+            [
+                (crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true")),
+                (rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("60")),
+            ],
+            async {
+                let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+                let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+                let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+                let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+                let bucket = "multipart-cancelled-tail-lock-bucket";
+                let object = "object";
+                make_bucket_on_all(&disk_stores, bucket).await;
+                let (upload_id, parts) =
+                    stage_upload_with_create_opts(&set_disks, bucket, object, &[0x53; 4096], &ObjectOptions::default()).await;
+                let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+                signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
+                signaling.clear_observed();
+                let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+                let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterRename);
+
+                let complete_store = set_disks.clone();
+                let complete_upload_id = upload_id.clone();
+                let complete = tokio::spawn(async move {
+                    complete_store
+                        .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
+                        .await
+                });
+                barrier.wait_until_paused().await;
+
+                let abort_store = set_disks.clone();
+                let abort_upload_id = upload_id.clone();
+                let abort = tokio::spawn(async move {
+                    abort_store
+                        .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
+                        .await
+                });
+                signaling.wait_for_attempts(2).await;
+                tokio::task::yield_now().await;
+                assert!(!abort.is_finished(), "abort must wait while completion tail owns the upload lock");
+
+                complete.abort();
+                assert!(
+                    complete
+                        .await
+                        .expect_err("the completion request should be cancellable while the tail is paused")
+                        .is_cancelled()
+                );
+                tokio::task::yield_now().await;
+                assert!(!abort.is_finished(), "cancelling the completion waiter must not release the upload lock");
+
+                barrier.release();
+                let abort_err = abort
+                    .await
+                    .expect("abort task should not panic")
+                    .expect_err("the committed upload should no longer exist when abort acquires the lock");
+                assert!(
+                    matches!(abort_err, StorageError::InvalidUploadID(..)),
+                    "abort should return InvalidUploadID after the detached completion tail, got {abort_err:?}"
+                );
+            },
+        )
+        .await;
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn complete_validates_parts_after_an_inflight_upload_part_commit() {

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

@@ -188,6 +188,74 @@ async fn get_object_reader_with_context(
     GetObjectReader::new_with_resolver(reader, range, object_info, opts, headers, ctx.object_encryption_resolver()).await
 }
 
+fn data_read_metadata_early_stop_request_shape_allowed(range: &Option<HTTPRangeSpec>, opts: &ObjectOptions) -> bool {
+    range.is_none()
+        && opts.part_number.is_none()
+        && opts.version_id.is_none()
+        && !opts.incl_free_versions
+        && !opts.skip_free_version
+        && !opts.raw_data_movement_read
+        && !opts.data_movement
+        && !crate::object_api::restore_request_active(opts)
+}
+
+#[cfg(test)]
+mod data_read_metadata_early_stop_request_shape_tests {
+    use super::*;
+
+    #[test]
+    fn data_read_metadata_early_stop_only_allows_whole_latest_plain_get_shape() {
+        assert!(data_read_metadata_early_stop_request_shape_allowed(&None, &ObjectOptions::default()));
+
+        let range = Some(HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 0,
+            end: 0,
+        });
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&range, &ObjectOptions::default()));
+
+        let part_opts = ObjectOptions {
+            part_number: Some(1),
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &part_opts));
+
+        let version_opts = ObjectOptions {
+            version_id: Some(Uuid::new_v4().to_string()),
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &version_opts));
+
+        let incl_free_opts = ObjectOptions {
+            incl_free_versions: true,
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &incl_free_opts));
+
+        let skip_free_opts = ObjectOptions {
+            skip_free_version: true,
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &skip_free_opts));
+
+        let data_movement_opts = ObjectOptions {
+            data_movement: true,
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &data_movement_opts));
+
+        let raw_data_movement_opts = ObjectOptions {
+            raw_data_movement_read: true,
+            ..Default::default()
+        };
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &raw_data_movement_opts));
+
+        let mut restore_opts = ObjectOptions::default();
+        restore_opts.transition.restore_request.days = Some(1);
+        assert!(!data_read_metadata_early_stop_request_shape_allowed(&None, &restore_opts));
+    }
+}
+
 /// Length of the full plaintext body when — and only when — this read's output
 /// is exactly the object's complete plaintext, so the app-layer body cache may
 /// serve it in place of the erasure read.
@@ -431,7 +499,16 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let (snapshot, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() {
             (prepared.snapshot, prepared.object_info)
         } else {
-            match self.get_object_fileinfo(bucket, object, opts, true, true).await {
+            match self
+                .get_object_fileinfo(
+                    bucket,
+                    object,
+                    opts,
+                    true,
+                    data_read_metadata_early_stop_request_shape_allowed(&range, opts),
+                )
+                .await
+            {
                 Ok(snapshot) => (snapshot, None),
                 Err(err) => {
                     rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_stage_start.elapsed().as_secs_f64());
@@ -1106,11 +1183,13 @@ impl SetDisks {
 
         let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap());
 
+        let mut tmp_cleanup_owned = false;
         let result: Result<(ObjectInfo, Option<OldCurrentSize>)> = async {
             let erasure = Arc::new(erasure_from_file_info(&fi, false)?);
 
             let put_object_size = known_put_object_storage_size(data.size());
-            let is_inline_buffer = storage_class_config.should_inline(erasure.shard_file_size(put_object_size), opts.versioned);
+            let is_inline_buffer =
+                storage_class_config.should_inline(erasure.shard_file_size(put_object_size), erasure.data_shards, opts.versioned);
 
             let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
             let shard_file_size = erasure.shard_file_size(put_object_size);
@@ -1597,169 +1676,236 @@ impl SetDisks {
                 });
             }
 
-            let rename_stage_start = Instant::now();
-            let (online_disks, convergence, op_old_dir, cleanup_disks, old_current_size) = Self::rename_data(
-                &shuffle_disks,
-                RUSTFS_META_TMP_BUCKET,
-                tmp_dir.as_str(),
-                &parts_metadatas,
-                bucket,
-                object,
-                write_quorum,
-            )
-            .await?;
-            // Do this before any post-commit await so request cancellation cannot
-            // bypass best-effort admission. A process crash before admission
-            // remains subject to the existing scanner reconciliation path.
-            if convergence.needs_heal() {
-                let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
-                    bucket.to_string(),
-                    Some(object.to_string()),
-                    false,
-                    Some(HealChannelPriority::Normal),
-                    Some(self.pool_index),
-                    Some(self.set_index),
-                );
-                request.object_version_id = committed_version_id.map(|version_id| version_id.to_string());
-                tokio::spawn(async move {
-                    let _ = rustfs_common::heal_channel::send_heal_request(request).await;
-                });
-            }
+            let commit_set = self.clone();
+            let commit_bucket = bucket.to_owned();
+            let commit_object = object.to_owned();
+            let commit_tmp_dir = tmp_dir.clone();
+            let commit_object_lock_guard = object_lock_guard.take();
+            let commit_bucket_lifecycle_guard = bucket_lifecycle_guard.take();
+            let detach_commit_owner = commit_object_lock_guard.is_some() || commit_bucket_lifecycle_guard.is_some();
+            let commit_write_path_label = write_path.metric_label();
+            let commit_is_versioned = opts.versioned || opts.version_suspended;
+            let commit_capacity_scope_token = opts.capacity_scope_token;
+            let commit_replication_state = replication_state_to_filemeta(&opts.put_replication_state());
+            tmp_cleanup_owned = true;
 
-            let rename_stage_elapsed = rename_stage_start.elapsed();
-            let rename_stage_ms = rename_stage_elapsed.as_millis() as u64;
-
-            self.invalidate_get_object_metadata_cache(bucket, object).await;
-
-            // `rename_data` has completed the authoritative quorum commit. The
-            // exact old-data-dir reclamation below is best-effort space cleanup;
-            // it must not serialize the next operation on this object.
-            drop(object_lock_guard);
-
-            rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", duration_millis_f64(rename_stage_elapsed));
-            if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
-                warn!(
-                    event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_SET_DISK,
-                    stage = "rename_data",
-                    bucket = %bucket,
-                    object = %object,
-                    tmp_dir = %tmp_dir,
-                    duration_ms = { rename_stage_ms },
+            let commit = async move {
+                let _object_lock_guard = commit_object_lock_guard;
+                let _bucket_lifecycle_guard = commit_bucket_lifecycle_guard;
+                let rename_stage_start = Instant::now();
+                let rename_result = SetDisks::rename_data(
+                    &shuffle_disks,
+                    RUSTFS_META_TMP_BUCKET,
+                    commit_tmp_dir.as_str(),
+                    &parts_metadatas,
+                    &commit_bucket,
+                    &commit_object,
                     write_quorum,
-                    state = "slow",
-                    "SetDisk commit tail stage is slow"
-                );
-            }
+                )
+                .await;
+                let (online_disks, convergence, op_old_dir, cleanup_disks, old_current_size) = match rename_result {
+                    Ok(commit) => commit,
+                    Err(err) => {
+                        if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await {
+                            warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data");
+                        } else if issue3031_diag_enabled() {
+                            warn!(
+                                target: "rustfs_ecstore::set_disk",
+                                bucket = %commit_bucket,
+                                object = %commit_object,
+                                tmp_dir = %commit_tmp_dir,
+                                "issue3031_put_object_tmp_cleanup_done"
+                            );
+                        }
+                        return Err(err.into());
+                    }
+                };
+                // Do this before any post-commit await so request cancellation cannot
+                // bypass best-effort admission. A process crash before admission
+                // remains subject to the existing scanner reconciliation path.
+                if convergence.needs_heal() {
+                    let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
+                        commit_bucket.clone(),
+                        Some(commit_object.clone()),
+                        false,
+                        Some(HealChannelPriority::Normal),
+                        Some(commit_set.pool_index),
+                        Some(commit_set.set_index),
+                    );
+                    request.object_version_id = committed_version_id.map(|version_id| version_id.to_string());
+                    tokio::spawn(async move {
+                        let _ = rustfs_common::heal_channel::send_heal_request(request).await;
+                    });
+                }
 
-            let mut cleanup_stage_ms: Option<u64> = None;
-            if let Some(old_dir) = op_old_dir {
-                let committed_dir = committed_data_dir.unwrap_or_default().to_string();
-                let cleanup_stage_start = Instant::now();
-                // backlog#898: reclaiming the dereferenced old data dir is
-                // best-effort and returns a receipt (never `Err`). A failed GC
-                // here must not negate an already-committed, durable write, so we
-                // deliberately do NOT `?`-propagate it into a 503. On residue the
-                // report path emits the leak metric and enqueues a heal.
-                let cleanup = self
-                    .commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum)
+                let rename_stage_elapsed = rename_stage_start.elapsed();
+                let rename_stage_ms = rename_stage_elapsed.as_millis() as u64;
+
+                commit_set
+                    .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
                     .await;
-                let cleanup_elapsed = cleanup_stage_start.elapsed();
-                let cleanup_ms = cleanup_elapsed.as_millis() as u64;
-                cleanup_stage_ms = Some(cleanup_ms);
-                rustfs_io_metrics::record_put_object_stage_duration(
-                    "set_disk_old_data_cleanup",
-                    duration_millis_f64(cleanup_elapsed),
-                );
-                self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup)
-                    .await;
-                if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
+
+                // `rename_data` has completed the authoritative quorum commit. The
+                // exact old-data-dir reclamation below is best-effort space cleanup;
+                // it must not serialize the next operation on this object.
+                drop(_object_lock_guard);
+                drop(_bucket_lifecycle_guard);
+
+                rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", duration_millis_f64(rename_stage_elapsed));
+                if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
                     warn!(
                         event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
                         component = LOG_COMPONENT_ECSTORE,
                         subsystem = LOG_SUBSYSTEM_SET_DISK,
-                        stage = "commit_rename_data_dir",
-                        bucket = %bucket,
-                        object = %object,
-                        tmp_dir = %tmp_dir,
-                        old_dir = %old_dir,
-                        duration_ms = cleanup_ms,
+                        stage = "rename_data",
+                        bucket = %commit_bucket,
+                        object = %commit_object,
+                        tmp_dir = %commit_tmp_dir,
+                        duration_ms = { rename_stage_ms },
                         write_quorum,
                         state = "slow",
                         "SetDisk commit tail stage is slow"
                     );
                 }
+
+                let mut cleanup_stage_ms: Option<u64> = None;
+                if let Some(old_dir) = op_old_dir {
+                    let committed_dir = committed_data_dir.unwrap_or_default().to_string();
+                    let cleanup_stage_start = Instant::now();
+                    // backlog#898: reclaiming the dereferenced old data dir is
+                    // best-effort and returns a receipt (never `Err`). A failed GC
+                    // here must not negate an already-committed, durable write, so we
+                    // deliberately do NOT `?`-propagate it into a 503. On residue the
+                    // report path emits the leak metric and enqueues a heal.
+                    let cleanup = commit_set
+                        .commit_rename_data_dir(
+                            &cleanup_disks,
+                            &commit_bucket,
+                            &commit_object,
+                            &old_dir.to_string(),
+                            &committed_dir,
+                            write_quorum,
+                        )
+                        .await;
+                    let cleanup_elapsed = cleanup_stage_start.elapsed();
+                    let cleanup_ms = cleanup_elapsed.as_millis() as u64;
+                    cleanup_stage_ms = Some(cleanup_ms);
+                    rustfs_io_metrics::record_put_object_stage_duration(
+                        "set_disk_old_data_cleanup",
+                        duration_millis_f64(cleanup_elapsed),
+                    );
+                    commit_set
+                        .report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
+                        .await;
+                    if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
+                        warn!(
+                            event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
+                            component = LOG_COMPONENT_ECSTORE,
+                            subsystem = LOG_SUBSYSTEM_SET_DISK,
+                            stage = "commit_rename_data_dir",
+                            bucket = %commit_bucket,
+                            object = %commit_object,
+                            tmp_dir = %commit_tmp_dir,
+                            old_dir = %old_dir,
+                            duration_ms = cleanup_ms,
+                            write_quorum,
+                            state = "slow",
+                            "SetDisk commit tail stage is slow"
+                        );
+                    }
+                }
+
+                let committed_metadata_slot = committed_response_metadata_slot(&online_disks, response_metadata_slot);
+                let mut fi = std::mem::take(&mut parts_metadatas[committed_metadata_slot]);
+
+                if is_compressed {
+                    record_compression_total_memory(actual_size as u64, w_size as u64).await;
+                }
+                commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
+
+                fi.replication_state_internal = Some(commit_replication_state);
+
+                fi.is_latest = true;
+
+                if issue3031_diag_enabled() {
+                    let online_success_count = online_disks.iter().filter(|disk| disk.is_some()).count();
+                    warn!(
+                        target: "rustfs_ecstore::set_disk",
+                        bucket = %commit_bucket,
+                        object = %commit_object,
+                        tmp_dir = %commit_tmp_dir,
+                        data_dir = ?fi.data_dir,
+                        write_quorum,
+                        online_success_count,
+                        op_old_dir = ?op_old_dir,
+                        "issue3031_put_object_commit_succeeded"
+                    );
+                }
+
+                let total_commit_tail_ms = rename_stage_start.elapsed().as_millis();
+                if total_commit_tail_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
+                    warn!(
+                        event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_SET_DISK,
+                        stage = "put_object_commit_tail",
+                        bucket = %commit_bucket,
+                        object = %commit_object,
+                        tmp_dir = %commit_tmp_dir,
+                        duration_ms = total_commit_tail_ms as u64,
+                        write_quorum,
+                        state = "slow",
+                        "SetDisk commit tail is slow"
+                    );
+                }
+
+                if issue3031_diag_enabled() {
+                    warn!(
+                        event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_SET_DISK,
+                        bucket = %commit_bucket,
+                        object = %commit_object,
+                        write_quorum,
+                        write_path = commit_write_path_label,
+                        writer_setup_ms,
+                        encode_ms,
+                        rename_ms = rename_stage_ms,
+                        cleanup_ms = cleanup_stage_ms.unwrap_or_default(),
+                        cleanup_present = cleanup_stage_ms.is_some(),
+                        commit_tail_ms = total_commit_tail_ms as u64,
+                        result = "success",
+                        "SetDisk put_object stage summary"
+                    );
+                }
+
+                let cleanup_set = commit_set.clone();
+                let cleanup_tmp_dir = commit_tmp_dir.clone();
+                tokio::spawn(async move {
+                    if let Err(err) = cleanup_set.delete_all(RUSTFS_META_TMP_BUCKET, &cleanup_tmp_dir).await {
+                        warn!(tmp_dir = %cleanup_tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
+                    } else if issue3031_diag_enabled() {
+                        warn!(
+                            target: "rustfs_ecstore::set_disk",
+                            tmp_dir = %cleanup_tmp_dir,
+                            "issue3031_put_object_tmp_cleanup_done"
+                        );
+                    }
+                });
+
+                Ok((
+                    ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned),
+                    old_current_size,
+                ))
+            };
+
+            if detach_commit_owner {
+                tokio::spawn(commit)
+                    .await
+                    .map_err(|err| Error::other(format!("put_object commit task failed: {err}")))?
+            } else {
+                commit.await
             }
-
-            let committed_metadata_slot = committed_response_metadata_slot(&online_disks, response_metadata_slot);
-            let mut fi = std::mem::take(&mut parts_metadatas[committed_metadata_slot]);
-
-            if is_compressed {
-                record_compression_total_memory(actual_size as u64, w_size as u64).await;
-            }
-            self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks);
-
-            fi.replication_state_internal = Some(replication_state_to_filemeta(&opts.put_replication_state()));
-
-            fi.is_latest = true;
-
-            if issue3031_diag_enabled() {
-                let online_success_count = online_disks.iter().filter(|disk| disk.is_some()).count();
-                warn!(
-                    target: "rustfs_ecstore::set_disk",
-                    bucket = %bucket,
-                    object = %object,
-                    tmp_dir = %tmp_dir,
-                    data_dir = ?fi.data_dir,
-                    write_quorum,
-                    online_success_count,
-                    op_old_dir = ?op_old_dir,
-                    "issue3031_put_object_commit_succeeded"
-                );
-            }
-
-            let total_commit_tail_ms = rename_stage_start.elapsed().as_millis();
-            if total_commit_tail_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
-                warn!(
-                    event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_SET_DISK,
-                    stage = "put_object_commit_tail",
-                    bucket = %bucket,
-                    object = %object,
-                    tmp_dir = %tmp_dir,
-                    duration_ms = total_commit_tail_ms as u64,
-                    write_quorum,
-                    state = "slow",
-                    "SetDisk commit tail is slow"
-                );
-            }
-
-            if issue3031_diag_enabled() {
-                warn!(
-                    event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_SET_DISK,
-                    bucket = %bucket,
-                    object = %object,
-                    write_quorum,
-                    write_path = write_path.metric_label(),
-                    writer_setup_ms,
-                    encode_ms,
-                    rename_ms = rename_stage_ms,
-                    cleanup_ms = cleanup_stage_ms.unwrap_or_default(),
-                    cleanup_present = cleanup_stage_ms.is_some(),
-                    commit_tail_ms = total_commit_tail_ms as u64,
-                    result = "success",
-                    "SetDisk put_object stage summary"
-                );
-            }
-
-            Ok((
-                ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended),
-                old_current_size,
-            ))
         }
         .await;
 
@@ -1795,7 +1941,8 @@ impl SetDisks {
             );
         }
 
-        if result.is_ok() {
+        if tmp_cleanup_owned && result.is_ok() {
+        } else if result.is_ok() {
             // Success path: `rename_data` has already moved the data dir out of
             // the tmp workspace and removed the (empty) tmp dir where it could,
             // so this delete_all is a speculative safety net that normally hits
@@ -5775,8 +5922,10 @@ pub(in crate::set_disk::ops) mod hermetic_set_disks_support {
 mod inline_put_commit_path_tests {
     use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
     use super::*;
+    use crate::config::storageclass::lookup_config_for_pools_without_env;
     use crate::disk::{DiskAPI as _, ReadOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
+    use rustfs_config::server_config::KVS;
     use tokio::io::AsyncReadExt;
 
     async fn make_bucket(disks: &[DiskStore], bucket: &str) {
@@ -5840,6 +5989,91 @@ mod inline_put_commit_path_tests {
         assert_eq!(restored, payload);
     }
 
+    #[tokio::test]
+    async fn ec_8_4_default_budget_keeps_large_inline_candidate_out_of_xl_meta() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(12).await;
+        set_disks.set_test_storage_class_config(
+            lookup_config_for_pools_without_env(&KVS::new(), &[12]).expect("EC8+4 storage class should resolve"),
+        );
+        let bucket = "ec-8-4-inline-budget";
+        let object = "object.bin";
+        let payload = vec![0x5c; 300 * 1024];
+        make_bucket(&disk_stores, bucket).await;
+
+        let mut reader = PutObjReader::from_vec(payload.clone());
+        set_disks
+            .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("EC8+4 PUT should commit through the non-inline path");
+
+        for (disk_index, disk) in disk_stores.iter().enumerate() {
+            let file_info = disk
+                .read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .unwrap_or_else(|err| panic!("disk {disk_index} should persist EC8+4 metadata: {err}"));
+            assert_eq!(file_info.erasure.data_blocks, 8);
+            assert_eq!(file_info.erasure.parity_blocks, 4);
+            assert!(!file_info.inline_data(), "disk {disk_index} must keep the shard outside xl.meta");
+        }
+
+        let mut object_reader = set_disks
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("non-inline EC8+4 object should remain readable");
+        let mut restored = Vec::new();
+        object_reader
+            .stream
+            .read_to_end(&mut restored)
+            .await
+            .expect("non-inline EC8+4 object should stream");
+        assert_eq!(restored, payload);
+    }
+
+    #[tokio::test]
+    async fn ec_8_4_versioned_budget_reaches_put_placement_decision() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(12).await;
+        set_disks.set_test_storage_class_config(
+            lookup_config_for_pools_without_env(&KVS::new(), &[12]).expect("EC8+4 storage class should resolve"),
+        );
+        let bucket = "ec-8-4-versioned-inline-budget";
+        let object = "object.bin";
+        let payload = vec![0x73; 64 * 1024];
+        make_bucket(&disk_stores, bucket).await;
+
+        let options = ObjectOptions {
+            versioned: true,
+            ..Default::default()
+        };
+        let mut reader = PutObjReader::from_vec(payload.clone());
+        set_disks
+            .put_object(bucket, object, &mut reader, &options)
+            .await
+            .expect("versioned EC8+4 PUT should use the reduced inline budget");
+
+        for (disk_index, disk) in disk_stores.iter().enumerate() {
+            let file_info = disk
+                .read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .unwrap_or_else(|err| panic!("disk {disk_index} should persist versioned EC8+4 metadata: {err}"));
+            assert!(
+                !file_info.inline_data(),
+                "disk {disk_index} must keep the versioned shard outside xl.meta"
+            );
+        }
+
+        let mut object_reader = set_disks
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &options)
+            .await
+            .expect("versioned non-inline EC8+4 object should remain readable");
+        let mut restored = Vec::new();
+        object_reader
+            .stream
+            .read_to_end(&mut restored)
+            .await
+            .expect("versioned non-inline EC8+4 object should stream");
+        assert_eq!(restored, payload);
+    }
+
     #[tokio::test]
     async fn inline_put_direct_commit_accepts_exact_quorum_and_rejects_quorum_minus_one() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -6033,7 +6267,10 @@ mod inline_put_commit_path_tests {
 mod get_object_downstream_close_accounting_tests {
     use super::hermetic_set_disks_support::hermetic_set_disks;
     use super::*;
-    use crate::diagnostics::get::{GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT, GetObjectFailureReason};
+    use crate::diagnostics::get::{
+        GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST, GET_OBJECT_PATH_INTERNAL_META, GET_STAGE_DECODE, GET_STAGE_EMIT,
+        GetObjectFailureReason,
+    };
     use crate::disk::RUSTFS_META_BUCKET;
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
@@ -6152,7 +6389,22 @@ mod get_object_downstream_close_accounting_tests {
         let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
         rustfs_io_metrics::set_get_stage_metrics_enabled(true);
 
-        let (internal_missing, legacy_unknown, internal_fanout, legacy_fanout) = metrics::with_local_recorder(&recorder, || {
+        let (
+            internal_missing,
+            legacy_unknown,
+            internal_fanout,
+            legacy_fanout,
+            internal_scheduled,
+            legacy_scheduled,
+            internal_completed,
+            legacy_completed,
+            internal_cancelled,
+            legacy_cancelled,
+            internal_unsafe_miss,
+            legacy_unsafe_miss,
+            internal_saved,
+            legacy_saved,
+        ) = metrics::with_local_recorder(&recorder, || {
             runtime.block_on(async {
                 let (_temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
                 let options = ObjectOptions {
@@ -6196,6 +6448,54 @@ mod get_object_downstream_close_accounting_tests {
                         "rustfs_io_get_object_metadata_fanout_error_responses",
                         &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
                     ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_scheduled",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_scheduled",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_completed",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_completed",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_cancelled",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_fanout_cancelled",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
+                    ),
+                    recorder.counter_value(
+                        "rustfs_io_get_object_metadata_early_stop_total",
+                        &[
+                            ("path", GET_OBJECT_PATH_INTERNAL_META),
+                            ("decision", "miss"),
+                            ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
+                        ],
+                    ),
+                    recorder.counter_value(
+                        "rustfs_io_get_object_metadata_early_stop_total",
+                        &[
+                            ("path", GET_OBJECT_PATH_LEGACY_DUPLEX),
+                            ("decision", "miss"),
+                            ("reason", GET_METADATA_EARLY_STOP_REASON_UNSAFE_REQUEST),
+                        ],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_early_stop_saved_responses",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META)],
+                    ),
+                    recorder.histogram_values(
+                        "rustfs_io_get_object_metadata_early_stop_saved_responses",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)],
+                    ),
                 )
             })
         });
@@ -6208,6 +6508,50 @@ mod get_object_downstream_close_accounting_tests {
         );
         assert_eq!(internal_fanout, vec![4.0], "internal metadata fanout must retain its path label");
         assert!(legacy_fanout.is_empty(), "internal metadata fanout must not leak into legacy_duplex");
+        assert_eq!(
+            internal_scheduled,
+            vec![4.0],
+            "internal metadata lifecycle scheduled count must retain its path label"
+        );
+        assert!(
+            legacy_scheduled.is_empty(),
+            "internal metadata lifecycle scheduled count must not leak into legacy_duplex"
+        );
+        assert_eq!(
+            internal_completed,
+            vec![4.0],
+            "internal metadata lifecycle completed count must retain its path label"
+        );
+        assert!(
+            legacy_completed.is_empty(),
+            "internal metadata lifecycle completed count must not leak into legacy_duplex"
+        );
+        assert_eq!(
+            internal_cancelled,
+            vec![0.0],
+            "internal metadata full-wait lifecycle must record zero cancellations"
+        );
+        assert!(
+            legacy_cancelled.is_empty(),
+            "internal metadata lifecycle cancelled count must not leak into legacy_duplex"
+        );
+        assert_eq!(
+            internal_unsafe_miss, 1,
+            "internal metadata unsafe early-stop miss must retain its path label"
+        );
+        assert_eq!(
+            legacy_unsafe_miss, 0,
+            "internal metadata unsafe early-stop miss must not leak into legacy_duplex"
+        );
+        assert_eq!(
+            internal_saved,
+            vec![0.0],
+            "internal metadata unsafe miss must record zero saved responses on internal_meta"
+        );
+        assert!(
+            legacy_saved.is_empty(),
+            "internal metadata unsafe miss saved responses must not leak into legacy_duplex"
+        );
     }
 }
 
@@ -9762,6 +10106,74 @@ mod put_object_tmp_cleanup_tests {
         assert_eq!(body, vec![b'2'; TEST_OBJECT_SIZE]);
     }
 
+    #[tokio::test]
+    async fn cancelled_rename_keeps_namespace_lock_until_publication() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "put-commit-lock-cancelled-rename";
+        let object = "commit-lock-cancelled-rename-object";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let rename_tasks = rename_fanout_barrier::observe_tasks(object);
+        let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let first_store = Arc::clone(&set_disks);
+        let first = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+            first_store
+                .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
+            .await
+            .expect("first PUT should pause during the authoritative rename");
+
+        let second_namespace_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
+        let second_store = Arc::clone(&set_disks);
+        let second = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![b'2'; TEST_OBJECT_SIZE]);
+            second_store
+                .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+                .await
+        });
+        second_namespace_barrier.release_and_wait_until_namespace_pending().await;
+
+        first.abort();
+        assert!(
+            first
+                .await
+                .expect_err("the first request should be cancelled while rename is parked")
+                .is_cancelled()
+        );
+        tokio::task::yield_now().await;
+        assert!(
+            !second.is_finished(),
+            "the second writer must remain blocked by the cancelled commit owner"
+        );
+
+        rename_barrier.release();
+        drop(rename_barrier);
+        tokio::time::timeout(Duration::from_secs(30), async {
+            while rename_tasks.running() != 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("the cancelled owner's rename fanout should drain");
+        second
+            .await
+            .expect("second overwrite task should join")
+            .expect("second overwrite should commit after the cancelled owner reaches publication");
+
+        let mut reader = set_disks
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("the latest overwrite should be readable");
+        let mut body = Vec::new();
+        reader.stream.read_to_end(&mut body).await.expect("latest body should drain");
+        assert_eq!(body, vec![b'2'; TEST_OBJECT_SIZE]);
+    }
+
     #[tokio::test]
     async fn put_object_no_lock_aborts_after_outer_namespace_lock_loss() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;

crates/heal/tests/endpoint_index_test.rs

@@ -14,6 +14,8 @@
 
 //! test endpoint index settings
 
+#![recursion_limit = "256"]
+
 use std::net::SocketAddr;
 use tempfile::TempDir;
 use tokio_util::sync::CancellationToken;

crates/heal/tests/heal_b5_versioned_regression_test.rs

@@ -22,6 +22,8 @@
 //! bucket-metadata-sys OnceCell) — under `cargo nextest` each test runs
 //! in its own process so the OnceCell never collides.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/heal/tests/heal_b920_subquorum_union_test.rs

@@ -21,6 +21,8 @@
 //! These drive the REAL `ECStoreHealStorage` + `ECStore` against real disks.
 //! Every test is `#[serial]`; under `cargo nextest` each runs in its own process.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::storage::{

crates/heal/tests/heal_integration_test.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/io-metrics/src/lib.rs

@@ -812,6 +812,17 @@ pub fn record_get_object_metadata_fanout_shape(path: &'static str, total: usize,
         .record(metadata_fanout_count_to_f64(non_valid));
 }
 
+/// Record task lifecycle shape for one GetObject metadata fanout.
+#[inline(always)]
+pub fn record_get_object_metadata_fanout_lifecycle(path: &'static str, scheduled: usize, completed: usize, cancelled: usize) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
+    histogram!("rustfs_io_get_object_metadata_fanout_scheduled", "path" => path).record(metadata_fanout_count_to_f64(scheduled));
+    histogram!("rustfs_io_get_object_metadata_fanout_completed", "path" => path).record(metadata_fanout_count_to_f64(completed));
+    histogram!("rustfs_io_get_object_metadata_fanout_cancelled", "path" => path).record(metadata_fanout_count_to_f64(cancelled));
+}
+
 /// Record a guarded metadata early-stop hit for GetObject.
 #[inline(always)]
 pub fn record_get_object_metadata_early_stop_hit(path: &'static str, reason: &'static str) {
@@ -2698,6 +2709,7 @@ mod tests {
         record_get_object_quorum_reached_latency("legacy_duplex", 0.002);
         record_get_object_metadata_response("legacy_duplex", "valid");
         record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1);
+        record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
         record_get_object_metadata_early_stop_hit("legacy_duplex", "valid_quorum");
         record_get_object_metadata_early_stop_miss("legacy_duplex", "insufficient_quorum");
         record_get_object_metadata_early_stop_saved_responses("legacy_duplex", 1);
@@ -2768,6 +2780,38 @@ mod tests {
         assert!(remote_scheduled >= remote_avoid_potential);
     }
 
+    #[test]
+    fn metadata_fanout_lifecycle_records_named_histograms() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+
+        metrics::with_local_recorder(&recorder, || {
+            set_get_stage_metrics_enabled(true);
+            record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
+            set_get_stage_metrics_enabled(false);
+        });
+
+        let metrics = snapshotter.snapshot().into_vec();
+        for (name, expected) in [
+            ("rustfs_io_get_object_metadata_fanout_scheduled", 4.0),
+            ("rustfs_io_get_object_metadata_fanout_completed", 3.0),
+            ("rustfs_io_get_object_metadata_fanout_cancelled", 1.0),
+        ] {
+            let value = metrics.iter().find_map(|(composite, _, _, value)| {
+                let has_path = composite
+                    .key()
+                    .labels()
+                    .any(|label| label.key() == "path" && label.value() == "legacy_duplex");
+                (composite.kind() == MetricKind::Histogram && composite.key().name() == name && has_path).then_some(value)
+            });
+            assert!(
+                matches!(value, Some(DebugValue::Histogram(values)) if values.len() == 1 && values[0].0 == expected),
+                "{name} must record the exact fanout lifecycle sample"
+            );
+        }
+    }
+
     #[test]
     fn test_record_get_object_fill_metrics() {
         record_get_object_fill_queued("codec_streaming", "single_inflight", 1);

crates/protocols/tests/swift_metadata_persistence.rs

@@ -23,6 +23,7 @@
 //! two are tested together because a reload is the only way to tell a real
 //! merge from one that happened to look right in the cache.
 
+#![recursion_limit = "256"]
 #![cfg(feature = "swift")]
 
 use std::collections::HashMap;

crates/scanner/src/lib.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 #![warn(
     // missing_docs,

crates/scanner/tests/lifecycle_integration_test.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 use futures::FutureExt;
 use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT;
 use rustfs_scanner::scanner_folder::ScannerItem;

rustfs/src/admin/handlers/table_catalog/tests.rs

@@ -11,6 +11,12 @@ use datafusion::{
 };
 use std::sync::Arc;
 
+use crate::table_catalog::test_support::{
+    manifest_avro_bytes as test_manifest_avro_bytes,
+    manifest_avro_bytes_with_nullable_sequences as test_manifest_avro_bytes_with_nullable_sequences,
+    manifest_list_avro_bytes as test_manifest_list_avro_bytes, manifest_list_avro_entries as test_manifest_list_avro_entries,
+    table_metadata_json as test_table_metadata_json,
+};
 use rustfs_iam::store::{Store as _, UserType};
 use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq};
 
@@ -8033,33 +8039,6 @@ fn trusted_table_commit_backend(
     TableCommitObjectBackend::trusted(backend.clone())
 }
 
-fn test_table_metadata_json(table_uuid: &str, location: &str) -> serde_json::Value {
-    serde_json::json!({
-        "format-version": 2,
-        "table-uuid": table_uuid,
-        "location": location,
-        "last-sequence-number": 0,
-        "last-updated-ms": 1,
-        "last-column-id": 1,
-        "schemas": [{
-            "type": "struct",
-            "schema-id": 0,
-            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
-        }],
-        "current-schema-id": 0,
-        "partition-specs": [{"spec-id": 0, "fields": []}],
-        "default-spec-id": 0,
-        "last-partition-id": 999,
-        "sort-orders": [{"order-id": 0, "fields": []}],
-        "default-sort-order-id": 0,
-        "properties": {},
-        "snapshots": [],
-        "snapshot-log": [],
-        "metadata-log": [],
-        "refs": {}
-    })
-}
-
 fn test_snapshot_object_key(bucket: &str, location: &str) -> String {
     crate::table_catalog::table_catalog_object_key_from_location(bucket, location)
         .expect("test snapshot object location should be valid")
@@ -8078,184 +8057,6 @@ fn test_parquet_i32_bytes(values: &[i32]) -> Vec<u8> {
     bytes
 }
 
-fn test_manifest_list_avro_bytes(manifest_paths: &[&str], sequence_number: i64, snapshot_id: i64) -> Vec<u8> {
-    let manifests = manifest_paths
-        .iter()
-        .map(|manifest_path| (*manifest_path, 0, sequence_number, snapshot_id))
-        .collect::<Vec<_>>();
-    test_manifest_list_avro_entries_with_partition_specs(&manifests)
-}
-
-fn test_manifest_list_avro_entries(manifests: &[(&str, i64, i64)]) -> Vec<u8> {
-    let manifests = manifests
-        .iter()
-        .map(|(manifest_path, sequence_number, snapshot_id)| (*manifest_path, 0, *sequence_number, *snapshot_id))
-        .collect::<Vec<_>>();
-    test_manifest_list_avro_entries_with_partition_specs(&manifests)
-}
-
-fn test_manifest_list_avro_entries_with_partition_specs(manifests: &[(&str, i32, i64, i64)]) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_file",
-              "fields": [
-                {"name": "manifest_path", "type": "string"},
-                {"name": "manifest_length", "type": "long"},
-                {"name": "partition_spec_id", "type": "int"},
-                {"name": "content", "type": "int"},
-                {"name": "sequence_number", "type": "long"},
-                {"name": "min_sequence_number", "type": "long"},
-                {"name": "added_snapshot_id", "type": "long"},
-                {"name": "added_files_count", "type": "int"},
-                {"name": "existing_files_count", "type": "int"},
-                {"name": "deleted_files_count", "type": "int"},
-                {"name": "added_rows_count", "type": "long"},
-                {"name": "existing_rows_count", "type": "long"},
-                {"name": "deleted_rows_count", "type": "long"}
-              ]
-            }
-            "#,
-    )
-    .expect("manifest list avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize");
-    for (manifest_path, partition_spec_id, sequence_number, snapshot_id) in manifests {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                (
-                    "manifest_path".to_string(),
-                    apache_avro::types::Value::String((*manifest_path).to_string()),
-                ),
-                ("manifest_length".to_string(), apache_avro::types::Value::Long(1)),
-                ("partition_spec_id".to_string(), apache_avro::types::Value::Int(*partition_spec_id)),
-                ("content".to_string(), apache_avro::types::Value::Int(0)),
-                ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
-                ("min_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
-                ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
-                ("added_files_count".to_string(), apache_avro::types::Value::Int(1)),
-                ("existing_files_count".to_string(), apache_avro::types::Value::Int(0)),
-                ("deleted_files_count".to_string(), apache_avro::types::Value::Int(0)),
-                ("added_rows_count".to_string(), apache_avro::types::Value::Long(1)),
-                ("existing_rows_count".to_string(), apache_avro::types::Value::Long(0)),
-                ("deleted_rows_count".to_string(), apache_avro::types::Value::Long(0)),
-            ]))
-            .expect("manifest list record should append");
-    }
-    writer.into_inner().expect("manifest list avro bytes should flush")
-}
-
-fn test_manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_entry",
-              "fields": [
-                {"name": "status", "type": "int"},
-                {"name": "snapshot_id", "type": "long"},
-                {"name": "sequence_number", "type": "long"},
-                {"name": "file_sequence_number", "type": "long"},
-                {
-                  "name": "data_file",
-                  "type": {
-                    "type": "record",
-                    "name": "data_file",
-                    "fields": [
-                      {"name": "content", "type": "int"},
-                      {"name": "file_path", "type": "string"},
-                      {"name": "record_count", "type": "long"},
-                      {"name": "file_size_in_bytes", "type": "long"}
-                    ]
-                  }
-                }
-              ]
-            }
-            "#,
-    )
-    .expect("manifest avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
-    for (file_path, content, status, snapshot_id, sequence_number) in files {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                ("status".to_string(), apache_avro::types::Value::Int(*status)),
-                ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
-                ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
-                ("file_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
-                (
-                    "data_file".to_string(),
-                    apache_avro::types::Value::Record(vec![
-                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
-                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
-                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
-                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
-                    ]),
-                ),
-            ]))
-            .expect("manifest record should append");
-    }
-    writer.into_inner().expect("manifest avro bytes should flush")
-}
-
-fn test_nullable_long(value: Option<i64>) -> apache_avro::types::Value {
-    match value {
-        Some(value) => apache_avro::types::Value::Union(1, Box::new(apache_avro::types::Value::Long(value))),
-        None => apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
-    }
-}
-
-fn test_manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i32, i64, Option<i64>)]) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_entry",
-              "fields": [
-                {"name": "status", "type": "int"},
-                {"name": "snapshot_id", "type": "long"},
-                {"name": "sequence_number", "type": ["null", "long"], "default": null},
-                {"name": "file_sequence_number", "type": ["null", "long"], "default": null},
-                {
-                  "name": "data_file",
-                  "type": {
-                    "type": "record",
-                    "name": "data_file",
-                    "fields": [
-                      {"name": "content", "type": "int"},
-                      {"name": "file_path", "type": "string"},
-                      {"name": "record_count", "type": "long"},
-                      {"name": "file_size_in_bytes", "type": "long"}
-                    ]
-                  }
-                }
-              ]
-            }
-            "#,
-    )
-    .expect("manifest avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
-    for (file_path, content, status, snapshot_id, sequence_number) in files {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                ("status".to_string(), apache_avro::types::Value::Int(*status)),
-                ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
-                ("sequence_number".to_string(), test_nullable_long(*sequence_number)),
-                ("file_sequence_number".to_string(), test_nullable_long(*sequence_number)),
-                (
-                    "data_file".to_string(),
-                    apache_avro::types::Value::Record(vec![
-                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
-                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
-                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
-                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
-                    ]),
-                ),
-            ]))
-            .expect("manifest record should append");
-    }
-    writer.into_inner().expect("manifest avro bytes should flush")
-}
-
 async fn seed_test_manifest_list(
     backend: &TestTableCatalogObjectBackend,
     bucket: &str,

rustfs/src/app/object_usecase.rs

@@ -86,7 +86,7 @@ use super::storage_api::object_usecase::options::{
     namespace_reserved_user_metadata, normalize_content_encoding_for_storage, preserve_unclassified_user_metadata,
     put_opts_with_replication_authorization, validate_archive_content_encoding,
 };
-use super::storage_api::object_usecase::request_context::{self, spawn_traced};
+use super::storage_api::object_usecase::request_context::{self, spawn_traced, spawn_traced_join};
 use super::storage_api::object_usecase::s3_api::multipart::parse_list_parts_params;
 use super::storage_api::object_usecase::set_disk::{
     get_lock_acquire_timeout, get_object_disk_read_timeout, is_valid_storage_class,
@@ -7506,31 +7506,50 @@ impl DefaultObjectUsecase {
         let cache_adapter = self.object_data_cache();
         let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await;
 
-        let oi = store
-            .copy_object(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts)
-            .await
-            .map_err(ApiError::from)?;
-        drop(_self_copy_lock_guard);
+        let copy_commit = spawn_traced_join({
+            let store = Arc::clone(&store);
+            let src_bucket = src_bucket.clone();
+            let src_key = src_key.clone();
+            let bucket = bucket.clone();
+            let key = key.clone();
+            let src_opts = src_opts.clone();
+            let dst_opts = dst_opts.clone();
+            async move {
+                let _source_bucket_lifecycle_guard = source_bucket_lifecycle_guard;
+                let _destination_bucket_lifecycle_guard_storage = destination_bucket_lifecycle_guard_storage;
+                let _self_copy_lock_guard = _self_copy_lock_guard;
 
-        // Reuse the single pre-commit replication decision (see `dsc` above) so
-        // the persisted pending marker and the schedule always agree, mirroring
-        // the PUT path.
-        if dsc.replicate_any() {
-            schedule_object_replication(oi.clone(), store.clone(), dsc).await;
-        }
+                let oi = store
+                    .copy_object(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts)
+                    .await
+                    .map_err(ApiError::from)?;
 
-        maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await;
-        let _ = invalidate_object_data_cache_after_copy_success(&cache_adapter, &bucket, &key).await;
+                // Reuse the single pre-commit replication decision (see `dsc` above) so
+                // the persisted pending marker and the schedule always agree, mirroring
+                // the PUT path.
+                if dsc.replicate_any() {
+                    schedule_object_replication(oi.clone(), Arc::clone(&store), dsc).await;
+                }
 
-        let dest_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
-        // Update quota tracking after successful copy
-        if has_bucket_metadata {
-            if dest_versioned {
-                record_bucket_object_version_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
-            } else {
-                record_bucket_object_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
+                maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await;
+                let _ = invalidate_object_data_cache_after_copy_success(&cache_adapter, &bucket, &key).await;
+
+                let dest_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
+                if has_bucket_metadata {
+                    if dest_versioned {
+                        record_bucket_object_version_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
+                    } else {
+                        record_bucket_object_write_memory(&bucket, previous_current_size, oi.size.max(0) as u64).await;
+                    }
+                }
+
+                rustfs_scanner::record_dirty_usage_bucket(&bucket);
+                Ok::<_, S3Error>((oi, dest_versioned))
             }
-        }
+        });
+        let (oi, dest_versioned) = copy_commit.await.map_err(|err| {
+            S3Error::with_message(S3ErrorCode::InternalError, format!("copy object commit owner task failed: {err}"))
+        })??;
 
         let raw_dest_version = oi.version_id.map(|v| v.to_string());
         let dest_version = if dest_versioned { raw_dest_version } else { None };
@@ -7578,7 +7597,7 @@ impl DefaultObjectUsecase {
             }
         }
         let copy_object_result = CopyObjectResult {
-            e_tag: oi.etag.map(|etag| to_s3s_etag(&etag)),
+            e_tag: oi.etag.as_ref().map(|etag| to_s3s_etag(etag)),
             last_modified: oi.mod_time.map(Timestamp::from),
             checksum_crc32: response_checksums.crc32,
             checksum_crc32c: response_checksums.crc32c,
@@ -7609,7 +7628,6 @@ impl DefaultObjectUsecase {
 
         let result = Ok(S3Response::new(output));
         let _ = helper.complete(&result);
-        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         result
     }
 

rustfs/src/app/storage_api.rs

@@ -998,7 +998,7 @@ pub(crate) mod options {
 }
 
 pub(crate) mod request_context {
-    pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced};
+    pub(crate) use crate::storage::storage_api::request_context_consumer::{RequestContext, spawn_traced, spawn_traced_join};
 }
 
 pub(crate) mod sse {

rustfs/src/storage/request_context.rs

@@ -257,6 +257,15 @@ where
     tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current()));
 }
 
+/// Spawn a request-internal task and return its join handle to the caller.
+pub fn spawn_traced_join<F>(fut: F) -> tokio::task::JoinHandle<F::Output>
+where
+    F: std::future::Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    tokio::spawn(tracing::Instrument::instrument(fut, tracing::Span::current()))
+}
+
 #[cfg(test)]
 #[allow(unused_imports)]
 mod tests {

rustfs/src/storage/storage_api.rs

@@ -203,7 +203,9 @@ pub(crate) mod options_consumer {
 }
 
 pub(crate) mod request_context_consumer {
-    pub(crate) use super::super::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced};
+    pub(crate) use super::super::request_context::{
+        RequestContext, extract_request_id_from_headers, spawn_traced, spawn_traced_join,
+    };
 }
 
 pub(crate) mod rpc_consumer {

rustfs/src/table_catalog/mod.rs

@@ -361,5 +361,8 @@ fn storage_error_to_catalog(action: &str, err: StorageError) -> TableCatalogStor
     }
 }
 
+#[cfg(test)]
+pub(crate) mod test_support;
+
 #[cfg(test)]
 mod tests;

rustfs/src/table_catalog/test_support.rs

@@ -0,0 +1,228 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Shared table-catalog test fixtures (backlog#1837).
+//!
+//! Pure data constructors for Iceberg metadata JSON and avro
+//! manifest-list/manifest bytes, shared by the store-level tests
+//! (`table_catalog/tests.rs`) and the admin handler tests
+//! (`admin/handlers/table_catalog/tests.rs`). The parameterized admin
+//! variants are canonical; the store tests wrap them with their historical
+//! fixed values (sequence 7 / snapshot 20), which keeps every produced byte
+//! identical to the pre-extraction fixtures.
+
+pub(crate) fn table_metadata_json(table_uuid: &str, location: &str) -> serde_json::Value {
+    serde_json::json!({
+        "format-version": 2,
+        "table-uuid": table_uuid,
+        "location": location,
+        "last-sequence-number": 0,
+        "last-updated-ms": 1,
+        "last-column-id": 1,
+        "schemas": [{
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+        }],
+        "current-schema-id": 0,
+        "partition-specs": [{"spec-id": 0, "fields": []}],
+        "default-spec-id": 0,
+        "last-partition-id": 999,
+        "sort-orders": [{"order-id": 0, "fields": []}],
+        "default-sort-order-id": 0,
+        "properties": {},
+        "snapshots": [],
+        "snapshot-log": [],
+        "metadata-log": [],
+        "refs": {}
+    })
+}
+
+pub(crate) fn manifest_list_avro_bytes(manifest_paths: &[&str], sequence_number: i64, snapshot_id: i64) -> Vec<u8> {
+    let manifests = manifest_paths
+        .iter()
+        .map(|manifest_path| (*manifest_path, 0, sequence_number, snapshot_id))
+        .collect::<Vec<_>>();
+    manifest_list_avro_entries_with_partition_specs(&manifests)
+}
+
+pub(crate) fn manifest_list_avro_entries(manifests: &[(&str, i64, i64)]) -> Vec<u8> {
+    let manifests = manifests
+        .iter()
+        .map(|(manifest_path, sequence_number, snapshot_id)| (*manifest_path, 0, *sequence_number, *snapshot_id))
+        .collect::<Vec<_>>();
+    manifest_list_avro_entries_with_partition_specs(&manifests)
+}
+
+pub(crate) fn manifest_list_avro_entries_with_partition_specs(manifests: &[(&str, i32, i64, i64)]) -> Vec<u8> {
+    let schema = apache_avro::Schema::parse_str(
+        r#"
+            {
+              "type": "record",
+              "name": "manifest_file",
+              "fields": [
+                {"name": "manifest_path", "type": "string"},
+                {"name": "manifest_length", "type": "long"},
+                {"name": "partition_spec_id", "type": "int"},
+                {"name": "content", "type": "int"},
+                {"name": "sequence_number", "type": "long"},
+                {"name": "min_sequence_number", "type": "long"},
+                {"name": "added_snapshot_id", "type": "long"},
+                {"name": "added_files_count", "type": "int"},
+                {"name": "existing_files_count", "type": "int"},
+                {"name": "deleted_files_count", "type": "int"},
+                {"name": "added_rows_count", "type": "long"},
+                {"name": "existing_rows_count", "type": "long"},
+                {"name": "deleted_rows_count", "type": "long"}
+              ]
+            }
+            "#,
+    )
+    .expect("manifest list avro schema should parse");
+    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize");
+    for (manifest_path, partition_spec_id, sequence_number, snapshot_id) in manifests {
+        writer
+            .append_value(apache_avro::types::Value::Record(vec![
+                (
+                    "manifest_path".to_string(),
+                    apache_avro::types::Value::String((*manifest_path).to_string()),
+                ),
+                ("manifest_length".to_string(), apache_avro::types::Value::Long(1)),
+                ("partition_spec_id".to_string(), apache_avro::types::Value::Int(*partition_spec_id)),
+                ("content".to_string(), apache_avro::types::Value::Int(0)),
+                ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                ("min_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
+                ("added_files_count".to_string(), apache_avro::types::Value::Int(1)),
+                ("existing_files_count".to_string(), apache_avro::types::Value::Int(0)),
+                ("deleted_files_count".to_string(), apache_avro::types::Value::Int(0)),
+                ("added_rows_count".to_string(), apache_avro::types::Value::Long(1)),
+                ("existing_rows_count".to_string(), apache_avro::types::Value::Long(0)),
+                ("deleted_rows_count".to_string(), apache_avro::types::Value::Long(0)),
+            ]))
+            .expect("manifest list record should append");
+    }
+    writer.into_inner().expect("manifest list avro bytes should flush")
+}
+
+pub(crate) fn manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec<u8> {
+    let schema = apache_avro::Schema::parse_str(
+        r#"
+            {
+              "type": "record",
+              "name": "manifest_entry",
+              "fields": [
+                {"name": "status", "type": "int"},
+                {"name": "snapshot_id", "type": "long"},
+                {"name": "sequence_number", "type": "long"},
+                {"name": "file_sequence_number", "type": "long"},
+                {
+                  "name": "data_file",
+                  "type": {
+                    "type": "record",
+                    "name": "data_file",
+                    "fields": [
+                      {"name": "content", "type": "int"},
+                      {"name": "file_path", "type": "string"},
+                      {"name": "record_count", "type": "long"},
+                      {"name": "file_size_in_bytes", "type": "long"}
+                    ]
+                  }
+                }
+              ]
+            }
+            "#,
+    )
+    .expect("manifest avro schema should parse");
+    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
+    for (file_path, content, status, snapshot_id, sequence_number) in files {
+        writer
+            .append_value(apache_avro::types::Value::Record(vec![
+                ("status".to_string(), apache_avro::types::Value::Int(*status)),
+                ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
+                ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                ("file_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                (
+                    "data_file".to_string(),
+                    apache_avro::types::Value::Record(vec![
+                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
+                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
+                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
+                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
+                    ]),
+                ),
+            ]))
+            .expect("manifest record should append");
+    }
+    writer.into_inner().expect("manifest avro bytes should flush")
+}
+
+pub(crate) fn nullable_long(value: Option<i64>) -> apache_avro::types::Value {
+    match value {
+        Some(value) => apache_avro::types::Value::Union(1, Box::new(apache_avro::types::Value::Long(value))),
+        None => apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
+    }
+}
+
+pub(crate) fn manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i32, i64, Option<i64>)]) -> Vec<u8> {
+    let schema = apache_avro::Schema::parse_str(
+        r#"
+            {
+              "type": "record",
+              "name": "manifest_entry",
+              "fields": [
+                {"name": "status", "type": "int"},
+                {"name": "snapshot_id", "type": "long"},
+                {"name": "sequence_number", "type": ["null", "long"], "default": null},
+                {"name": "file_sequence_number", "type": ["null", "long"], "default": null},
+                {
+                  "name": "data_file",
+                  "type": {
+                    "type": "record",
+                    "name": "data_file",
+                    "fields": [
+                      {"name": "content", "type": "int"},
+                      {"name": "file_path", "type": "string"},
+                      {"name": "record_count", "type": "long"},
+                      {"name": "file_size_in_bytes", "type": "long"}
+                    ]
+                  }
+                }
+              ]
+            }
+            "#,
+    )
+    .expect("manifest avro schema should parse");
+    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
+    for (file_path, content, status, snapshot_id, sequence_number) in files {
+        writer
+            .append_value(apache_avro::types::Value::Record(vec![
+                ("status".to_string(), apache_avro::types::Value::Int(*status)),
+                ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
+                ("sequence_number".to_string(), nullable_long(*sequence_number)),
+                ("file_sequence_number".to_string(), nullable_long(*sequence_number)),
+                (
+                    "data_file".to_string(),
+                    apache_avro::types::Value::Record(vec![
+                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
+                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
+                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
+                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
+                    ]),
+                ),
+            ]))
+            .expect("manifest record should append");
+    }
+    writer.into_inner().expect("manifest avro bytes should flush")
+}

rustfs/src/table_catalog/tests.rs

@@ -1429,54 +1429,12 @@ fn manifest_list_avro_bytes(manifest_paths: &[&str]) -> Vec<u8> {
 }
 
 fn manifest_list_avro_bytes_with_spec(manifest_paths: &[&str], partition_spec_id: i32) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_file",
-              "fields": [
-                {"name": "manifest_path", "type": "string"},
-                {"name": "manifest_length", "type": "long"},
-                {"name": "partition_spec_id", "type": "int"},
-                {"name": "content", "type": "int"},
-                {"name": "sequence_number", "type": "long"},
-                {"name": "min_sequence_number", "type": "long"},
-                {"name": "added_snapshot_id", "type": "long"},
-                {"name": "added_files_count", "type": "int"},
-                {"name": "existing_files_count", "type": "int"},
-                {"name": "deleted_files_count", "type": "int"},
-                {"name": "added_rows_count", "type": "long"},
-                {"name": "existing_rows_count", "type": "long"},
-                {"name": "deleted_rows_count", "type": "long"}
-              ]
-            }
-            "#,
-    )
-    .expect("manifest list avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize");
-    for manifest_path in manifest_paths {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                (
-                    "manifest_path".to_string(),
-                    apache_avro::types::Value::String((*manifest_path).to_string()),
-                ),
-                ("manifest_length".to_string(), apache_avro::types::Value::Long(1)),
-                ("partition_spec_id".to_string(), apache_avro::types::Value::Int(partition_spec_id)),
-                ("content".to_string(), apache_avro::types::Value::Int(0)),
-                ("sequence_number".to_string(), apache_avro::types::Value::Long(7)),
-                ("min_sequence_number".to_string(), apache_avro::types::Value::Long(7)),
-                ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(20)),
-                ("added_files_count".to_string(), apache_avro::types::Value::Int(1)),
-                ("existing_files_count".to_string(), apache_avro::types::Value::Int(0)),
-                ("deleted_files_count".to_string(), apache_avro::types::Value::Int(0)),
-                ("added_rows_count".to_string(), apache_avro::types::Value::Long(1)),
-                ("existing_rows_count".to_string(), apache_avro::types::Value::Long(0)),
-                ("deleted_rows_count".to_string(), apache_avro::types::Value::Long(0)),
-            ]))
-            .expect("manifest list record should append");
-    }
-    writer.into_inner().expect("manifest list avro bytes should flush")
+    // Historical fixed values of this file's fixtures: sequence 7, snapshot 20.
+    let manifests = manifest_paths
+        .iter()
+        .map(|path| (*path, partition_spec_id, 7_i64, 20_i64))
+        .collect::<Vec<_>>();
+    crate::table_catalog::test_support::manifest_list_avro_entries_with_partition_specs(&manifests)
 }
 
 fn v1_manifest_list_avro_bytes(manifest_path: &str) -> Vec<u8> {
@@ -2336,55 +2294,13 @@ fn manifest_avro_bytes(files: &[(&str, i32)]) -> Vec<u8> {
 }
 
 fn manifest_avro_bytes_with_status(files: &[(&str, i32, i32)]) -> Vec<u8> {
-    let schema = apache_avro::Schema::parse_str(
-        r#"
-            {
-              "type": "record",
-              "name": "manifest_entry",
-              "fields": [
-                {"name": "status", "type": "int"},
-                {"name": "snapshot_id", "type": "long"},
-                {"name": "sequence_number", "type": "long"},
-                {"name": "file_sequence_number", "type": "long"},
-                {
-                  "name": "data_file",
-                  "type": {
-                    "type": "record",
-                    "name": "data_file",
-                    "fields": [
-                      {"name": "content", "type": "int"},
-                      {"name": "file_path", "type": "string"},
-                      {"name": "record_count", "type": "long"},
-                      {"name": "file_size_in_bytes", "type": "long"}
-                    ]
-                  }
-                }
-              ]
-            }
-            "#,
-    )
-    .expect("manifest avro schema should parse");
-    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
-    for (file_path, content, status) in files {
-        writer
-            .append_value(apache_avro::types::Value::Record(vec![
-                ("status".to_string(), apache_avro::types::Value::Int(*status)),
-                ("snapshot_id".to_string(), apache_avro::types::Value::Long(20)),
-                ("sequence_number".to_string(), apache_avro::types::Value::Long(7)),
-                ("file_sequence_number".to_string(), apache_avro::types::Value::Long(7)),
-                (
-                    "data_file".to_string(),
-                    apache_avro::types::Value::Record(vec![
-                        ("content".to_string(), apache_avro::types::Value::Int(*content)),
-                        ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
-                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
-                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
-                    ]),
-                ),
-            ]))
-            .expect("manifest record should append");
-    }
-    writer.into_inner().expect("manifest avro bytes should flush")
+    // Historical fixed values of this file's fixtures: snapshot 20, sequence 7
+    // (the shared constructor takes snapshot_id fourth, sequence fifth).
+    let files = files
+        .iter()
+        .map(|(path, content, status)| (*path, *content, *status, 20_i64, 7_i64))
+        .collect::<Vec<_>>();
+    crate::table_catalog::test_support::manifest_avro_bytes(&files)
 }
 
 fn manifest_avro_bytes_with_dt_partition(files: &[(&str, i32, &str)]) -> Vec<u8> {