diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs
index a8358b4c5..672e290dc 100644
--- a/crates/ecstore/src/api/mod.rs
+++ b/crates/ecstore/src/api/mod.rs
@@ -261,9 +261,9 @@ pub mod data_usage {
         DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend,
         load_compression_total_from_memory, load_data_usage_from_backend, record_bucket_delete_marker_memory,
         record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory,
-        record_compression_total_memory, refresh_bucket_usage_from_object_layer,
-        refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend,
-        replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
+        record_bucket_object_write_unknown_previous_memory, record_compression_total_memory,
+        refresh_bucket_usage_from_object_layer, refresh_versioned_bucket_usage_from_object_layer,
+        remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
     };
 }
 
@@ -273,9 +273,9 @@ pub mod disk {
     pub use crate::disk::{
         BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp,
         CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, DiskStore,
-        FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp,
-        ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk,
-        validate_batch_read_version_item_count,
+        FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, OldCurrentSize, RUSTFS_META_BUCKET, ReadMultipleReq,
+        ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
+        new_disk, validate_batch_read_version_item_count,
     };
     pub use bytes::Bytes;
     pub use endpoint::Endpoint;
diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs
index a4a40ef21..0b518ab45 100644
--- a/crates/ecstore/src/cluster/rpc/remote_disk.rs
+++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs
@@ -3060,6 +3060,7 @@ mod tests {
         let response = RenameDataResp {
             old_data_dir: Some(Uuid::new_v4()),
             sign: Some(vec![1_u8; 32]),
+            old_current_size: Some(crate::disk::OldCurrentSize::Present(4096)),
         };
         let json = serde_json::to_vec(&response).expect("rename data response json should encode");
         let named_msgpack = encode_msgpack_named(&response).expect("rename data response named msgpack should encode");
diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs
index cd99272cc..16b6f183b 100644
--- a/crates/ecstore/src/core/sets.rs
+++ b/crates/ecstore/src/core/sets.rs
@@ -389,6 +389,22 @@ impl crate::storage_api_contracts::object::ObjectIO for Sets {
     }
 }
 
+impl Sets {
+    /// `put_object` plus the rename_data old-size backfill
+    /// (rustfs/backlog#1009); see `SetDisks::put_object_with_old_current_size`.
+    pub async fn put_object_with_old_current_size(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: &mut PutObjReader,
+        opts: &ObjectOptions,
+    ) -> Result<(ObjectInfo, Option<crate::disk::OldCurrentSize>)> {
+        self.get_disks_by_key(object)
+            .put_object_with_old_current_size(bucket, object, data, opts)
+            .await
+    }
+}
+
 #[async_trait::async_trait]
 impl BucketOperations for Sets {
     type Error = Error;
diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs
index defad2f3f..dc8d9e3ba 100644
--- a/crates/ecstore/src/data_usage/mod.rs
+++ b/crates/ecstore/src/data_usage/mod.rs
@@ -689,6 +689,34 @@ async fn record_bucket_object_write_memory_inner(
     entry.stale_snapshot_pending = false;
 }
 
+/// Degraded in-memory update for an object write whose previous current size
+/// could not be determined (rustfs/backlog#1009: the pre-PUT lookup was
+/// skipped and the rename_data backfill came back unknown — mixed-version
+/// peers or sub-quorum metadata divergence). Applies only the components that
+/// are correct regardless of the previous state: the new bytes always count,
+/// and a versioned write always adds a version. objects_count (and the
+/// non-versioned overwrite's old-size subtraction) are left to the next
+/// scanner refresh, which replaces this cache with authoritative numbers.
+pub async fn record_bucket_object_write_unknown_previous_memory(bucket: &str, new_size: u64, creates_new_version: bool) {
+    ensure_bucket_usage_cached(bucket).await;
+
+    let mut cache = memory_cache().write().await;
+    let entry = cache
+        .entry(bucket.to_string())
+        .or_insert_with(|| cached_bucket_usage_now(BucketUsageInfo::default()));
+
+    entry.usage.size = entry.usage.size.saturating_add(new_size);
+    if creates_new_version {
+        entry.usage.versions_count = entry.usage.versions_count.saturating_add(1);
+    }
+
+    let now = SystemTime::now();
+    entry.refreshed_at = now;
+    entry.usage_updated_at = now;
+    entry.dirty = true;
+    entry.stale_snapshot_pending = false;
+}
+
 /// Fast in-memory increment for immediate quota consistency.
 pub async fn increment_bucket_usage_memory(bucket: &str, size_increment: u64) {
     record_bucket_object_write_memory(bucket, None, size_increment).await;
@@ -1470,6 +1498,57 @@ mod tests {
         );
     }
 
+    /// rustfs/backlog#1009: with an unknown previous state, only the
+    /// always-correct components are recorded — bytes are added and a
+    /// versioned write adds a version, but objects_count never moves.
+    #[tokio::test]
+    #[serial]
+    async fn memory_overlay_unknown_previous_adds_size_only_for_unversioned_write() {
+        clear_usage_memory_cache_for_test().await;
+
+        let persisted = data_usage_info_for_test("bucket-a", 1, 10, SystemTime::now() - Duration::from_secs(10));
+        replace_bucket_usage_memory_from_info(&persisted).await;
+        record_bucket_object_write_unknown_previous_memory("bucket-a", 20, false).await;
+
+        let mut response = persisted.clone();
+        apply_bucket_usage_memory_overlay(&mut response).await;
+
+        assert_eq!(response.objects_total_count, 1);
+        assert_eq!(response.versions_total_count, 1);
+        assert_eq!(response.objects_total_size, 30);
+        assert_eq!(
+            response
+                .buckets_usage
+                .get("bucket-a")
+                .map(|usage| (usage.objects_count, usage.versions_count, usage.size)),
+            Some((1, 1, 30))
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn memory_overlay_unknown_previous_adds_size_and_version_for_versioned_write() {
+        clear_usage_memory_cache_for_test().await;
+
+        let persisted = data_usage_info_for_test("bucket-a", 1, 10, SystemTime::now() - Duration::from_secs(10));
+        replace_bucket_usage_memory_from_info(&persisted).await;
+        record_bucket_object_write_unknown_previous_memory("bucket-a", 20, true).await;
+
+        let mut response = persisted.clone();
+        apply_bucket_usage_memory_overlay(&mut response).await;
+
+        assert_eq!(response.objects_total_count, 1);
+        assert_eq!(response.versions_total_count, 2);
+        assert_eq!(response.objects_total_size, 30);
+        assert_eq!(
+            response
+                .buckets_usage
+                .get("bucket-a")
+                .map(|usage| (usage.objects_count, usage.versions_count, usage.size)),
+            Some((1, 2, 30))
+        );
+    }
+
     #[tokio::test]
     #[serial]
     async fn memory_overlay_records_delete_marker_without_removing_versions() {
diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs
index 273ca8d70..22acd18bc 100644
--- a/crates/ecstore/src/disk/local.rs
+++ b/crates/ecstore/src/disk/local.rs
@@ -18,7 +18,7 @@ use crate::disk::disk_store::get_object_disk_read_timeout;
 use crate::disk::{
     BUCKET_META_PREFIX, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN,
     CHECK_PART_VOLUME_NOT_FOUND, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics,
-    FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET,
+    FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, OldCurrentSize, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET,
     RUSTFS_META_TMP_DELETED_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE,
     STORAGE_FORMAT_FILE_BACKUP, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, conv_part_err_to_int,
     endpoint::Endpoint,
@@ -4076,6 +4076,50 @@ fn rename_data_versions_signature(meta: &FileMeta) -> Option<Vec<u8>> {
     Some(signature)
 }
 
+/// rustfs/backlog#1009: observe the destination key's *current* (latest)
+/// version in the dst `xl.meta` that `rename_data` already loaded, before
+/// `add_version` commits the incoming one. Replicates the pre-PUT
+/// `get_object_info` outcome bit for bit, as the per-disk `read_version`
+/// pipeline (`get_file_info`) would report it:
+///
+/// - `dst_meta_existed == false` (no dst xl.meta on this disk): the lookup
+///   errors FileNotFound → app-level `None` → `Absent`.
+/// - Existing meta whose versions are all hidden free versions: the lookup
+///   errors FileNotFound the same way → `Absent`.
+/// - Existing meta with zero versions: `get_file_info` synthesizes a deleted
+///   `FileInfo` with size 0 and the lookup returns `Ok` → `Present(0)`.
+/// - A resolvable latest version — live object, delete marker, or a
+///   purge-pending version flagged `deleted` — returns `Ok` with
+///   `ObjectInfo.size == fi.size` (0 for markers) → `Present(fi.size)`.
+///   Delete markers deliberately do NOT map to `Absent`: today's lookup
+///   returns `Ok(size 0)` for them, and delete-marker creation never
+///   decrements `objects_count`, so `Some(0)` is what keeps versioned
+///   accounting bit-identical.
+/// - A latest version that fails to decode — including the part-array length
+///   guard that `all_parts=true` enables, the same flag the per-disk lookup
+///   uses — yields `None` (unknown): the old lookup surfaced a per-disk error
+///   there, so this disk must not vote in the set-level quorum reduction.
+///
+/// One deliberate divergence: `read_data` stays `false` (the lookup used
+/// `true`), so a corrupt inline-data map that would have errored the old
+/// lookup votes the version's own `size` here instead of abstaining. The
+/// size field decodes independently of the inline data, so the vote carries
+/// the same value healthy disks report, and skipping the lookup's inline
+/// bytes clone keeps the observation allocation-light.
+fn observe_old_current_size(dst_meta_existed: bool, xlmeta: &FileMeta) -> Option<OldCurrentSize> {
+    if !dst_meta_existed {
+        return Some(OldCurrentSize::Absent);
+    }
+    if xlmeta.versions.is_empty() {
+        return Some(OldCurrentSize::Present(0));
+    }
+    match xlmeta.into_fileinfo("", "", "", false, false, true) {
+        Ok(fi) => Some(OldCurrentSize::Present(fi.size)),
+        Err(rustfs_filemeta::Error::FileNotFound) => Some(OldCurrentSize::Absent),
+        Err(_) => None,
+    }
+}
+
 fn is_root_path(path: impl AsRef<Path>) -> bool {
     path.as_ref().components().count() == 1 && path.as_ref().has_root()
 }
@@ -5108,13 +5152,27 @@ impl DiskAPI for LocalDisk {
             };
 
             let mut xlmeta = FileMeta::new();
-            if let Some(dst_buf) = has_dst_buf.as_ref()
-                && FileMeta::is_xl2_v1_format(dst_buf)
-                && let Ok(nmeta) = FileMeta::load(dst_buf)
-            {
-                xlmeta = nmeta
+            // An existing dst xl.meta that fails to parse leaves `xlmeta` empty
+            // and gets overwritten by the commit below (pre-existing behavior);
+            // track that so the old-size observation reports unknown instead of
+            // a false `Absent` (rustfs/backlog#1009).
+            let mut dst_meta_unparsable = false;
+            if let Some(dst_buf) = has_dst_buf.as_ref() {
+                if FileMeta::is_xl2_v1_format(dst_buf)
+                    && let Ok(nmeta) = FileMeta::load(dst_buf)
+                {
+                    xlmeta = nmeta
+                } else {
+                    dst_meta_unparsable = true;
+                }
             }
 
+            let old_current_size = if dst_meta_unparsable {
+                None
+            } else {
+                observe_old_current_size(has_dst_buf.is_some(), &xlmeta)
+            };
+
             let mut skip_parent = dst_volume_dir.clone();
             if has_dst_buf.as_ref().is_some()
                 && let Some(parent) = dst_file_path.parent()
@@ -5325,6 +5383,7 @@ impl DiskAPI for LocalDisk {
             Ok(RenameDataResp {
                 old_data_dir: has_old_data_dir,
                 sign: version_signature,
+                old_current_size,
             })
         } else {
             // Inline: merge read + parse + write + rename into single spawn_blocking
@@ -5338,7 +5397,7 @@ impl DiskAPI for LocalDisk {
                 None
             };
 
-            let (old_data_dir, version_signature) = tokio::task::spawn_blocking(move || {
+            let (old_data_dir, version_signature, old_current_size) = tokio::task::spawn_blocking(move || {
                 // Read existing xl.meta
                 let has_dst_buf = match std::fs::read(&dst) {
                     Ok(buf) => Some(Bytes::from(buf)),
@@ -5347,13 +5406,26 @@ impl DiskAPI for LocalDisk {
                 };
 
                 let mut xlmeta = FileMeta::new();
-                if let Some(ref buf) = has_dst_buf
-                    && FileMeta::is_xl2_v1_format(buf)
-                    && let Ok(nmeta) = FileMeta::load(buf)
-                {
-                    xlmeta = nmeta
+                // Same as the non-inline branch: an unparsable existing dst
+                // xl.meta must surface as unknown, not `Absent`
+                // (rustfs/backlog#1009).
+                let mut dst_meta_unparsable = false;
+                if let Some(ref buf) = has_dst_buf {
+                    if FileMeta::is_xl2_v1_format(buf)
+                        && let Ok(nmeta) = FileMeta::load(buf)
+                    {
+                        xlmeta = nmeta
+                    } else {
+                        dst_meta_unparsable = true;
+                    }
                 }
 
+                let old_current_size = if dst_meta_unparsable {
+                    None
+                } else {
+                    observe_old_current_size(has_dst_buf.is_some(), &xlmeta)
+                };
+
                 let version_id = fi.version_id.unwrap_or_default();
                 let old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id));
                 if let Some(d) = old_data_dir.as_ref() {
@@ -5450,7 +5522,11 @@ impl DiskAPI for LocalDisk {
                     }
                 }
 
-                Ok::<(Option<uuid::Uuid>, Option<Vec<u8>>), std::io::Error>((old_data_dir, version_signature))
+                Ok::<(Option<uuid::Uuid>, Option<Vec<u8>>, Option<OldCurrentSize>), std::io::Error>((
+                    old_data_dir,
+                    version_signature,
+                    old_current_size,
+                ))
             })
             .await
             .map_err(DiskError::from)??;
@@ -5465,6 +5541,7 @@ impl DiskAPI for LocalDisk {
             Ok(RenameDataResp {
                 old_data_dir,
                 sign: version_signature,
+                old_current_size,
             })
         }
     }
@@ -7669,6 +7746,266 @@ mod test {
                 .join(STORAGE_FORMAT_FILE)
                 .exists()
         );
+        // rustfs/backlog#1009: the overwritten live current version (size 1
+        // from `test_file_info`) must be surfaced through the backfill.
+        assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(1)));
+    }
+
+    /// rustfs/backlog#1009: `observe_old_current_size` must mirror the pre-PUT
+    /// `get_object_info` semantics bit for bit — latest version's
+    /// `ObjectInfo.size` (0 for a delete-marker latest, which that lookup
+    /// returns as `Ok`, not as not-found), missing key → `Absent` — and
+    /// `rename_data` must report it for both the inline and non-inline commit
+    /// branches.
+    mod old_current_size_backfill {
+        use super::*;
+        use tempfile::tempdir;
+
+        fn live_file_info(name: &str, version_id: Uuid, size: i64, mod_time: OffsetDateTime) -> FileInfo {
+            FileInfo {
+                name: name.to_string(),
+                version_id: Some(version_id),
+                size,
+                mod_time: Some(mod_time),
+                ..Default::default()
+            }
+        }
+
+        fn delete_marker_file_info(name: &str, version_id: Uuid, mod_time: OffsetDateTime) -> FileInfo {
+            FileInfo {
+                name: name.to_string(),
+                version_id: Some(version_id),
+                deleted: true,
+                mod_time: Some(mod_time),
+                ..Default::default()
+            }
+        }
+
+        #[test]
+        fn observe_reports_absent_for_missing_key() {
+            assert_eq!(observe_old_current_size(false, &FileMeta::default()), Some(OldCurrentSize::Absent));
+        }
+
+        /// An existing xl.meta with zero versions reads back through
+        /// `get_file_info` as a synthetic deleted FileInfo of size 0, so the
+        /// pre-PUT lookup reported `Some(0)`, not not-found.
+        #[test]
+        fn observe_reports_present_zero_for_existing_versionless_meta() {
+            assert_eq!(observe_old_current_size(true, &FileMeta::default()), Some(OldCurrentSize::Present(0)));
+        }
+
+        #[test]
+        fn observe_reports_latest_live_version_size() {
+            let now = OffsetDateTime::now_utc();
+            let mut meta = FileMeta::new();
+            meta.add_version(live_file_info("object", Uuid::new_v4(), 10, now - time::Duration::seconds(10)))
+                .expect("older live version should be added");
+            meta.add_version(live_file_info("object", Uuid::new_v4(), 42, now))
+                .expect("newer live version should be added");
+
+            assert_eq!(observe_old_current_size(true, &meta), Some(OldCurrentSize::Present(42)));
+        }
+
+        /// The pre-PUT lookup returns `Ok(size 0)` for a delete-marker latest
+        /// (RustFS's `SetDisks::get_object_info` does not convert markers to
+        /// not-found), and delete-marker creation never decrements
+        /// objects_count — so the backfill must report `Present(0)` here, not
+        /// `Absent`, to keep versioned accounting bit-identical.
+        #[test]
+        fn observe_reports_present_zero_for_delete_marker_latest() {
+            let now = OffsetDateTime::now_utc();
+            let mut meta = FileMeta::new();
+            meta.add_version(live_file_info("object", Uuid::new_v4(), 42, now - time::Duration::seconds(10)))
+                .expect("live version should be added");
+            meta.add_version(delete_marker_file_info("object", Uuid::new_v4(), now))
+                .expect("delete marker should be added");
+
+            assert_eq!(observe_old_current_size(true, &meta), Some(OldCurrentSize::Present(0)));
+        }
+
+        /// A latest version whose part arrays are corrupt (lengths disagree)
+        /// made the old per-disk lookup error out — this disk must abstain
+        /// (`None`), never vote. Pins the `all_parts=true` flag that enables
+        /// the part-array length guard.
+        #[test]
+        fn observe_abstains_for_corrupt_part_arrays() {
+            let now = OffsetDateTime::now_utc();
+            let mut fi = live_file_info("object", Uuid::new_v4(), 42, now);
+            fi.add_object_part(1, "etag".to_string(), 42, Some(now), 42, None, None);
+            let mut version = rustfs_filemeta::FileMetaVersion::from(fi);
+            version
+                .object
+                .as_mut()
+                .expect("object version should carry a MetaObject")
+                .part_sizes
+                .clear();
+
+            let mut meta = FileMeta::new();
+            meta.add_version_filemata(version)
+                .expect("corrupt-part version should still insert");
+
+            assert_eq!(observe_old_current_size(true, &meta), None);
+        }
+
+        async fn test_disk(dir: &tempfile::TempDir) -> LocalDisk {
+            let endpoint =
+                Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+            let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+            ensure_test_volume(&disk, "bucket").await;
+            ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+            disk
+        }
+
+        #[tokio::test]
+        async fn inline_rename_data_reports_absent_for_fresh_key() {
+            let dir = tempdir().expect("temp dir should be created");
+            let disk = test_disk(&dir).await;
+
+            let new_fi = test_file_info("object", Uuid::new_v4(), None, Some(Bytes::from_static(b"inline-new")));
+            let resp = disk
+                .rename_data(RUSTFS_META_TMP_BUCKET, "tmp-fresh", new_fi, "bucket", "object")
+                .await
+                .expect("inline rename_data should commit");
+
+            assert_eq!(resp.old_current_size, Some(OldCurrentSize::Absent));
+        }
+
+        #[tokio::test]
+        async fn inline_rename_data_reports_present_zero_for_delete_marker_latest() {
+            let dir = tempdir().expect("temp dir should be created");
+            let disk = test_disk(&dir).await;
+
+            let now = OffsetDateTime::now_utc();
+            let mut old_meta = FileMeta::new();
+            old_meta
+                .add_version(live_file_info("object", Uuid::new_v4(), 42, now - time::Duration::seconds(10)))
+                .expect("live version should be added");
+            old_meta
+                .add_version(delete_marker_file_info("object", Uuid::new_v4(), now))
+                .expect("delete marker should be added");
+            let dst_object_dir = dir.path().join("bucket").join("object");
+            fs::create_dir_all(&dst_object_dir).await.expect("dst dir should be created");
+            fs::write(
+                dst_object_dir.join(STORAGE_FORMAT_FILE),
+                old_meta.marshal_msg().expect("old metadata should encode"),
+            )
+            .await
+            .expect("old metadata should be written");
+
+            let new_fi = test_file_info("object", Uuid::new_v4(), None, Some(Bytes::from_static(b"inline-new")));
+            let resp = disk
+                .rename_data(RUSTFS_META_TMP_BUCKET, "tmp-marker", new_fi, "bucket", "object")
+                .await
+                .expect("inline rename_data should commit");
+
+            // The pre-PUT lookup returns Ok(size 0) for a marker latest, so
+            // the backfill must match it (see observe_old_current_size docs).
+            assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(0)));
+        }
+
+        #[tokio::test]
+        async fn inline_rename_data_reports_unknown_for_unparsable_dst_meta() {
+            let dir = tempdir().expect("temp dir should be created");
+            let disk = test_disk(&dir).await;
+
+            let dst_object_dir = dir.path().join("bucket").join("object");
+            fs::create_dir_all(&dst_object_dir).await.expect("dst dir should be created");
+            fs::write(dst_object_dir.join(STORAGE_FORMAT_FILE), b"not-an-xl-meta")
+                .await
+                .expect("garbage metadata should be written");
+
+            let new_fi = test_file_info("object", Uuid::new_v4(), None, Some(Bytes::from_static(b"inline-new")));
+            let resp = disk
+                .rename_data(RUSTFS_META_TMP_BUCKET, "tmp-garbage", new_fi, "bucket", "object")
+                .await
+                .expect("inline rename_data should commit");
+
+            assert_eq!(resp.old_current_size, None);
+        }
+
+        #[tokio::test]
+        async fn non_inline_rename_data_reports_absent_then_previous_size() {
+            let dir = tempdir().expect("temp dir should be created");
+            let disk = test_disk(&dir).await;
+
+            // First non-inline commit: fresh key must report Absent.
+            let first_data_dir = Uuid::new_v4();
+            let tmp_data_dir = dir
+                .path()
+                .join(RUSTFS_META_TMP_BUCKET)
+                .join("tmp-first")
+                .join(first_data_dir.to_string());
+            fs::create_dir_all(&tmp_data_dir)
+                .await
+                .expect("tmp data dir should be created");
+            fs::write(tmp_data_dir.join("part.1"), b"first")
+                .await
+                .expect("part should be written");
+            let mut first_fi = test_file_info("object", Uuid::new_v4(), Some(first_data_dir), None);
+            first_fi.size = 5;
+            let resp = disk
+                .rename_data(RUSTFS_META_TMP_BUCKET, "tmp-first", first_fi, "bucket", "object")
+                .await
+                .expect("first non-inline rename_data should commit");
+            assert_eq!(resp.old_current_size, Some(OldCurrentSize::Absent));
+
+            // Overwrite: the committed live version (size 5) must be reported.
+            let second_data_dir = Uuid::new_v4();
+            let tmp_data_dir = dir
+                .path()
+                .join(RUSTFS_META_TMP_BUCKET)
+                .join("tmp-second")
+                .join(second_data_dir.to_string());
+            fs::create_dir_all(&tmp_data_dir)
+                .await
+                .expect("tmp data dir should be created");
+            fs::write(tmp_data_dir.join("part.1"), b"second-longer")
+                .await
+                .expect("part should be written");
+            let mut second_fi = test_file_info("object", Uuid::new_v4(), Some(second_data_dir), None);
+            second_fi.size = 13;
+            let resp = disk
+                .rename_data(RUSTFS_META_TMP_BUCKET, "tmp-second", second_fi, "bucket", "object")
+                .await
+                .expect("second non-inline rename_data should commit");
+            assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(5)));
+        }
+
+        /// Twin of the inline unparsable-dst test: the `dst_meta_unparsable`
+        /// tracking is duplicated per branch, so the non-inline copy needs its
+        /// own regression coverage.
+        #[tokio::test]
+        async fn non_inline_rename_data_reports_unknown_for_unparsable_dst_meta() {
+            let dir = tempdir().expect("temp dir should be created");
+            let disk = test_disk(&dir).await;
+
+            let dst_object_dir = dir.path().join("bucket").join("object");
+            fs::create_dir_all(&dst_object_dir).await.expect("dst dir should be created");
+            fs::write(dst_object_dir.join(STORAGE_FORMAT_FILE), b"not-an-xl-meta")
+                .await
+                .expect("garbage metadata should be written");
+
+            let data_dir = Uuid::new_v4();
+            let tmp_data_dir = dir
+                .path()
+                .join(RUSTFS_META_TMP_BUCKET)
+                .join("tmp-garbage-noninline")
+                .join(data_dir.to_string());
+            fs::create_dir_all(&tmp_data_dir)
+                .await
+                .expect("tmp data dir should be created");
+            fs::write(tmp_data_dir.join("part.1"), b"payload")
+                .await
+                .expect("part should be written");
+            let mut new_fi = test_file_info("object", Uuid::new_v4(), Some(data_dir), None);
+            new_fi.size = 7;
+            let resp = disk
+                .rename_data(RUSTFS_META_TMP_BUCKET, "tmp-garbage-noninline", new_fi, "bucket", "object")
+                .await
+                .expect("non-inline rename_data should commit over garbage metadata");
+
+            assert_eq!(resp.old_current_size, None);
+        }
     }
 
     #[tokio::test]
diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs
index 9e253f4c8..66287e6dc 100644
--- a/crates/ecstore/src/disk/mod.rs
+++ b/crates/ecstore/src/disk/mod.rs
@@ -839,10 +839,43 @@ pub struct DiskOption {
     pub health_check: bool,
 }
 
+/// Per-disk observation of the destination key's *current* (latest) version
+/// in the dst `xl.meta` that `rename_data` reads before it commits the
+/// incoming version (rustfs/backlog#1009). Mirrors the pre-PUT
+/// `get_object_info` outcome — which the app layer previously obtained with an
+/// extra full-disk fanout — bit for bit: `Absent` iff that lookup would report
+/// object-not-found (missing xl.meta, or only hidden free versions), and
+/// `Present(size)` iff it would return `Ok` with `ObjectInfo.size == size`.
+/// Note a delete-marker latest is `Present(0)`, not `Absent`: the lookup
+/// returns markers as `Ok(size 0)` and delete-marker accounting never
+/// decrements objects_count, so `Present(0)` is what keeps usage numbers
+/// identical.
+///
+/// The parity target is the set-level (`SetDisks`) lookup. Two app-visible
+/// lookup variants deviated from it and from each other — the multi-pool
+/// `ECStore` path converts a delete-marker latest to not-found, and
+/// directory-key lookups pin the nil version — and both deviations
+/// over-incremented objects_count relative to delete-marker accounting, so
+/// the backfill standardizes on the self-consistent set-level answer.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub enum OldCurrentSize {
+    /// The pre-PUT lookup would have reported object-not-found for this key.
+    Absent,
+    /// The pre-PUT lookup would have returned `Ok` with this `ObjectInfo.size`
+    /// (0 for a delete-marker latest).
+    Present(i64),
+}
+
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct RenameDataResp {
     pub old_data_dir: Option<Uuid>,
     pub sign: Option<Vec<u8>>,
+    /// `None` means unknown — the disk could not determine the previous
+    /// current version (pre-#1009 peer on the wire, or an existing dst
+    /// `xl.meta` that failed to parse). Consumers must treat unknown as
+    /// "cannot vote", never as `Absent`.
+    #[serde(default)]
+    pub old_current_size: Option<OldCurrentSize>,
 }
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
@@ -1229,10 +1262,56 @@ mod tests {
         let resp = RenameDataResp {
             old_data_dir: Some(uuid),
             sign: Some(signature.clone()),
+            old_current_size: Some(OldCurrentSize::Present(42)),
         };
 
         assert_eq!(resp.old_data_dir, Some(uuid));
         assert_eq!(resp.sign, Some(signature));
+        assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(42)));
+    }
+
+    /// rustfs/backlog#1009: `old_current_size` must survive a named-msgpack
+    /// round trip (the internode RPC encoding) for every variant.
+    #[test]
+    fn test_rename_data_resp_old_current_size_msgpack_roundtrip() {
+        for old_current_size in [None, Some(OldCurrentSize::Absent), Some(OldCurrentSize::Present(1337))] {
+            let resp = RenameDataResp {
+                old_data_dir: Some(Uuid::new_v4()),
+                sign: Some(vec![0x01, 0x02, 0x03]),
+                old_current_size,
+            };
+
+            let encoded = rmp_serde::encode::to_vec_named(&resp).expect("named msgpack should encode");
+            let decoded: RenameDataResp = rmp_serde::decode::from_slice(&encoded).expect("named msgpack should decode");
+
+            assert_eq!(decoded.old_data_dir, resp.old_data_dir);
+            assert_eq!(decoded.sign, resp.sign);
+            assert_eq!(decoded.old_current_size, resp.old_current_size);
+        }
+    }
+
+    /// rustfs/backlog#1009: a payload from a peer that predates
+    /// `old_current_size` must decode with the field defaulting to `None`
+    /// (unknown), keeping mixed-version clusters wire-compatible.
+    #[test]
+    fn test_rename_data_resp_decodes_payload_without_old_current_size() {
+        #[derive(Serialize)]
+        struct LegacyRenameDataResp {
+            old_data_dir: Option<Uuid>,
+            sign: Option<Vec<u8>>,
+        }
+
+        let legacy = LegacyRenameDataResp {
+            old_data_dir: Some(Uuid::new_v4()),
+            sign: Some(vec![0x0a, 0x0b]),
+        };
+
+        let encoded = rmp_serde::encode::to_vec_named(&legacy).expect("legacy named msgpack should encode");
+        let decoded: RenameDataResp = rmp_serde::decode::from_slice(&encoded).expect("legacy payload should decode");
+
+        assert_eq!(decoded.old_data_dir, legacy.old_data_dir);
+        assert_eq!(decoded.sign, legacy.sign);
+        assert_eq!(decoded.old_current_size, None);
     }
 
     /// Test constants
diff --git a/crates/ecstore/src/ecstore_validation_blackbox.rs b/crates/ecstore/src/ecstore_validation_blackbox.rs
index 44badf145..3c94e3e9b 100644
--- a/crates/ecstore/src/ecstore_validation_blackbox.rs
+++ b/crates/ecstore/src/ecstore_validation_blackbox.rs
@@ -492,3 +492,179 @@ async fn blackbox_issue3031_diag_covers_put_success_cleanup_and_error_summary()
     })
     .await;
 }
+
+/// rustfs/backlog#1009: the `put_object` old-size backfill must reproduce, for
+/// every PUT shape, exactly what the app layer's pre-PUT `get_object_info`
+/// lookup would have observed — that is the invariant that lets the app skip
+/// the lookup without changing a single usage-accounting number.
+mod old_current_size_backfill {
+    use super::*;
+    use crate::disk::OldCurrentSize;
+    use crate::error::is_err_object_not_found;
+    use crate::set_disk::SetDisks;
+
+    /// What the pre-PUT lookup would report right now for `object`.
+    async fn prelookup_expectation(set_disks: &Arc<SetDisks>, bucket: &str, object: &str) -> OldCurrentSize {
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+        match set_disks.get_object_info(bucket, object, &opts).await {
+            Ok(object_info) => OldCurrentSize::Present(object_info.size),
+            Err(err) => {
+                assert!(is_err_object_not_found(&err), "prelookup expectation hit unexpected error: {err:?}");
+                OldCurrentSize::Absent
+            }
+        }
+    }
+
+    async fn put_and_backfill(
+        set_disks: &Arc<SetDisks>,
+        bucket: &str,
+        object: &str,
+        len: usize,
+        opts: &ObjectOptions,
+    ) -> Option<OldCurrentSize> {
+        let payload = (0..len).map(|idx| ((idx * 31) % 251) as u8).collect::<Vec<_>>();
+        let mut reader = PutObjReader::from_vec(payload);
+        let (_object_info, backfill) = set_disks
+            .put_object_with_old_current_size(bucket, object, &mut reader, opts)
+            .await
+            .expect("put_object should succeed");
+        backfill
+    }
+
+    #[tokio::test]
+    async fn backfill_matches_prelookup_for_every_put_shape() {
+        let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+        let bucket = "bb-old-current-size";
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+
+        // Fresh key (inline-sized): both the lookup and the backfill say Absent.
+        let object = "object.bin";
+        assert_eq!(prelookup_expectation(&set_disks, bucket, object).await, OldCurrentSize::Absent);
+        assert_eq!(
+            put_and_backfill(&set_disks, bucket, object, 1024, &opts).await,
+            Some(OldCurrentSize::Absent)
+        );
+
+        // Unversioned inline overwrite: the previous live size must surface.
+        let expected = prelookup_expectation(&set_disks, bucket, object).await;
+        assert_eq!(expected, OldCurrentSize::Present(1024));
+        assert_eq!(put_and_backfill(&set_disks, bucket, object, 2048, &opts).await, Some(expected));
+
+        // Non-inline overwrite (payload above the inline block threshold).
+        let expected = prelookup_expectation(&set_disks, bucket, object).await;
+        assert_eq!(expected, OldCurrentSize::Present(2048));
+        assert_eq!(
+            put_and_backfill(&set_disks, bucket, object, BLOCK_SIZE_V2 + 123, &opts).await,
+            Some(expected)
+        );
+
+        // Overwriting a non-inline object reports its size back.
+        let expected = prelookup_expectation(&set_disks, bucket, object).await;
+        assert_eq!(expected, OldCurrentSize::Present((BLOCK_SIZE_V2 + 123) as i64));
+        assert_eq!(put_and_backfill(&set_disks, bucket, object, 64, &opts).await, Some(expected));
+
+        // Versioned writes: a new version over a live latest reports the
+        // latest's size, not the incoming version's.
+        let versioned = "versioned.bin";
+        let first_version_opts = ObjectOptions {
+            no_lock: true,
+            versioned: true,
+            version_id: Some(uuid::Uuid::new_v4().to_string()),
+            ..Default::default()
+        };
+        assert_eq!(
+            put_and_backfill(&set_disks, bucket, versioned, 111, &first_version_opts).await,
+            Some(OldCurrentSize::Absent)
+        );
+        let expected = prelookup_expectation(&set_disks, bucket, versioned).await;
+        assert_eq!(expected, OldCurrentSize::Present(111));
+        let second_version_opts = ObjectOptions {
+            no_lock: true,
+            versioned: true,
+            version_id: Some(uuid::Uuid::new_v4().to_string()),
+            ..Default::default()
+        };
+        assert_eq!(
+            put_and_backfill(&set_disks, bucket, versioned, 222, &second_version_opts).await,
+            Some(expected)
+        );
+
+        // Delete-marker latest: today's lookup returns the marker as
+        // Ok(size 0) — NOT not-found — and delete-marker creation never
+        // decremented objects_count, so the backfill must reproduce
+        // Present(0) to keep versioned accounting identical.
+        set_disks
+            .delete_object(
+                bucket,
+                versioned,
+                ObjectOptions {
+                    no_lock: true,
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("versioned delete should create a delete marker");
+        let expected = prelookup_expectation(&set_disks, bucket, versioned).await;
+        assert_eq!(expected, OldCurrentSize::Present(0));
+        let third_version_opts = ObjectOptions {
+            no_lock: true,
+            versioned: true,
+            version_id: Some(uuid::Uuid::new_v4().to_string()),
+            ..Default::default()
+        };
+        assert_eq!(
+            put_and_backfill(&set_disks, bucket, versioned, 333, &third_version_opts).await,
+            Some(expected)
+        );
+    }
+
+    /// Sub-quorum divergence through the real set-level fanout: when only 2 of
+    /// 4 disks can still decode the destination's xl.meta (write quorum is 3),
+    /// the PUT must still commit but the backfill must come back unknown —
+    /// never a fabricated `Absent`/`Present` from a below-quorum vote.
+    #[tokio::test]
+    async fn backfill_is_unknown_below_quorum_but_put_succeeds() {
+        let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+        let bucket = "bb-old-current-subquorum";
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+
+        let object = "object.bin";
+        assert_eq!(
+            put_and_backfill(&set_disks, bucket, object, 1024, &opts).await,
+            Some(OldCurrentSize::Absent)
+        );
+
+        // Corrupt the committed xl.meta on two disks: those observations turn
+        // unknown, leaving 2 definite votes < write quorum 3.
+        {
+            let disks = set_disks.disks.read().await;
+            for disk in disks.iter().flatten().take(2) {
+                let meta_path = disk.path().join(bucket).join(object).join("xl.meta");
+                fs::write(&meta_path, b"not-an-xl-meta")
+                    .await
+                    .expect("xl.meta should be overwritable with garbage");
+            }
+        }
+
+        let backfill = put_and_backfill(&set_disks, bucket, object, 2048, &opts).await;
+        assert_eq!(backfill, None, "a below-quorum vote must surface as unknown");
+    }
+}
diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs
index 81a3fbd08..994c29068 100644
--- a/crates/ecstore/src/set_disk/core/io_primitives.rs
+++ b/crates/ecstore/src/set_disk/core/io_primitives.rs
@@ -46,6 +46,7 @@ use crate::diagnostics::get::{
     GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
     record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
 };
+use crate::disk::OldCurrentSize;
 use crate::erasure::coding::BitrotReader;
 use crate::io_support::bitrot::{
     BitrotReaderStageMetrics, DeferredReaderStripeHandle, create_bitrot_reader_with_stage_metrics,
@@ -2490,7 +2491,13 @@ impl SetDisks {
         dst_bucket: &str,
         dst_object: &str,
         write_quorum: usize,
-    ) -> disk::error::Result<(Vec<Option<DiskStore>>, Option<Vec<u8>>, Option<Uuid>, Vec<Option<DiskStore>>)> {
+    ) -> disk::error::Result<(
+        Vec<Option<DiskStore>>,
+        Option<Vec<u8>>,
+        Option<Uuid>,
+        Vec<Option<DiskStore>>,
+        Option<OldCurrentSize>,
+    )> {
         let mut futures = Vec::with_capacity(disks.len());
 
         let mut errs = Vec::with_capacity(disks.len());
@@ -2528,6 +2535,7 @@ impl SetDisks {
 
         let mut disk_versions = vec![None; disks.len()];
         let mut data_dirs = vec![None; disks.len()];
+        let mut old_current_sizes = vec![None; disks.len()];
 
         let results = join_all(futures).await;
 
@@ -2536,6 +2544,7 @@ impl SetDisks {
                 Ok(res) => {
                     data_dirs[idx] = res.old_data_dir;
                     disk_versions[idx].clone_from(&res.sign);
+                    old_current_sizes[idx] = res.old_current_size;
                     errs.push(None);
                 }
                 Err(e) => {
@@ -2639,6 +2648,7 @@ impl SetDisks {
 
         let data_dir = Self::reduce_common_data_dir(&data_dirs, write_quorum);
         let versions = Self::select_rename_data_versions(&disk_versions, &errs, write_quorum);
+        let old_current_size = Self::reduce_common_old_current_size(&old_current_sizes, write_quorum);
         let online_disks = Self::eval_disks(disks, &errs);
         let cleanup_disks = if let Some(data_dir) = data_dir {
             disks
@@ -2657,7 +2667,35 @@ impl SetDisks {
             vec![None; disks.len()]
         };
 
-        Ok((online_disks, versions, data_dir, cleanup_disks))
+        Ok((online_disks, versions, data_dir, cleanup_disks, old_current_size))
+    }
+
+    /// rustfs/backlog#1009: reduce the per-disk observations of the
+    /// destination's previous current version to one set-level value, mirroring
+    /// `reduce_common_data_dir`: the observation reported by at least
+    /// `write_quorum` disks wins; anything short of that (disk errors, unknown
+    /// votes from pre-#1009 peers or unparsable metadata, genuine divergence)
+    /// yields `None` (unknown). Unknown per-disk entries never vote.
+    pub(in crate::set_disk) fn reduce_common_old_current_size(
+        old_current_sizes: &[Option<OldCurrentSize>],
+        write_quorum: usize,
+    ) -> Option<OldCurrentSize> {
+        let mut counts: HashMap<OldCurrentSize, usize> = HashMap::new();
+
+        for observation in old_current_sizes.iter().flatten().copied() {
+            *counts.entry(observation).or_insert(0) += 1;
+        }
+
+        let mut max = 0;
+        let mut old_current_size = None;
+        for (observation, count) in counts {
+            if count > max {
+                max = count;
+                old_current_size = Some(observation);
+            }
+        }
+
+        if max >= write_quorum { old_current_size } else { None }
     }
 
     pub(in crate::set_disk) fn reduce_common_versions(disk_versions: &[Option<Vec<u8>>], write_quorum: usize) -> Option<Vec<u8>> {
@@ -4664,6 +4702,76 @@ mod tests {
         assert!(!r.has_residue());
     }
 
+    // rustfs/backlog#1009: quorum reduction of the per-disk old-current-size
+    // observations returned by rename_data.
+    mod reduce_common_old_current_size {
+        use super::*;
+
+        #[test]
+        fn agreement_at_quorum_wins() {
+            let observations = vec![
+                Some(OldCurrentSize::Present(5)),
+                Some(OldCurrentSize::Present(5)),
+                Some(OldCurrentSize::Present(5)),
+                Some(OldCurrentSize::Absent),
+            ];
+            assert_eq!(
+                SetDisks::reduce_common_old_current_size(&observations, 3),
+                Some(OldCurrentSize::Present(5))
+            );
+        }
+
+        #[test]
+        fn absent_is_a_definite_vote() {
+            let observations = vec![
+                Some(OldCurrentSize::Absent),
+                Some(OldCurrentSize::Absent),
+                Some(OldCurrentSize::Absent),
+                None,
+            ];
+            assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), Some(OldCurrentSize::Absent));
+        }
+
+        #[test]
+        fn divergence_below_quorum_is_unknown() {
+            let observations = vec![
+                Some(OldCurrentSize::Present(5)),
+                Some(OldCurrentSize::Present(7)),
+                Some(OldCurrentSize::Absent),
+                Some(OldCurrentSize::Absent),
+            ];
+            assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), None);
+        }
+
+        #[test]
+        fn unknown_disks_do_not_vote() {
+            // Two agreeing disks plus two unknowns must not fabricate quorum.
+            let observations = vec![Some(OldCurrentSize::Present(5)), Some(OldCurrentSize::Present(5)), None, None];
+            assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), None);
+        }
+
+        /// Kills the "unknown counts as an Absent vote" mutant: a full set of
+        /// unknowns (rolling upgrade, every peer pre-#1009) must stay unknown —
+        /// fabricating `Absent` would record "new object" on every overwrite
+        /// and inflate objects_count.
+        #[test]
+        fn all_unknown_is_unknown() {
+            let observations: Vec<Option<OldCurrentSize>> = vec![None; 4];
+            assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), None);
+        }
+
+        #[test]
+        fn unknown_majority_does_not_become_absent_quorum() {
+            let observations = vec![Some(OldCurrentSize::Present(5)), None, None, None];
+            assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), None);
+        }
+
+        #[test]
+        fn empty_observations_are_unknown() {
+            assert_eq!(SetDisks::reduce_common_old_current_size(&[], 2), None);
+        }
+    }
+
     // A2: not-found normalized to success (parity with MinIO commitRenameDataDir).
     #[test]
     fn cleanup_receipt_not_found_counts_as_reclaimed() {
diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs
index eb1e003fb..7053e401d 100644
--- a/crates/ecstore/src/set_disk/ops/multipart.rs
+++ b/crates/ecstore/src/set_disk/ops/multipart.rs
@@ -1420,7 +1420,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
 
-        let (online_disks, versions, op_old_dir, cleanup_disks) = Self::rename_data(
+        // 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, versions, op_old_dir, cleanup_disks, _) = Self::rename_data(
             &shuffle_disks,
             RUSTFS_META_MULTIPART_BUCKET,
             &upload_id_path,
diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs
index 5890ab906..91b162e18 100644
--- a/crates/ecstore/src/set_disk/ops/object.rs
+++ b/crates/ecstore/src/set_disk/ops/object.rs
@@ -21,6 +21,8 @@
 
 use super::super::*;
 
+use crate::disk::OldCurrentSize;
+
 #[async_trait::async_trait]
 impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
     type Error = Error;
@@ -556,8 +558,29 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         Ok(reader)
     }
 
-    #[tracing::instrument(skip(self, data,))]
     async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
+        self.put_object_with_old_current_size(bucket, object, data, opts)
+            .await
+            .map(|(object_info, _)| object_info)
+    }
+}
+
+impl SetDisks {
+    /// `put_object` plus the destination key's previous current-version size,
+    /// quorum-reduced from the dst `xl.meta` copies `rename_data` reads while
+    /// committing (rustfs/backlog#1009). `None` means unknown (mixed-version
+    /// peers, unparsable metadata, or sub-quorum divergence) — callers must
+    /// fall back to degraded accounting, never assume "absent". The extra
+    /// value is deliberately *not* part of `ObjectInfo`, which feeds S3
+    /// responses, event payloads, replication, and ILM verbatim.
+    #[tracing::instrument(skip(self, data,))]
+    pub async fn put_object_with_old_current_size(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: &mut PutObjReader,
+        opts: &ObjectOptions,
+    ) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
         crate::hp_guard!("SetDisks::put_object");
         self.invalidate_get_object_metadata_cache(bucket, object).await;
 
@@ -629,7 +652,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
 
         let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap());
 
-        let result: Result<ObjectInfo> = async {
+        let result: Result<(ObjectInfo, Option<OldCurrentSize>)> = async {
             let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
 
             let put_object_size = known_put_object_storage_size(data.size());
@@ -888,7 +911,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
             }
 
             let rename_stage_start = Instant::now();
-            let (online_disks, _, op_old_dir, cleanup_disks) = Self::rename_data(
+            let (online_disks, _, op_old_dir, cleanup_disks, old_current_size) = Self::rename_data(
                 &shuffle_disks,
                 RUSTFS_META_TMP_BUCKET,
                 tmp_dir.as_str(),
@@ -1023,7 +1046,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 );
             }
 
-            Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
+            Ok((
+                ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended),
+                old_current_size,
+            ))
         }
         .await;
 
diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs
index cab791ce0..3fa2b0dd2 100644
--- a/crates/ecstore/src/store/mod.rs
+++ b/crates/ecstore/src/store/mod.rs
@@ -386,9 +386,31 @@ impl crate::storage_api_contracts::object::ObjectIO for ECStore {
     }
     #[instrument(level = "debug", skip(self, data))]
     async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
-        let result =
-            enqueue_transition_after_write(self.handle_put_object(bucket, object, data, opts).await, LcEventSrc::S3PutObject)
-                .await;
+        self.put_object_with_old_current_size(bucket, object, data, opts)
+            .await
+            .map(|(object_info, _)| object_info)
+    }
+}
+
+impl ECStore {
+    /// `put_object` plus the rename_data old-size backfill
+    /// (rustfs/backlog#1009); see `SetDisks::put_object_with_old_current_size`.
+    /// Post-write hooks (immediate ILM transition enqueue, list-cache
+    /// invalidation) match the plain `put_object` path exactly.
+    #[instrument(level = "debug", skip(self, data))]
+    pub async fn put_object_with_old_current_size(
+        &self,
+        bucket: &str,
+        object: &str,
+        data: &mut PutObjReader,
+        opts: &ObjectOptions,
+    ) -> Result<(ObjectInfo, Option<crate::disk::OldCurrentSize>)> {
+        let result = match self.handle_put_object(bucket, object, data, opts).await {
+            Ok((object_info, old_current_size)) => enqueue_transition_after_write(Ok(object_info), LcEventSrc::S3PutObject)
+                .await
+                .map(|object_info| (object_info, old_current_size)),
+            Err(err) => Err(err),
+        };
         if result.is_ok() {
             list_objects::observe_list_objects_mutation(self, bucket).await;
         }
diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs
index a0e03187b..6b75f25c7 100644
--- a/crates/ecstore/src/store/object.rs
+++ b/crates/ecstore/src/store/object.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::*;
+use crate::disk::OldCurrentSize;
 use crate::set_disk::{
     get_lock_acquire_timeout, get_object_lock_diag_slow_acquire_threshold, get_object_lock_diag_slow_hold_threshold,
     is_lock_optimization_enabled, is_object_lock_diag_enabled,
@@ -718,7 +719,7 @@ impl ECStore {
         object: &str,
         data: &mut PutObjReader,
         opts: &ObjectOptions,
-    ) -> Result<ObjectInfo> {
+    ) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
         check_put_object_args(bucket, object)?;
 
         let object = encode_dir_object(object);
@@ -726,7 +727,9 @@ impl ECStore {
         // Keep PUT atomic-read friendly: SetDisks takes the object write lock only
         // around precondition checks and the final rename/commit.
         if self.single_pool() {
-            return self.pools[0].put_object(bucket, object.as_str(), data, opts).await;
+            return self.pools[0]
+                .put_object_with_old_current_size(bucket, object.as_str(), data, opts)
+                .await;
         }
 
         let idx = if opts.data_movement && opts.version_id.is_some() {
@@ -744,7 +747,9 @@ impl ECStore {
             ));
         }
 
-        self.pools[idx].put_object(bucket, &object, data, opts).await
+        self.pools[idx]
+            .put_object_with_old_current_size(bucket, &object, data, opts)
+            .await
     }
 
     #[instrument(skip(self))]
diff --git a/rustfs/src/app/mod.rs b/rustfs/src/app/mod.rs
index 433ef79f6..ca73f337f 100644
--- a/rustfs/src/app/mod.rs
+++ b/rustfs/src/app/mod.rs
@@ -32,3 +32,5 @@ mod capacity_dirty_scope_test;
 mod delete_objects_stat_gating_test;
 #[cfg(test)]
 mod lifecycle_transition_api_test;
+#[cfg(test)]
+mod put_prelookup_gating_test;
diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs
index 77e92548a..d32b5ea40 100644
--- a/rustfs/src/app/object_usecase.rs
+++ b/rustfs/src/app/object_usecase.rs
@@ -19,7 +19,6 @@ use rustfs_io_metrics::buffered_write;
 
 use crate::storage_api::table::get_bucket_metadata;
 
-use super::storage_api::object_usecase::ECStore;
 use super::storage_api::object_usecase::access::{
     PostObjectRequestMarker, authorize_request, has_bypass_governance_header, req_info_mut,
 };
@@ -64,7 +63,7 @@ use super::storage_api::object_usecase::contract::object::{ObjectIO as _, Object
 use super::storage_api::object_usecase::contract::range::HTTPRangeSpec;
 use super::storage_api::object_usecase::data_usage::{
     record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory,
-    record_bucket_object_write_memory,
+    record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory,
 };
 use super::storage_api::object_usecase::deadlock_detector;
 use super::storage_api::object_usecase::ecfs::FS;
@@ -93,6 +92,7 @@ use super::storage_api::object_usecase::sse::{
 };
 use super::storage_api::object_usecase::storage_class as storageclass;
 use super::storage_api::object_usecase::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper};
+use super::storage_api::object_usecase::{ECStore, OldCurrentSize};
 use super::storage_api::object_usecase::{
     RFC1123, check_preconditions, get_validated_store, has_replication_rules, parse_object_lock_legal_hold,
     parse_object_lock_retention, parse_part_number_i32_to_usize, remove_object_lock_metadata_for_copy,
@@ -3510,26 +3510,42 @@ impl DefaultObjectUsecase {
         )
         .await?;
 
-        let current_opts: ObjectOptions = internal_object_info_lookup_opts(
-            get_opts(&bucket, &key, version_id.clone(), None, &req.headers)
-                .await
-                .map_err(ApiError::from)?,
-        );
-        let previous_current_info = {
-            crate::hp_guard!("S3::put_object_prelookup");
-            store.get_object_info(&bucket, &key, &current_opts).await
-        };
-        let previous_current_size = match previous_current_info {
-            Ok(existing_obj_info) => {
-                validate_existing_object_lock_for_write(&existing_obj_info, &opts)?;
-                Some(existing_obj_info.size.max(0) as u64)
-            }
-            Err(err) => {
-                if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
-                    return Err(ApiError::from(err).into());
+        // rustfs/backlog#1009: the pre-PUT lookup has exactly two consumers —
+        // the existing-object WORM validation and usage accounting's
+        // previous_current_size. When the bucket has no object locking (WORM is
+        // a provable no-op; the gate fails closed on metadata errors) and the
+        // PUT targets the latest version (no explicit version_id from internal
+        // replication), the lookup is skipped and accounting is backfilled from
+        // the dst xl.meta that rename_data already reads, saving a full-disk
+        // metadata fanout per PUT.
+        let prelookup_required = version_id.is_some() || put_prelookup_worm_gate(&bucket).await;
+        // Outer None = prelookup skipped (accounting comes from the commit
+        // backfill); Some(inner) = the previous current size as observed by the
+        // lookup, with the pre-#1009 semantics kept bit-for-bit.
+        let prelookup_previous_current_size: Option<Option<u64>> = if prelookup_required {
+            let current_opts: ObjectOptions = internal_object_info_lookup_opts(
+                get_opts(&bucket, &key, version_id.clone(), None, &req.headers)
+                    .await
+                    .map_err(ApiError::from)?,
+            );
+            let previous_current_info = {
+                crate::hp_guard!("S3::put_object_prelookup");
+                store.get_object_info(&bucket, &key, &current_opts).await
+            };
+            Some(match previous_current_info {
+                Ok(existing_obj_info) => {
+                    validate_existing_object_lock_for_write(&existing_obj_info, &opts)?;
+                    Some(existing_obj_info.size.max(0) as u64)
                 }
-                None
-            }
+                Err(err) => {
+                    if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
+                        return Err(ApiError::from(err).into());
+                    }
+                    None
+                }
+            })
+        } else {
+            None
         };
 
         let actual_size = size;
@@ -3713,8 +3729,8 @@ impl DefaultObjectUsecase {
             }
         });
 
-        let obj_info = match store
-            .put_object(&bucket, &key, &mut reader, &opts)
+        let (obj_info, backfilled_old_current_size) = match store
+            .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts)
             .await
             .map_err(ApiError::from)
         {
@@ -3764,11 +3780,35 @@ impl DefaultObjectUsecase {
         let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &key).await;
 
         let put_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
-        // Fast in-memory update for immediate quota and admin usage consistency
-        if put_versioned {
-            record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
-        } else {
-            record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
+        // Fast in-memory update for immediate quota and admin usage consistency.
+        // The previous current size comes from the prelookup when it ran,
+        // otherwise from the rename_data backfill (rustfs/backlog#1009); the
+        // backfill reproduces the lookup's observation bit for bit (latest
+        // version's ObjectInfo.size — 0 for a delete-marker latest — or
+        // not-found → None).
+        match prelookup_previous_current_size.or_else(|| previous_current_size_from_backfill(backfilled_old_current_size)) {
+            Some(previous_current_size) => {
+                if put_versioned {
+                    record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
+                } else {
+                    record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
+                }
+            }
+            None => {
+                // Neither source could determine the previous state (peers
+                // predating the backfill field during a rolling upgrade, or
+                // sub-quorum metadata divergence). Record the components that
+                // are correct regardless; the next authoritative scanner
+                // refresh replaces the in-memory numbers.
+                debug!(
+                    target: "rustfs::app::object_usecase",
+                    bucket = %bucket,
+                    key = %key,
+                    put_versioned,
+                    "put_object old-size backfill unknown; recording degraded usage delta"
+                );
+                record_bucket_object_write_unknown_previous_memory(&bucket, obj_info.size.max(0) as u64, put_versioned).await;
+            }
         }
 
         let raw_version = obj_info.version_id.map(|v| v.to_string());
@@ -6567,6 +6607,28 @@ async fn bucket_object_locking_enabled(bucket: &str) -> bool {
         .is_ok_and(|metadata| metadata.object_locking())
 }
 
+/// Fail-closed WORM gate for skipping the pre-PUT lookup
+/// (rustfs/backlog#1009): a bucket-metadata read failure counts as "locking
+/// enabled" so the existing-object lock validation can never silently
+/// disappear on a degraded metadata subsystem.
+pub(super) async fn put_prelookup_worm_gate(bucket: &str) -> bool {
+    match get_bucket_metadata(bucket).await {
+        Ok(metadata) => metadata.object_locking(),
+        Err(_) => true,
+    }
+}
+
+/// rustfs/backlog#1009: map the rename_data old-size backfill onto the
+/// `previous_current_size` value the usage-accounting helpers expect. Outer
+/// `None` = unknown (no quorum agreement, or a peer predates the field) — the
+/// caller must fall back to the degraded accounting path.
+fn previous_current_size_from_backfill(backfill: Option<OldCurrentSize>) -> Option<Option<u64>> {
+    backfill.map(|observation| match observation {
+        OldCurrentSize::Present(size) => Some(size.max(0) as u64),
+        OldCurrentSize::Absent => None,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -6662,6 +6724,19 @@ mod tests {
         object_info_with_lock_metadata(metadata)
     }
 
+    /// rustfs/backlog#1009: the backfill→accounting mapping must mirror the
+    /// prelookup exactly — a live latest version maps to `Some(size)` (clamped
+    /// at 0 like the prelookup's `.max(0)`), absent/delete-marker maps to
+    /// `None`, and an unknown backfill maps to outer `None` so the caller
+    /// takes the degraded path instead of fabricating "new object".
+    #[test]
+    fn previous_current_size_from_backfill_mirrors_prelookup_semantics() {
+        assert_eq!(previous_current_size_from_backfill(Some(OldCurrentSize::Present(42))), Some(Some(42)));
+        assert_eq!(previous_current_size_from_backfill(Some(OldCurrentSize::Present(-7))), Some(Some(0)));
+        assert_eq!(previous_current_size_from_backfill(Some(OldCurrentSize::Absent)), Some(None));
+        assert_eq!(previous_current_size_from_backfill(None), None);
+    }
+
     #[test]
     fn validate_existing_object_lock_allows_versioned_new_version_with_compliance_retention() {
         let opts = ObjectOptions {
diff --git a/rustfs/src/app/put_prelookup_gating_test.rs b/rustfs/src/app/put_prelookup_gating_test.rs
new file mode 100644
index 000000000..754f8d469
--- /dev/null
+++ b/rustfs/src/app/put_prelookup_gating_test.rs
@@ -0,0 +1,150 @@
+// 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.
+
+//! Regression coverage for rustfs/backlog#1009: the PUT path skips its
+//! pre-PUT `get_object_info` only when `put_prelookup_worm_gate` proves the
+//! existing-object WORM validation is a no-op. These tests pin the gate's
+//! truth table against a real 4-disk `ECStore` with the bucket metadata sys
+//! initialized, mirroring the HP-8 delete-gating fixture:
+//!
+//! - a bucket created with Object Lock keeps the prelookup (gate = true);
+//! - a plain bucket takes the skip path (gate = false);
+//! - an unknown bucket (metadata lookup error) fails closed (gate = true),
+//!   so a degraded metadata subsystem can never silently drop the WORM check.
+
+use super::object_usecase::put_prelookup_worm_gate;
+use super::storage_api::test::bucket::metadata_sys;
+use super::storage_api::test::contract::bucket::{BucketOperations, BucketOptions, MakeBucketOptions};
+use super::storage_api::test::{ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints};
+use serial_test::serial;
+use std::path::PathBuf;
+use std::sync::{Arc, OnceLock};
+use tempfile::TempDir;
+use tokio::fs;
+use tokio_util::sync::CancellationToken;
+use uuid::Uuid;
+
+static PUT_GATING_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>, TempDir)> = OnceLock::new();
+
+async fn setup_put_gating_env() -> Arc<ECStore> {
+    if let Some((_paths, store, _)) = PUT_GATING_ENV.get() {
+        return store.clone();
+    }
+
+    let temp_dir = TempDir::new().expect("create temp dir for put prelookup gating test");
+    let temp_path = temp_dir.path().to_path_buf();
+
+    let disk_paths = vec![
+        temp_path.join("disk1"),
+        temp_path.join("disk2"),
+        temp_path.join("disk3"),
+        temp_path.join("disk4"),
+    ];
+    for disk_path in &disk_paths {
+        fs::create_dir_all(disk_path).await.unwrap();
+    }
+
+    let mut endpoints = Vec::new();
+    for (i, disk_path) in disk_paths.iter().enumerate() {
+        let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(i);
+        endpoints.push(endpoint);
+    }
+
+    let pool_endpoints = PoolEndpoints {
+        legacy: false,
+        set_count: 1,
+        drives_per_set: 4,
+        endpoints: Endpoints::from(endpoints),
+        cmd_line: "put-prelookup-gating-test".to_string(),
+        platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
+    };
+
+    let endpoint_pools = EndpointServerPools(vec![pool_endpoints]);
+    super::storage_api::test::runtime::init_local_disks(endpoint_pools.clone())
+        .await
+        .unwrap();
+
+    let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
+    let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new())
+        .await
+        .unwrap();
+
+    let buckets_list = ecstore
+        .list_bucket(&BucketOptions {
+            no_metadata: true,
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+    let buckets = buckets_list.into_iter().map(|v| v.name).collect();
+    metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await;
+
+    let _ = PUT_GATING_ENV.set((disk_paths, ecstore.clone(), temp_dir));
+    ecstore
+}
+
+#[tokio::test]
+#[serial]
+async fn worm_gate_keeps_prelookup_for_object_lock_bucket() {
+    let ecstore = setup_put_gating_env().await;
+    let bucket = format!("put-gate-lock-{}", Uuid::new_v4());
+
+    ecstore
+        .make_bucket(
+            &bucket,
+            &MakeBucketOptions {
+                lock_enabled: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("create object-lock bucket");
+
+    assert!(
+        put_prelookup_worm_gate(&bucket).await,
+        "an object-lock bucket must keep the pre-PUT lookup"
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn worm_gate_allows_skip_for_plain_bucket() {
+    let ecstore = setup_put_gating_env().await;
+    let bucket = format!("put-gate-plain-{}", Uuid::new_v4());
+
+    ecstore
+        .make_bucket(&bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create plain bucket");
+
+    assert!(
+        !put_prelookup_worm_gate(&bucket).await,
+        "a bucket without object locking must take the prelookup-skip path"
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn worm_gate_fails_closed_when_bucket_metadata_is_unavailable() {
+    let _ecstore = setup_put_gating_env().await;
+    let missing_bucket = format!("put-gate-missing-{}", Uuid::new_v4());
+
+    assert!(
+        put_prelookup_worm_gate(&missing_bucket).await,
+        "a bucket-metadata lookup failure must fail closed and keep the pre-PUT lookup"
+    );
+}
diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs
index 2ba0635e0..084b5a64b 100644
--- a/rustfs/src/app/storage_api.rs
+++ b/rustfs/src/app/storage_api.rs
@@ -117,6 +117,19 @@ pub(crate) mod data_usage {
         .await;
     }
 
+    pub(crate) async fn record_bucket_object_write_unknown_previous_memory(
+        bucket: &str,
+        new_size: u64,
+        creates_new_version: bool,
+    ) {
+        crate::storage::storage_api::ecstore_data_usage::record_bucket_object_write_unknown_previous_memory(
+            bucket,
+            new_size,
+            creates_new_version,
+        )
+        .await;
+    }
+
     pub(crate) async fn remove_bucket_usage_from_backend(
         store: Arc<crate::storage::storage_api::ECStore>,
         bucket: &str,
@@ -982,9 +995,9 @@ pub(crate) mod object_usecase {
         object_utils, options, request_context, s3_api, set_disk, sse, storage_class, timeout_wrapper,
     };
     pub(crate) use crate::storage::storage_api::{
-        ECStore, RFC1123, StorageDeletedObject, StorageObjectInfo, StorageObjectLockDeleteOptions, StorageObjectOptions,
-        StorageObjectToDelete, StoragePutObjReader, check_preconditions, get_validated_store, has_replication_rules,
-        parse_object_lock_legal_hold, parse_object_lock_retention, parse_part_number_i32_to_usize,
+        ECStore, OldCurrentSize, RFC1123, StorageDeletedObject, StorageObjectInfo, StorageObjectLockDeleteOptions,
+        StorageObjectOptions, StorageObjectToDelete, StoragePutObjReader, check_preconditions, get_validated_store,
+        has_replication_rules, parse_object_lock_legal_hold, parse_object_lock_retention, parse_part_number_i32_to_usize,
         remove_object_lock_metadata_for_copy, strip_managed_encryption_metadata, validate_bucket_object_lock_enabled,
         validate_object_key, validate_sse_headers_for_read, validate_sse_headers_for_write, validate_ssec_for_read,
         wrap_response_with_cors,
diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs
index 0e8a9211f..30012fb0f 100644
--- a/rustfs/src/storage/storage_api.rs
+++ b/rustfs/src/storage/storage_api.rs
@@ -360,9 +360,9 @@ pub(crate) mod ecstore_data_usage {
     pub(crate) use rustfs_ecstore::api::data_usage::{
         apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend, load_data_usage_from_backend,
         record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory,
-        record_bucket_object_write_memory, refresh_bucket_usage_from_object_layer,
-        refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend,
-        replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
+        record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory,
+        refresh_bucket_usage_from_object_layer, refresh_versioned_bucket_usage_from_object_layer,
+        remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
     };
 }
 
@@ -370,8 +370,8 @@ pub(crate) mod ecstore_data_usage {
 pub(crate) mod ecstore_disk {
     pub(crate) use rustfs_ecstore::api::disk::{
         BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskStore,
-        FileInfoVersions, FileReader, FileWriter, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions,
-        RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout,
+        FileInfoVersions, FileReader, FileWriter, OldCurrentSize, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp,
+        ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout,
         validate_batch_read_version_item_count,
     };
     pub(crate) use rustfs_ecstore::api::disk::{endpoint, error, error_reduce};
@@ -531,6 +531,7 @@ pub(crate) type BatchReadVersionResp = ecstore_disk::BatchReadVersionResp;
 pub(crate) type ReadMultipleReq = ecstore_disk::ReadMultipleReq;
 pub(crate) type ReadMultipleResp = ecstore_disk::ReadMultipleResp;
 pub(crate) type ReadOptions = ecstore_disk::ReadOptions;
+pub(crate) type OldCurrentSize = ecstore_disk::OldCurrentSize;
 pub(crate) type RenameDataResp = ecstore_disk::RenameDataResp;
 pub(crate) type ReplicationStatusType = ecstore_bucket::replication::ReplicationStatusType;
 pub(crate) type ReplicationStats = StorageReplicationStatsHandle;