test(ecstore): preserve inline budget semantics

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

.github/workflows/cache-warm.yml

@@ -182,12 +182,7 @@ jobs:
             echo '```'
           } >> "$GITHUB_STEP_SUMMARY"
 
-  # Readers: test-and-lint-rio-v2 (per-PR), build-rustfs-debug-binary-rio-v2
-  # (weekly schedule / manual dispatch only — dormant rio-v2 variant, see
-  # rustfs/backlog#1835 and docs/architecture/minio-file-format-compat.md).
-  # The second build below stays despite the reduced cadence: it warms the
-  # rio-v2,e2e-test-hooks feature resolution the scheduled build restores,
-  # which keeps that lane inside its 30-minute timeout.
+  # Readers: test-and-lint-rio-v2, build-rustfs-debug-binary-rio-v2.
   warm-ci-feat-rio:
     name: Warm ci-feat-rio
     runs-on: sm-standard-4

.github/workflows/ci.yml

@@ -533,12 +533,7 @@ jobs:
 
   build-rustfs-debug-binary-rio-v2:
     name: Build RustFS Debug Binary (rio-v2)
-    # Dormant rio-v2 variant (rustfs/backlog#1835): the feature ships in no
-    # default build, so this full-suite lane runs only on the weekly schedule
-    # and manual dispatch. Per-PR cfg-seam coverage stays with
-    # test-and-lint-rio-v2. Lifecycle and the promote-or-delete condition:
-    # docs/architecture/minio-file-format-compat.md ("rio-v2 variant lifecycle").
-    if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch'
+    if: github.event_name != 'pull_request' || github.event.action != 'closed'
     needs: [ quick-checks ]
     runs-on: sm-standard-4
     timeout-minutes: 30
@@ -829,9 +824,6 @@ jobs:
 
   e2e-tests-rio-v2:
     name: End-to-End Tests (rio-v2)
-    # Inherits the schedule/dispatch-only gate through needs: on every other
-    # event build-rustfs-debug-binary-rio-v2 is skipped, so this job skips
-    # with it (see the dormant-variant comment on that job).
     needs: [ build-rustfs-debug-binary-rio-v2 ]
     runs-on: sm-standard-2
     timeout-minutes: 30

ARCHITECTURE.md

@@ -101,10 +101,7 @@ refactors.
 
 The `rustfs` binary crate composes these libraries into the running server.
 `ecstore` remains the storage engine at the architectural center; its internal
-module split is tracked under `docs/architecture/`. `rio-v2` is the
-feature-gated MinIO on-disk format compatibility I/O layer; it ships in no
-default build (lifecycle:
-[docs/architecture/minio-file-format-compat.md](docs/architecture/minio-file-format-compat.md)).
+module split is tracked under `docs/architecture/`.
 
 ## Architecture Invariants
 

Cargo.toml

@@ -41,7 +41,7 @@ members = [
     "crates/protocols", # Protocol implementations (FTPS, SFTP, etc.)
     "crates/protos", # Protocol buffer definitions
     "crates/rio", # Rust I/O utilities and abstractions
-    "crates/rio-v2", # MinIO on-disk format compatibility I/O layer (feature-gated, ships in no default build)
+    "crates/rio-v2", # Next-generation Rust I/O compatibility layer
     "crates/replication", # Replication contracts and wire formats
     "crates/concurrency", # Concurrency management for RustFS - timeout, locking, backpressure, and I/O scheduling
     "crates/s3-types", # S3 event type definitions

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;

docs/architecture/compat-cleanup-register.md

@@ -34,7 +34,6 @@ for later deletion.
 - `tonic-013-status-render` peer RPC failure classification: internode failures that reach a node only as text (a peer's error_info payload, a status flattened through format!) are classified by matching the rendering of an Unavailable gRPC status. Releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered that status as "status: Unavailable, message: ..."; tonic 0.14 renders it as "code: 'The service is currently unavailable', message: ...". Both forms are matched so an older peer's relayed text still marks an unreachable peer offline. Remove the tonic 0.13 form after the minimum supported RustFS peer version ships tonic 0.14 or later.
 - `rustfs-5063` pre-beta.9 Local KMS recovery: persisted Local KMS configs from beta.8 and earlier predate the explicit insecure-development flag, and encrypted key files use the legacy SHA-256 KDF. Remove the config fallback after supported upgrades have rewritten or explicitly resaved all pre-beta.9 configs with the development-default field, and remove the legacy KDF after supported upgrades have rewritten all pre-beta.9 Local KMS key files with explicit at-rest protection.
 - `sse-local-dek-json-v1` legacy local SSE DEK decoding: releases before the JSON envelope wrote wrapped DEKs as `base64(nonce):base64(ciphertext)`, so readers retain that decoder while all new writes use the versioned JSON envelope. Remove the colon decoder after the minimum supported direct-upgrade release writes JSON envelopes and migration tooling has rewritten every retained legacy object.
-- `rio-v2-dormant-variant` dormant `rio-v2` build variant: `crates/rio-v2` and the `rio-v2` feature ship in no default or release build and exist only as the candidate MinIO stream-format implementation for the rustfs/backlog#1638 SSE-interop adjudication. Per-PR CI keeps only `test-and-lint-rio-v2` to guard the `#[cfg(feature = "rio-v2")]` seam; the full-suite lanes (`build-rustfs-debug-binary-rio-v2`, `e2e-tests-rio-v2` in `.github/workflows/ci.yml`) run on schedule/workflow_dispatch only — see the "`rio-v2` variant lifecycle" section in [minio-file-format-compat.md](minio-file-format-compat.md). While both implementations exist, DARE/S2 stream fixes must land in both `crates/rio` and `crates/rio-v2`. No `RUSTFS_COMPAT_TODO` source marker applies: the temporary surface is CI workflow YAML plus an entire feature-gated candidate crate, not a compatibility code path inside shipping code, and workflow files are outside the marker convention's Rust scope. Remove after the #1638 adjudication lands and converges on one implementation: delete the losing implementation, its feature seam, and the gating CI jobs.
 
 ## Review Checklist
 

docs/architecture/minio-file-format-compat.md

@@ -271,32 +271,6 @@ Seam 2 surfaces its own error, but only for objects that got past seam 1.
 
 The interop harness reflects this. The reader tests are `#[ignore]` (`rustfs/src/storage/minio_generated_read_test.rs:244`, `:250`), the workflow that would run them is disabled at the GitHub Actions level and states in its own header that end-to-end MinIO-to-RustFS SSE interop is not implemented (`.github/workflows/minio-interop.yml:24-29`, `:34-39`), and the fixture suite's scope note says the tests "do not yet validate full plaintext reconstruction from MinIO-written encrypted data" (`crates/rio-v2/tests/README.md:55`).
 
-### `rio-v2` variant lifecycle
-
-The variant is deliberately **dormant** until rustfs/backlog#1638 is
-adjudicated. Dormant means:
-
-- **Per-PR CI keeps one guard job.** Only `test-and-lint-rio-v2` in
-  `.github/workflows/ci.yml` runs per PR; its job is to keep the
-  `#[cfg(feature = "rio-v2")]` seam compiling and its unit tests green so the
-  variant does not bit-rot. The full-suite lanes —
-  `build-rustfs-debug-binary-rio-v2` and `e2e-tests-rio-v2` — run only on the
-  weekly `schedule` and on `workflow_dispatch`, not per PR, per main push, or
-  in the merge queue.
-- **Post-1.0 the variant is promoted or deleted.** The #1638 adjudication
-  converges on one implementation: either `rio-v2` becomes a shipped
-  configuration, or the losing side is removed together with its feature seam
-  and its gating CI jobs. Tracked as `rio-v2-dormant-variant` in
-  [compat-cleanup-register.md](compat-cleanup-register.md).
-- **DARE/S2 fixes land in both crates.** While both implementations exist,
-  any fix to the DARE V2 stream format or the S2 compression framing/index
-  must be applied to `crates/rio` **and** `crates/rio-v2` (each has its own
-  `encrypt_reader.rs` and `compress_reader.rs`). Both implement the same
-  stream primitives; a single-sided fix forks on-disk behavior between
-  default and `rio-v2` builds and invalidates the dormant variant as an
-  interop baseline — and with full-suite CI now weekly-only, the divergence
-  could go unnoticed for up to a week.
-
 ### Reverse direction
 
 Migrating back is also unsupported. Under `rio-v2` RustFS writes its own DEK envelope into MinIO's sealed-key metadata slots and labels it with MinIO's seal algorithm (`rustfs/src/storage/sse.rs:1830-1852`), so the metadata is MinIO-shaped while the key bytes are not MinIO-openable. Default builds do not populate those slots at all (`rustfs/src/storage/sse.rs:1796-1798`). Treat RustFS-written SSE objects as readable only by RustFS.

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> {