Merge remote-tracking branch 'origin/main' into overtrue/docs-1923-free-version-disposition

* test(ecstore): cover suspended-owner heal semantics

* test(heal): cover suspended owner production path

crates/data-usage/src/data_usage.rs

@@ -307,38 +307,6 @@ pub struct DiskUsageStatus {
     pub snapshot_exists: bool,
 }
 
-/// A bounded reconciliation record for an object whose logical size could not
-/// be trusted at the scanner boundary. The scanner persists these records in
-/// its cache; keeping the model here avoids a second, incompatible accounting
-/// representation in storage-facing crates.
-#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
-pub struct SizeReconciliationEntry {
-    /// Stable object/version identity key (not a metrics label).
-    pub key: String,
-    pub bucket: String,
-    pub object: String,
-    #[serde(default)]
-    pub version_id: Option<String>,
-    #[serde(default)]
-    pub generation: Option<String>,
-    /// Structured reason label; raw metadata values must never be stored here.
-    pub reason: String,
-    #[serde(default)]
-    pub physical_size: Option<u64>,
-    #[serde(default)]
-    pub first_seen: u64,
-    #[serde(default)]
-    pub attempts: u32,
-}
-
-/// Object scope refreshed by one scanner pass. Existing debts in this scope
-/// are removed before the pass's unresolved records are inserted.
-#[derive(Debug, Default, Clone, PartialEq, Eq)]
-pub struct SizeReconciliationScope {
-    pub bucket: String,
-    pub object: String,
-}
-
 /// Size summary for a single object or group of objects
 #[derive(Debug, Default, Clone)]
 pub struct SizeSummary {
@@ -368,16 +336,6 @@ pub struct SizeSummary {
     pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
     /// Per-tier accounting, keyed by storage class or remote tier name
     pub tier_stats: HashMap<String, TierStats>,
-    /// Size-resolution debts observed while scanning this summary.
-    pub size_reconciliation: Vec<SizeReconciliationEntry>,
-    /// True when the per-object summary exceeded its bounded debt buffer.
-    /// Callers must retain prior ledger entries rather than treating the
-    /// partial list as a complete refresh.
-    pub size_reconciliation_truncated: bool,
-    /// Object scopes refreshed by this summary. They let the durable ledger
-    /// remove versions that resolved without allocating one key per healthy
-    /// version on the hot path.
-    pub reconciliation_scopes: Vec<SizeReconciliationScope>,
 }
 
 /// Replication target size summary
@@ -872,8 +830,7 @@ impl DataUsageEntry {
 ///
 /// The canonical wire format is written by the hand-written map-encoded
 /// `Serialize` on the scanner-side `DataUsageCacheInfo`
-/// (`crates/scanner/src/data_usage_define.rs`), which carries the original 16
-/// fields plus an optional reconciliation field.
+/// (`crates/scanner/src/data_usage_define.rs`), which carries 16 fields.
 /// This type decodes only the shared subset and is deliberately not
 /// `Serialize`: a derived (array) encoding of this 6-field subset would
 /// corrupt the cache for scanner readers, so no write path may exist here.
@@ -1817,51 +1774,6 @@ impl SizeSummary {
             entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
             entry.failed_count = entry.failed_count.saturating_add(stats.failed_count);
         }
-
-        for entry in &other.size_reconciliation {
-            self.record_size_reconciliation(entry.clone());
-        }
-        self.size_reconciliation_truncated |= other.size_reconciliation_truncated;
-        for scope in &other.reconciliation_scopes {
-            self.record_reconciliation_scope(&scope.bucket, &scope.object);
-        }
-    }
-
-    /// Add one reconciliation debt, coalescing repeated observations in the
-    /// same object summary. The scanner cache applies its own larger bound.
-    pub fn record_size_reconciliation(&mut self, entry: SizeReconciliationEntry) {
-        const MAX_SUMMARY_RECONCILIATION_ENTRIES: usize = 1024;
-        if let Some(existing) = self.size_reconciliation.iter_mut().find(|value| value.key == entry.key) {
-            existing.reason = entry.reason;
-            existing.physical_size = entry.physical_size;
-            existing.generation = entry.generation;
-            existing.version_id = entry.version_id;
-            return;
-        }
-        if self.size_reconciliation.len() < MAX_SUMMARY_RECONCILIATION_ENTRIES {
-            self.size_reconciliation.push(entry);
-        } else {
-            self.size_reconciliation_truncated = true;
-        }
-    }
-
-    /// Mark one object scope as refreshed. Duplicate scopes are suppressed so
-    /// merging summaries remains bounded and deterministic.
-    pub fn record_reconciliation_scope(&mut self, bucket: &str, object: &str) {
-        if !self
-            .reconciliation_scopes
-            .iter()
-            .any(|scope| scope.bucket == bucket && scope.object == object)
-        {
-            if self.reconciliation_scopes.len() >= 1024 {
-                self.size_reconciliation_truncated = true;
-                return;
-            }
-            self.reconciliation_scopes.push(SizeReconciliationScope {
-                bucket: bucket.to_string(),
-                object: object.to_string(),
-            });
-        }
     }
 }
 

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -3425,6 +3425,44 @@ mod tests {
         assert!(mutexes.contains_key("second"));
     }
 
+    #[tokio::test]
+    async fn update_all_targets_publishes_disable_proxy_on_target_client() {
+        // The read-proxy selector (replication_proxy::get_proxy_targets) skips
+        // targets whose TargetClient carries disable_proxy — the persisted
+        // per-target opt-out must survive client publication.
+        let sys = BucketTargetSys::default();
+        let target = |arn: &str, disable_proxy: bool| BucketTarget {
+            arn: arn.to_string(),
+            endpoint: "192.168.1.10:9000".to_string(),
+            target_bucket: "target-bucket".to_string(),
+            region: "us-east-1".to_string(),
+            disable_proxy,
+            credentials: Some(Credentials {
+                access_key: "access".to_string(),
+                secret_key: "secret".to_string(),
+                session_token: None,
+                expiration: None,
+            }),
+            ..Default::default()
+        };
+        let targets = BucketTargets {
+            targets: vec![target("arn:proxied", false), target("arn:opted-out", true)],
+        };
+
+        sys.update_all_targets("bucket", Some(&targets)).await;
+
+        let proxied = sys
+            .get_remote_target_client("bucket", "arn:proxied")
+            .await
+            .expect("client should be published");
+        assert!(!proxied.disable_proxy);
+        let opted_out = sys
+            .get_remote_target_client("bucket", "arn:opted-out")
+            .await
+            .expect("client should be published");
+        assert!(opted_out.disable_proxy, "disable_proxy must reach the published TargetClient");
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn target_updates_serialize_client_build_through_publication_per_bucket() {
         let sys = Arc::new(BucketTargetSys::default());

crates/ecstore/src/core/pools.rs

@@ -1130,6 +1130,23 @@ fn should_cleanup_decommission_source_entry(decommissioned: usize, total_version
     decommissioned.saturating_add(expired) == total_versions
 }
 
+/// Disposition reason logged for tier free-version records that decommission
+/// skips instead of migrating.
+const DECOMMISSION_FREE_VERSION_SKIP_REASON: &str = "tier_free_version_not_migrated";
+
+/// Counts the tier free-version records present in a decommission entry
+/// inventory. The exact loader (`load_file_info_versions_exact`) keeps these
+/// records inline in `versions` instead of separating them into
+/// `free_versions`, and the migration loop then routes them through the
+/// generic delete-marker path: the free-version flag and its remote-tier
+/// identity are never carried to the target pool, and a lone record is skipped
+/// by the empty-delete-marker rule. Accounting for them here keeps the final
+/// sweep from silently omitting records whose free-version disposition was
+/// dropped (see docs/architecture/decommission-compatibility.md).
+fn decommission_free_versions_skipped(fivs: &FileInfoVersions) -> usize {
+    fivs.versions.iter().filter(|version| version.tier_free_version()).count()
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[allow(
     dead_code,
@@ -3097,6 +3114,22 @@ impl ECStore {
         fivs.versions
             .sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time)));
 
+        let skipped_free_versions = decommission_free_versions_skipped(&fivs);
+        if skipped_free_versions > 0 {
+            debug!(
+                event = EVENT_DECOMMISSION_ENTRY,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_POOLS,
+                pool_index = idx,
+                bucket = %bucket,
+                object = %entry.name,
+                skipped_free_versions,
+                reason = DECOMMISSION_FREE_VERSION_SKIP_REASON,
+                state = "free_versions_skipped",
+                "Decommission skipped free-version migration"
+            );
+        }
+
         let mut decommissioned: usize = 0;
         let mut expired: usize = 0;
         let mut cleanup_preflight_allowed_missing = Vec::new();
@@ -5458,11 +5491,12 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi
 #[cfg(test)]
 mod pools_tests {
     use super::{
-        DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF,
-        DecomBucketInfo, DecommissionStartPoolState, DecommissionTerminalState, ListCallback, PoolDecommissionInfo, PoolMeta,
-        PoolSpaceInfo, PoolStatus, apply_decommission_status_space_info, bind_decommission_cancelers,
-        bind_missing_decommission_cancelers, cancel_decommission_canceler, classify_decommission_terminal_state,
-        count_decommission_item, decommission_cancel_signal_result, decommission_item_size, decommission_meta_bucket_options,
+        DECOMMISSION_FREE_VERSION_SKIP_REASON, DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD,
+        DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF, DecomBucketInfo, DecommissionStartPoolState, DecommissionTerminalState,
+        ListCallback, PoolDecommissionInfo, PoolMeta, PoolSpaceInfo, PoolStatus, apply_decommission_status_space_info,
+        bind_decommission_cancelers, bind_missing_decommission_cancelers, cancel_decommission_canceler,
+        classify_decommission_terminal_state, count_decommission_item, decommission_cancel_signal_result,
+        decommission_free_versions_skipped, decommission_item_size, decommission_meta_bucket_options,
         decommission_start_pool_state, dedup_indices, default_decommission_bucket_concurrency,
         ensure_decommission_cancel_allowed, ensure_decommission_clear_allowed, ensure_decommission_listing_disks_available,
         ensure_decommission_not_rebalancing, ensure_decommission_start_allowed, ensure_decommission_start_keeps_active_pool,
@@ -6762,6 +6796,27 @@ mod pools_tests {
         assert!(!should_cleanup_decommission_source_entry(2, 2, 1));
     }
 
+    #[test]
+    fn decommission_free_version_accounting_reports_skipped_records() {
+        let mut fivs = FileInfoVersions::default();
+        assert_eq!(decommission_free_versions_skipped(&fivs), 0);
+
+        fivs.versions.push(FileInfo {
+            name: "object.txt".to_string(),
+            ..Default::default()
+        });
+        let mut free_one = FileInfo::default();
+        free_one.set_tier_free_version();
+        fivs.versions.push(free_one);
+        let mut free_two = FileInfo::default();
+        free_two.set_tier_free_version();
+        free_two.transition_tier = "WARM".to_string();
+        fivs.versions.push(free_two);
+
+        assert_eq!(decommission_free_versions_skipped(&fivs), 2);
+        assert_eq!(DECOMMISSION_FREE_VERSION_SKIP_REASON, "tier_free_version_not_migrated");
+    }
+
     #[test]
     fn test_pool_meta_update_after_rejects_out_of_range_index() {
         let mut meta = PoolMeta::default();

crates/ecstore/src/store/heal.rs

@@ -297,10 +297,16 @@ impl ECStore {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::bucket::metadata_sys;
     use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
-    use crate::disk::{DiskOption, format::FormatV3, new_disk};
-    use crate::layout::endpoints::{Endpoints, PoolEndpoints};
+    use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk};
+    use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
+    use crate::runtime::instance::InstanceContext;
+    use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
+    use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
     use crate::store::init_format::{load_format_erasure, save_format_file};
+    use crate::store::init_local_disks_with_instance_ctx;
+    use tokio_util::sync::CancellationToken;
 
     async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
         let format = FormatV3::new(1, 1);
@@ -347,6 +353,51 @@ mod tests {
         }
     }
 
+    async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc<ECStore>, CancellationToken) {
+        let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created");
+        let mut pool_endpoints = Vec::new();
+        for pool_index in 0..2 {
+            let mut endpoints = Vec::new();
+            for disk_index in 0..4 {
+                let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
+                tokio::fs::create_dir_all(&disk_path)
+                    .await
+                    .expect("multi-pool heal test disk should be created");
+                let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8"))
+                    .expect("test endpoint should parse");
+                endpoint.set_pool_index(pool_index);
+                endpoint.set_set_index(0);
+                endpoint.set_disk_index(disk_index);
+                endpoints.push(endpoint);
+            }
+            pool_endpoints.push(PoolEndpoints {
+                legacy: false,
+                set_count: 1,
+                drives_per_set: 4,
+                endpoints: Endpoints::from(endpoints),
+                cmd_line: format!("heal-owner-pool-{pool_index}"),
+                platform: "test".to_string(),
+            });
+        }
+
+        let endpoint_pools = EndpointServerPools::from(pool_endpoints);
+        let instance_ctx = Arc::new(InstanceContext::new());
+        init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
+            .await
+            .expect("multi-pool local disks should initialize");
+        let shutdown = CancellationToken::new();
+        let store = ECStore::new_with_instance_ctx(
+            "127.0.0.1:0".parse().expect("test address should parse"),
+            endpoint_pools,
+            shutdown.clone(),
+            instance_ctx,
+        )
+        .await
+        .expect("multi-pool test store should initialize");
+        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        (temp_dir, store, shutdown)
+    }
+
     #[tokio::test]
     async fn heal_object_pool_scope_selects_only_requested_pool() {
         let store = minimal_heal_store().await;
@@ -506,6 +557,229 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn unscoped_heal_object_suspended_owner_semantics() {
+        let (_temp_dir, store, shutdown) = multi_pool_heal_store().await;
+        let bucket = format!("heal-owner-{}", Uuid::new_v4().simple());
+        let active_object = "active-owner";
+        let suspended_only_object = "suspended-only";
+        let duplicate_object = "duplicate-owner";
+        let marker_object = "marker-owner";
+        let quorum_object = "quorum-owner";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created in all pools");
+
+        let mut active_reader = PutObjReader::from_vec(b"active owner".to_vec());
+        store.pools[0]
+            .put_object(&bucket, active_object, &mut active_reader, &ObjectOptions::default())
+            .await
+            .expect("active owner object should be written");
+        let active_disks = store.pools[0].disk_set[0].disks.read().await.clone();
+        let missing_active_disk = active_disks[0].clone().expect("active disk should be online");
+        missing_active_disk
+            .delete(
+                &bucket,
+                active_object,
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("active owner shard should be removed for repair");
+        assert!(
+            missing_active_disk.read_xl(&bucket, active_object, false).await.is_err(),
+            "the active owner fixture must start with one missing metadata copy"
+        );
+
+        let mut suspended_reader = PutObjReader::from_vec(b"suspended owner".to_vec());
+        store.pools[1]
+            .put_object(&bucket, suspended_only_object, &mut suspended_reader, &ObjectOptions::default())
+            .await
+            .expect("suspended owner object should be written");
+        for (pool_index, mod_time) in [1_i64, 2_i64].into_iter().enumerate() {
+            let mut duplicate_reader = PutObjReader::from_vec(format!("duplicate-pool-{pool_index}").into_bytes());
+            store.pools[pool_index]
+                .put_object(
+                    &bucket,
+                    duplicate_object,
+                    &mut duplicate_reader,
+                    &ObjectOptions {
+                        mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(mod_time)),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("duplicate owner object should be written");
+        }
+        let duplicate_missing_disk = store.pools[0].disk_set[0].disks.read().await[0]
+            .clone()
+            .expect("duplicate active owner disk should be online");
+        duplicate_missing_disk
+            .delete(
+                &bucket,
+                duplicate_object,
+                DeleteOptions {
+                    recursive: true,
+                    immediate: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("duplicate active owner shard should be removed for repair");
+        let history_version = Uuid::new_v4();
+        let mut history_reader = PutObjReader::from_vec(b"marker history".to_vec());
+        store.pools[0]
+            .put_object(
+                &bucket,
+                marker_object,
+                &mut history_reader,
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(history_version.to_string()),
+                    mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("versioned marker history should be written");
+        store.pools[0]
+            .delete_object(
+                &bucket,
+                marker_object,
+                ObjectOptions {
+                    versioned: true,
+                    mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(2)),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("delete marker should be written");
+        let mut quorum_reader = PutObjReader::from_vec(b"quorum boundary".to_vec());
+        store.pools[0]
+            .put_object(&bucket, quorum_object, &mut quorum_reader, &ObjectOptions::default())
+            .await
+            .expect("quorum boundary object should be written");
+        {
+            let mut pool_meta = store.pool_meta.write().await;
+            let mut next = PoolMeta::new(&store.pools, &pool_meta);
+            next.pools[1].decommission = Some(PoolDecommissionInfo {
+                start_time: Some(OffsetDateTime::UNIX_EPOCH),
+                ..Default::default()
+            });
+            *pool_meta = next;
+        }
+
+        let (_, duplicate_owner) = store
+            .get_latest_object_info_with_idx(&bucket, duplicate_object, &ObjectOptions::default())
+            .await
+            .expect("duplicate owner should resolve");
+        assert_eq!(duplicate_owner, 1, "latest duplicate must win when all pools are eligible");
+        let (_, active_duplicate_owner) = store
+            .get_latest_object_info_with_idx(
+                &bucket,
+                duplicate_object,
+                &ObjectOptions {
+                    skip_decommissioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("active duplicate owner should resolve");
+        assert_eq!(
+            active_duplicate_owner, 0,
+            "suspended duplicate must be excluded from active owner selection"
+        );
+        let (duplicate_result, duplicate_err) = store
+            .handle_heal_object(&bucket, duplicate_object, "", &HealOpts::default())
+            .await
+            .expect("duplicate owner heal should complete through the production path");
+        assert_eq!(duplicate_result.object, duplicate_object);
+        assert!(duplicate_err.is_none(), "active duplicate should be repaired: {duplicate_err:?}");
+        assert!(
+            duplicate_missing_disk.read_xl(&bucket, duplicate_object, false).await.is_ok(),
+            "production heal must repair the active duplicate owner rather than the suspended owner"
+        );
+        let (marker_info, marker_owner) = store
+            .get_latest_object_info_with_idx(
+                &bucket,
+                marker_object,
+                &ObjectOptions {
+                    skip_decommissioned: true,
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("latest delete marker should resolve");
+        assert_eq!(marker_owner, 0);
+        assert!(marker_info.delete_marker, "latest version must preserve delete-marker semantics");
+
+        let (active_result, active_err) = store
+            .handle_heal_object(&bucket, active_object, "", &HealOpts::default())
+            .await
+            .expect("unscoped active-owner heal should complete");
+        assert_eq!(active_result.object, active_object);
+        assert!(active_err.is_none(), "active owner must be selected even with a suspended pool");
+        assert!(
+            missing_active_disk.read_xl(&bucket, active_object, false).await.is_ok(),
+            "active owner heal must write the missing disk metadata: result={active_result:?}, err={active_err:?}"
+        );
+        assert!(
+            store.pools[1]
+                .get_object_info(&bucket, active_object, &ObjectOptions::default())
+                .await
+                .is_err(),
+            "the suspended pool must not be written for an active-owner object"
+        );
+
+        let (suspended_result, suspended_err) = store
+            .handle_heal_object(&bucket, suspended_only_object, "", &HealOpts::default())
+            .await
+            .expect("unscoped suspended-only heal should return a terminal result");
+        assert!(suspended_result.object.is_empty());
+        assert!(matches!(suspended_err, Some(Error::FileNotFound)));
+        assert!(
+            store.pools[1]
+                .get_object_info(&bucket, suspended_only_object, &ObjectOptions::default())
+                .await
+                .is_ok(),
+            "suspended-only data must remain untouched when unscoped heal reports absent"
+        );
+
+        let (_, explicit_err) = store
+            .handle_heal_object(
+                &bucket,
+                suspended_only_object,
+                "",
+                &HealOpts {
+                    pool: Some(1),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("explicit suspended-owner heal should return a mapped error");
+        assert!(matches!(explicit_err, Some(Error::SlowDown)));
+
+        let original_quorum_disks = store.pools[0].disk_set[0].disks.read().await.clone();
+        let surviving_quorum_disk = original_quorum_disks[3].clone();
+        *store.pools[0].disk_set[0].disks.write().await = vec![None, None, None, surviving_quorum_disk];
+        let (_, quorum_err) = store
+            .handle_heal_object(&bucket, quorum_object, "", &HealOpts::default())
+            .await
+            .expect("quorum boundary heal should return a mapped result");
+        *store.pools[0].disk_set[0].disks.write().await = original_quorum_disks;
+        assert!(
+            matches!(quorum_err, Some(Error::ErasureReadQuorum)),
+            "quorum-boundary heal must preserve quorum error, got {quorum_err:?}"
+        );
+        shutdown.cancel();
+    }
+
     #[tokio::test]
     async fn handle_heal_format_continues_after_a_pool_error() {
         let canonical_format = FormatV3::new(1, 3);

crates/filemeta/src/filemeta.rs

@@ -90,6 +90,22 @@ fn legacy_data_key_for_version(version_id: Option<Uuid>) -> Option<String> {
 pub const TRANSITION_COMPLETE: &str = "complete";
 pub const TRANSITION_PENDING: &str = "pending";
 
+/// xl.meta key marking a tier free-version record.
+///
+/// A free version is a delete-marker-shaped cleanup hint appended by
+/// [`MetaObject::delete_version`] when a version whose remote transition
+/// completed is removed from xl.meta; it carries the remote tier identity for
+/// an idempotent remote delete and is never a user-visible version
+/// (`num_versions` excludes it). While the record exists it is consumed by the
+/// lifecycle free-version recovery scan and the usage scanner, which re-enqueue
+/// the pending remote delete, and by heal metadata walks. On S3 and lifecycle
+/// delete paths the same obligation is also carried by a committed tier-journal
+/// entry; deletes without such an entry (for example a removed version whose
+/// transition state decodes as unknown) rely on this record alone until the
+/// worker removes it after a successful remote delete. Decommission does not
+/// preserve these semantics: its exact inventory keeps the records inline in
+/// `versions` and the migration loop treats them as ordinary delete markers —
+/// see docs/architecture/decommission-compatibility.md.
 pub const FREE_VERSION: &str = "free-version";
 
 pub const TRANSITION_STATUS: &str = "transition-status";

crates/filemeta/src/filemeta/version.rs

@@ -2725,6 +2725,15 @@ impl MetaObject {
         self.meta_sys.retain(|k, _| !k.starts_with("X-Amz-Restore"));
     }
 
+    /// Builds the free-version cleanup record appended when a transitioned
+    /// version is removed from xl.meta. The record keeps the remote tier
+    /// identity so the lifecycle worker can issue the idempotent remote delete
+    /// and only then remove the record; until then the recovery scan and the
+    /// usage scanner keep re-enqueueing it. S3 and lifecycle deletes also
+    /// persist a committed tier-journal entry for the same remote delete, so a
+    /// record destroyed without its remote delete (as decommission does when it
+    /// treats these records as ordinary delete markers) strands only the
+    /// journal-less cases — see docs/architecture/decommission-compatibility.md.
     pub fn init_free_version(&self, fi: &FileInfo) -> Result<(FileMetaVersion, bool)> {
         if fi.skip_tier_free_version() {
             return Ok((FileMetaVersion::default(), false));

crates/replication/src/config.rs

@@ -60,7 +60,9 @@ pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[
     "Destination.ReplicationTime",
 ];
 
-pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 1;
+// v2: disableProxy moved from unsupported to writable (per-target read-proxy
+// opt-out is accepted by set-remote-target and the `proxy` update op).
+pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 2;
 
 pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
     "sourcebucket",
@@ -83,9 +85,12 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
     // madmin default of 60s); the per-target health-check interval is not
     // yet applied — the heartbeat keeps its global env-configured interval.
     "healthCheckDuration",
+    // Per-target read-proxy opt-out, consumed by the proxy-target selector
+    // (contract v2; previously only importable via MinIO bucket-targets.json).
+    "disableProxy",
 ];
 
-pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["disableProxy", "edge", "edgeSyncBeforeExpiry"];
+pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["edge", "edgeSyncBeforeExpiry"];
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct ObjectOpts {

crates/scanner/src/data_usage_define.rs

@@ -29,8 +29,7 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 pub use rustfs_data_usage::{
     AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
     DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
-    PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeReconciliationEntry, SizeReconciliationScope, SizeSummary,
-    TierStats, hash_path, prefix_usage_in_cache,
+    PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use tokio::time::{Duration, Instant, sleep, timeout};
@@ -193,10 +192,6 @@ const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
 pub trait ScannerSizeSummaryExt {
     /// Fold one object's contribution into the summary, including its tier.
     fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64);
-    /// Fold counters and physical tier usage for an object whose metadata is
-    /// valid but whose logical size is currently unavailable. Logical totals
-    /// stay unchanged.
-    fn actions_accounting_unknown(&mut self, oi: &ObjectInfo);
 }
 
 impl ScannerSizeSummaryExt for SizeSummary {
@@ -230,34 +225,6 @@ impl ScannerSizeSummaryExt for SizeSummary {
             });
         }
     }
-
-    fn actions_accounting_unknown(&mut self, oi: &ObjectInfo) {
-        if oi.delete_marker {
-            self.delete_markers = self.delete_markers.saturating_add(1);
-            return;
-        }
-
-        if oi.version_id.is_some_and(|v| !v.is_nil()) {
-            self.versions = self.versions.saturating_add(1);
-        }
-
-        if oi.transitioned_object.free_version {
-            return;
-        }
-
-        let tier = if oi.transitioned_object.status == TRANSITION_COMPLETE {
-            oi.transitioned_object.tier.clone()
-        } else {
-            oi.storage_class.clone().unwrap_or_else(|| storageclass::STANDARD.to_string())
-        };
-        if let Some(tier_stats) = self.tier_stats.get_mut(&tier) {
-            *tier_stats = tier_stats.add(&TierStats {
-                total_size: u64::try_from(oi.size).unwrap_or(0),
-                num_versions: 1,
-                num_objects: u64::from(oi.is_latest),
-            });
-        }
-    }
 }
 
 // ===== Cache-related data structures =====
@@ -377,10 +344,6 @@ pub struct DataUsageCacheInfo {
     pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
     #[serde(default)]
     pub cache_key_format: u16,
-    /// Bounded durable debts for versions whose logical size was not trusted.
-    /// The map key is an identity key, never a user-controlled metric label.
-    #[serde(default)]
-    pub size_reconciliation: HashMap<String, SizeReconciliationEntry>,
 }
 
 impl Serialize for DataUsageCacheInfo {
@@ -390,8 +353,7 @@ impl Serialize for DataUsageCacheInfo {
     {
         // Keep this metadata map-encoded so older readers can ignore fields
         // appended by newer scanner versions during rolling upgrades.
-        let field_count = 16 + usize::from(!self.size_reconciliation.is_empty());
-        let mut state = serializer.serialize_map(Some(field_count))?;
+        let mut state = serializer.serialize_map(Some(16))?;
         state.serialize_entry("name", &self.name)?;
         state.serialize_entry("next_cycle", &self.next_cycle)?;
         state.serialize_entry("leader_epoch", &self.leader_epoch)?;
@@ -408,9 +370,6 @@ impl Serialize for DataUsageCacheInfo {
         state.serialize_entry("snapshot_complete", &self.snapshot_complete)?;
         state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
         state.serialize_entry("cache_key_format", &self.cache_key_format)?;
-        if !self.size_reconciliation.is_empty() {
-            state.serialize_entry("size_reconciliation", &self.size_reconciliation)?;
-        }
         state.end()
     }
 }
@@ -469,18 +428,14 @@ impl DataUsageCache {
                     self.checked_flatten(name).is_some()
                 });
         if !reusable {
-            let (pending_heals, size_reconciliation) = if self.info.name == name {
-                (
-                    std::mem::take(&mut self.info.pending_heals),
-                    std::mem::take(&mut self.info.size_reconciliation),
-                )
+            let pending_heals = if self.info.name == name {
+                std::mem::take(&mut self.info.pending_heals)
             } else {
-                (Vec::new(), HashMap::new())
+                Vec::new()
             };
             *self = Self::default();
             self.info.name = name.to_string();
             self.info.pending_heals = pending_heals;
-            self.info.size_reconciliation = size_reconciliation;
         }
 
         self.info.next_cycle = next_cycle;

crates/scanner/src/data_usage_define/tests.rs

@@ -673,34 +673,6 @@ fn size_summary_actions_accounting_accumulates_tier_stats() {
     );
 }
 
-#[test]
-fn size_summary_unknown_accounting_keeps_physical_tier_and_version_only() {
-    let mut summary = SizeSummary::default();
-    summary
-        .tier_stats
-        .insert(storageclass::STANDARD.to_string(), TierStats::default());
-    let object = ObjectInfo {
-        size: 12,
-        storage_class: Some(storageclass::STANDARD.to_string()),
-        version_id: Some(uuid::Uuid::new_v4()),
-        is_latest: true,
-        ..Default::default()
-    };
-
-    summary.actions_accounting_unknown(&object);
-
-    assert_eq!(summary.total_size, 0, "unknown logical size must not become zero or physical bytes");
-    assert_eq!(summary.versions, 1);
-    assert_eq!(
-        summary.tier_stats.get(storageclass::STANDARD),
-        Some(&TierStats {
-            total_size: 12,
-            num_versions: 1,
-            num_objects: 1,
-        })
-    );
-}
-
 #[test]
 fn test_data_usage_entry_merge_sums_failed_objects() {
     let mut left = DataUsageEntry {
@@ -1107,16 +1079,6 @@ fn data_usage_cache_prepare_for_scan_preserves_pending_heal_only_progress() {
             scan_plan_digest: Some(TEST_PLAN_DIGEST),
             cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
             pending_heals: vec![pending_heal.clone()],
-            size_reconciliation: HashMap::from([(
-                "size-key".to_string(),
-                SizeReconciliationEntry {
-                    key: "size-key".to_string(),
-                    bucket: "bucket".to_string(),
-                    object: "prefix/object".to_string(),
-                    reason: "invalid_declared_size".to_string(),
-                    ..Default::default()
-                },
-            )]),
             ..Default::default()
         },
         ..Default::default()
@@ -1126,7 +1088,6 @@ fn data_usage_cache_prepare_for_scan_preserves_pending_heal_only_progress() {
 
     assert_eq!(outcome, DataUsageCachePrepareOutcome::Reused);
     assert_eq!(cache.info.pending_heals, vec![pending_heal]);
-    assert!(cache.info.size_reconciliation.contains_key("size-key"));
     assert!(cache.cache.is_empty());
     assert!(!cache.info.snapshot_complete);
 }

crates/scanner/src/scanner_folder.rs

@@ -20,9 +20,8 @@ use std::time::{Duration, Instant, SystemTime};
 
 use crate::ReplTargetSizeSummary;
 use crate::data_usage_define::{
-    DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap,
-    DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt,
-    SizeReconciliationEntry, SizeSummary, hash_path,
+    DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageScanCheckpoint,
+    DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, SizeSummary, hash_path,
 };
 use crate::error::ScannerError;
 use crate::runtime_config::{
@@ -98,9 +97,6 @@ const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders
 const METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total";
 const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_total";
 const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
-const MAX_SIZE_RECONCILIATION_ENTRIES_PER_BUCKET: usize = 10_000;
-const MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET: usize = 8 * 1024 * 1024;
-const MAX_SIZE_RECONCILIATION_AGE_SECS: u64 = 7 * 24 * 60 * 60;
 
 // --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
 //
@@ -368,7 +364,7 @@ impl PendingScannerAccounting<'_> {
     fn apply(self, size_summary: &mut SizeSummary, cumulative_size: &mut i64, queued: bool) {
         let size = if queued { self.expired_size } else { self.retained_size };
         size_summary.actions_accounting(self.object, size, self.retained_size);
-        *cumulative_size = cumulative_size.saturating_add(size);
+        *cumulative_size += size;
     }
 }
 
@@ -679,54 +675,6 @@ pub struct FolderScanner {
     list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
 }
 
-fn size_reconciliation_entry_bytes(entry: &SizeReconciliationEntry) -> usize {
-    entry.key.len()
-        + entry.bucket.len()
-        + entry.object.len()
-        + entry.version_id.as_deref().map_or(0, str::len)
-        + entry.generation.as_deref().map_or(0, str::len)
-        + entry.reason.len()
-        + std::mem::size_of::<u64>()
-        + std::mem::size_of::<u32>()
-}
-
-fn prune_size_reconciliation(info: &mut DataUsageCacheInfo, now: u64) {
-    info.size_reconciliation.retain(|key, entry| {
-        if entry.first_seen == 0 || entry.first_seen > now {
-            entry.first_seen = now;
-        }
-        key == &entry.key
-            && entry.key.len() <= 4096
-            && entry.bucket.len() <= 512
-            && entry.object.len() <= 512
-            && entry.version_id.as_deref().is_none_or(|value| value.len() <= 64)
-            && entry.generation.as_deref().is_none_or(|value| value.len() <= 64)
-            && entry.reason.len() <= 64
-            && now.saturating_sub(entry.first_seen) <= MAX_SIZE_RECONCILIATION_AGE_SECS
-    });
-
-    while info.size_reconciliation.len() > MAX_SIZE_RECONCILIATION_ENTRIES_PER_BUCKET
-        || info
-            .size_reconciliation
-            .values()
-            .map(size_reconciliation_entry_bytes)
-            .sum::<usize>()
-            > MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET
-    {
-        let oldest = info
-            .size_reconciliation
-            .iter()
-            .min_by(|(left_key, left), (right_key, right)| {
-                left.first_seen.cmp(&right.first_seen).then_with(|| left_key.cmp(right_key))
-            })
-            .map(|(key, _)| key.clone());
-        let Some(oldest) = oldest else {
-            break;
-        };
-        info.size_reconciliation.remove(&oldest);
-    }
-}
-
 impl FolderScanner {
     fn now_secs() -> u64 {
         SystemTime::now()
@@ -800,55 +748,6 @@ impl FolderScanner {
         }
     }
 
-    /// Apply the per-object size-resolution ledger updates in one place. The
-    /// scanner cache is the durable boundary; both working copies are updated
-    /// so an incremental publication cannot lose a debt or its resolution.
-    fn apply_size_reconciliation(&mut self, summary: &SizeSummary) {
-        let now = Self::now_secs();
-        // Keep an unresolved identity in place while refreshing its object
-        // scope. This lets repeated observations increment `attempts`; only
-        // debts absent from the current pass are considered resolved.
-        let current_keys = summary
-            .size_reconciliation
-            .iter()
-            .map(|entry| entry.key.clone())
-            .collect::<HashSet<_>>();
-        for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
-            prune_size_reconciliation(info, now);
-
-            if !summary.size_reconciliation_truncated {
-                for scope in &summary.reconciliation_scopes {
-                    let scope_bucket = item_actions::bounded_reconciliation_field(&scope.bucket);
-                    let scope_object = item_actions::bounded_reconciliation_field(&scope.object);
-                    info.size_reconciliation.retain(|key, entry| {
-                        entry.bucket != scope_bucket || entry.object != scope_object || current_keys.contains(key)
-                    });
-                }
-            }
-
-            for incoming in &summary.size_reconciliation {
-                if let Some(existing) = info.size_reconciliation.get_mut(&incoming.key) {
-                    existing.reason = incoming.reason.clone();
-                    existing.physical_size = incoming.physical_size;
-                    existing.generation = incoming.generation.clone();
-                    existing.version_id = incoming.version_id.clone();
-                    existing.attempts = existing.attempts.saturating_add(1);
-                    continue;
-                }
-
-                if size_reconciliation_entry_bytes(incoming) > MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET {
-                    continue;
-                }
-
-                let mut entry = incoming.clone();
-                entry.first_seen = now;
-                entry.attempts = 1;
-                info.size_reconciliation.insert(entry.key.clone(), entry);
-            }
-            prune_size_reconciliation(info, now);
-        }
-    }
-
     fn record_scan_resume_hint(&mut self, folder: &str) {
         self.new_cache.info.scan_resume_after = Some(folder.to_string());
         self.update_cache.info.scan_resume_after = Some(folder.to_string());
@@ -1527,7 +1426,6 @@ impl FolderScanner {
                 abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()]));
 
                 apply_scanner_size_summary(into, &sz);
-                self.apply_size_reconciliation(&sz);
                 into.objects += 1;
                 object_count += 1;
                 self.budget.record_object_scanned();
@@ -2296,10 +2194,6 @@ pub async fn scan_data_folder(
         list_path_raw_options_observer: None,
     };
 
-    let now = FolderScanner::now_secs();
-    prune_size_reconciliation(&mut scanner.new_cache.info, now);
-    prune_size_reconciliation(&mut scanner.update_cache.info, now);
-
     // Check if context is cancelled
     if ctx.is_cancelled() {
         return Err(ScannerError::Other("Operation cancelled".to_string()));
@@ -2323,9 +2217,7 @@ pub async fn scan_data_folder(
             new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
             new_cache.info.last_update = Some(SystemTime::now());
             new_cache.info.next_cycle = cache.info.next_cycle;
-            let unresolved_objects = root.failed_objects > 0
-                || !new_cache.info.failed_objects.is_empty()
-                || !new_cache.info.size_reconciliation.is_empty();
+            let unresolved_objects = root.failed_objects > 0 || !new_cache.info.failed_objects.is_empty();
             new_cache.info.snapshot_complete = !unresolved_objects;
             let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some();
             new_cache.info.scan_resume_after = None;
@@ -2353,7 +2245,7 @@ pub async fn scan_data_folder(
                 if root_has_progress {
                     new_cache.replace_hashed(&root_hash, &None, &root);
                 }
-                if partial_cache_is_useful(&root, pending_heals_changed) || !new_cache.info.size_reconciliation.is_empty() {
+                if partial_cache_is_useful(&root, pending_heals_changed) {
                     if new_cache.root().is_some() {
                         new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
                     }

crates/scanner/src/scanner_folder/item_actions.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 /// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers.
 use super::*;
-use sha2::{Digest as _, Sha256};
 
 /// Cached folder information for scanning
 #[derive(Clone, Debug)]
@@ -33,259 +32,6 @@ pub(super) enum GetSizeFailureAction {
     HealMetadata { object: String },
 }
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub(super) enum SizeResolutionReason {
-    CompressedSizeUnknown,
-    InvalidPhysicalSize,
-    UnsupportedCompression,
-    InvalidObjectSize,
-    InvalidPartSize,
-    InvalidDeclaredSize,
-    SizeOverflowOrMismatch,
-}
-
-impl SizeResolutionReason {
-    fn as_str(self) -> &'static str {
-        match self {
-            Self::CompressedSizeUnknown => "compressed_size_unknown",
-            Self::InvalidPhysicalSize => "invalid_physical_size",
-            Self::UnsupportedCompression => "unsupported_compression",
-            Self::InvalidObjectSize => "invalid_object_size",
-            Self::InvalidPartSize => "invalid_part_size",
-            Self::InvalidDeclaredSize => "invalid_declared_size",
-            Self::SizeOverflowOrMismatch => "size_overflow_or_mismatch",
-        }
-    }
-}
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub(super) enum SizeResolution {
-    Known { logical: i64, physical: i64 },
-    Unknown { physical: i64, reason: SizeResolutionReason },
-    Corrupt { physical: i64, reason: SizeResolutionReason },
-}
-
-impl SizeResolution {
-    fn known_size(&self) -> Option<i64> {
-        match self {
-            Self::Known { logical, .. } => Some(*logical),
-            Self::Unknown { .. } | Self::Corrupt { .. } => None,
-        }
-    }
-}
-
-fn size_reconciliation_key(oi: &ObjectInfo, reason: SizeResolutionReason) -> String {
-    let version = oi
-        .version_id
-        .filter(|version| !version.is_nil())
-        .map(|version| version.to_string())
-        .unwrap_or_default();
-    let generation = oi
-        .data_dir
-        .filter(|generation| !generation.is_nil())
-        .map(|generation| generation.to_string())
-        .unwrap_or_default();
-    // Length-prefix each component so an object key containing the separator
-    // cannot alias another identity. S3 keys are bounded in normal operation;
-    // oversized persisted values use a digest so a corrupt metadata record
-    // cannot grow the ledger without bound.
-    fn component(value: &str) -> String {
-        const MAX_COMPONENT_LEN: usize = 512;
-        if value.len() <= MAX_COMPONENT_LEN {
-            return format!("{}:{}", value.len(), value);
-        }
-        let digest = Sha256::digest(value.as_bytes());
-        let digest = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower);
-        format!("hash:{}:{}", value.len(), digest)
-    }
-    format!(
-        "{}|{}|{}|{}|{}",
-        component(&oi.bucket),
-        component(&oi.name),
-        component(&version),
-        component(&generation),
-        component(reason.as_str())
-    )
-}
-
-pub(super) fn bounded_reconciliation_field(value: &str) -> String {
-    const MAX_FIELD_LEN: usize = 512;
-    if value.len() <= MAX_FIELD_LEN {
-        return value.to_string();
-    }
-    let digest = hex_simd::encode_to_string(Sha256::digest(value.as_bytes()), hex_simd::AsciiCase::Lower);
-    let prefix_len = MAX_FIELD_LEN - 65;
-    let prefix = value
-        .char_indices()
-        .take_while(|(offset, ch)| offset.saturating_add(ch.len_utf8()) <= prefix_len)
-        .map(|(_, ch)| ch)
-        .collect::<String>();
-    format!("{}~{}", prefix, digest)
-}
-
-fn record_size_resolution(summary: &mut SizeSummary, oi: &ObjectInfo, resolution: &SizeResolution) {
-    match resolution {
-        SizeResolution::Known { .. } => {}
-        SizeResolution::Unknown { physical, reason } | SizeResolution::Corrupt { physical, reason } => {
-            summary.record_size_reconciliation(SizeReconciliationEntry {
-                key: size_reconciliation_key(oi, *reason),
-                bucket: bounded_reconciliation_field(&oi.bucket),
-                object: bounded_reconciliation_field(&oi.name),
-                version_id: oi
-                    .version_id
-                    .filter(|version| !version.is_nil())
-                    .map(|version| version.to_string()),
-                generation: oi
-                    .data_dir
-                    .filter(|generation| !generation.is_nil())
-                    .map(|generation| generation.to_string()),
-                reason: reason.as_str().to_string(),
-                physical_size: u64::try_from(*physical).ok(),
-                first_seen: 0,
-                attempts: 0,
-            });
-        }
-    }
-}
-
-/// Resolve the size metadata once at the scanner trust boundary. A compressed
-/// -1 sentinel is valid legacy metadata, but it cannot participate in normal
-/// logical-size accounting or size-filtered lifecycle rules.
-pub(super) fn resolve_size(oi: &ObjectInfo) -> SizeResolution {
-    let physical = oi.size;
-    if physical < 0 {
-        return SizeResolution::Corrupt {
-            physical,
-            reason: SizeResolutionReason::InvalidPhysicalSize,
-        };
-    }
-
-    let compressed = match oi.compression_read_plan() {
-        Ok((_, _, compressed)) => compressed,
-        Err(_) => {
-            return SizeResolution::Corrupt {
-                physical,
-                reason: SizeResolutionReason::UnsupportedCompression,
-            };
-        }
-    };
-
-    if oi.actual_size < -1 || (oi.actual_size == -1 && !compressed) {
-        return SizeResolution::Corrupt {
-            physical,
-            reason: SizeResolutionReason::InvalidObjectSize,
-        };
-    }
-
-    // Match ObjectInfo::get_actual_size: a positive in-memory value is the
-    // authoritative decoded size. Stale declared/part metadata must not turn
-    // an otherwise valid object into a false corruption report.
-    if oi.actual_size > 0 {
-        return SizeResolution::Known {
-            logical: oi.actual_size,
-            physical,
-        };
-    }
-
-    if oi
-        .parts
-        .iter()
-        .any(|part| part.actual_size < -1 || (part.actual_size < 0 && !compressed))
-    {
-        return SizeResolution::Corrupt {
-            physical,
-            reason: SizeResolutionReason::InvalidPartSize,
-        };
-    }
-
-    let declared = rustfs_utils::http::get_str(&oi.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE);
-    let declared = match declared {
-        Some(value) if value.is_empty() => {
-            return SizeResolution::Corrupt {
-                physical,
-                reason: SizeResolutionReason::InvalidDeclaredSize,
-            };
-        }
-        Some(value) => match value.parse::<i64>() {
-            Ok(value) if value >= 0 => Some(value),
-            _ => {
-                return SizeResolution::Corrupt {
-                    physical,
-                    reason: SizeResolutionReason::InvalidDeclaredSize,
-                };
-            }
-        },
-        None => None,
-    };
-
-    let logical = match oi.get_actual_size() {
-        Ok(size) if size == -1 && compressed && declared.is_none() => {
-            return SizeResolution::Unknown {
-                physical,
-                reason: SizeResolutionReason::CompressedSizeUnknown,
-            };
-        }
-        Ok(size) if size >= 0 => size,
-        Ok(_) | Err(_) => {
-            return SizeResolution::Corrupt {
-                physical,
-                reason: SizeResolutionReason::SizeOverflowOrMismatch,
-            };
-        }
-    };
-
-    if compressed && logical == 0 && physical != 0 && oi.parts.is_empty() && declared.is_none() {
-        return SizeResolution::Corrupt {
-            physical,
-            reason: SizeResolutionReason::SizeOverflowOrMismatch,
-        };
-    }
-
-    SizeResolution::Known { logical, physical }
-}
-
-fn resolve_sizes(object_infos: &[ObjectInfo]) -> Vec<SizeResolution> {
-    object_infos.iter().map(resolve_size).collect()
-}
-
-fn lifecycle_rule_has_size_filter(lifecycle: &BucketLifecycleConfiguration, rule_id: &str) -> bool {
-    let filter_has_size = |filter: &s3s::dto::LifecycleRuleFilter| {
-        filter.object_size_greater_than.is_some()
-            || filter.object_size_less_than.is_some()
-            || filter
-                .and
-                .as_ref()
-                .is_some_and(|and| and.object_size_greater_than.is_some() || and.object_size_less_than.is_some())
-    };
-    lifecycle
-        .rules
-        .iter()
-        .find(|rule| rule.id.as_deref().unwrap_or_default() == rule_id)
-        .and_then(|rule| rule.filter.as_ref())
-        .is_some_and(filter_has_size)
-}
-
-fn lifecycle_event_allowed(resolution: &SizeResolution, event: &Event, lifecycle: &BucketLifecycleConfiguration) -> bool {
-    match resolution {
-        // Corrupt metadata cannot safely authorize a destructive action, even
-        // when the evaluator happened to produce a time-only event.
-        SizeResolution::Corrupt { .. } => false,
-        // A valid-but-unknown logical size may still execute lifecycle
-        // actions whose rule is independent of object-size predicates. The
-        // evaluator has already selected the rule; only that rule's filter
-        // can make the missing logical value action-critical.
-        SizeResolution::Unknown { .. } => !lifecycle_rule_has_size_filter(lifecycle, &event.rule_id),
-        SizeResolution::Known { .. } => true,
-    }
-}
-
-/// A successful newer-noncurrent batch consumes both known and unresolved
-/// versions from the retained-version alert count. The two accounting paths
-/// are separate because only known sizes can contribute byte totals.
-fn remaining_versions_after_queued_noncurrent(remaining_versions: usize, known_count: usize, unknown_count: usize) -> usize {
-    remaining_versions.saturating_sub(known_count.saturating_add(unknown_count))
-}
-
 /// How the corrupt-metadata branch records the repair after attempting an
 /// MRF intent (backlog#1894 axis A).
 #[derive(Debug, PartialEq, Eq)]
@@ -573,45 +319,34 @@ impl ScannerItem {
             "Scanner lifecycle evaluation started"
         );
 
-        let resolved_sizes = resolve_sizes(&object_infos);
-        if let Some(first) = object_infos.first() {
-            size_summary.record_reconciliation_scope(
-                &bounded_reconciliation_field(&first.bucket),
-                &bounded_reconciliation_field(&first.name),
-            );
-        }
-        for (oi, resolution) in object_infos.iter().zip(resolved_sizes.iter()) {
-            record_size_resolution(size_summary, oi, resolution);
-        }
-        let has_corrupt_size = resolved_sizes
-            .iter()
-            .any(|resolution| matches!(resolution, SizeResolution::Corrupt { .. }));
-
         // `versioning_config` is resolved once per object by the caller
         // (`get_size`) and handed in; only `prefix_enabled` is consulted here.
 
-        let Some(lifecycle) = self.lifecycle.clone() else {
-            let mut cumulative_size: i64 = 0;
-            for (oi, resolved_size) in object_infos.iter().zip(resolved_sizes.iter()) {
-                let accounting_size = match resolved_size {
-                    SizeResolution::Known { logical, .. } => *logical,
-                    // A valid compressed legacy sentinel has no logical size,
-                    // but heal and replication still need to run. The
-                    // physical size is only an input to those operations; it
-                    // is not folded into the logical total below.
-                    SizeResolution::Unknown { physical, .. } => {
-                        self.heal_actions(oi, *physical, size_summary).await;
-                        size_summary.actions_accounting_unknown(oi);
+        let Some(lifecycle) = self.lifecycle.as_ref() else {
+            let mut cumulative_size = 0;
+            for oi in object_infos.iter() {
+                let actual_size = match oi.get_actual_size() {
+                    Ok(size) => size,
+                    Err(_) => {
+                        warn!(
+                            target: "rustfs::scanner::folder",
+                            event = EVENT_SCANNER_LIFECYCLE_ACTION,
+                            component = LOG_COMPONENT_SCANNER,
+                            subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                            bucket = %self.bucket,
+                            object = %oi.name,
+                            state = "size_lookup_failed",
+                            "Scanner lifecycle action used fallback size"
+                        );
                         continue;
                     }
-                    SizeResolution::Corrupt { .. } => continue,
                 };
 
-                let size = self.heal_actions(oi, accounting_size, size_summary).await;
+                let size = self.heal_actions(oi, actual_size, size_summary).await;
 
-                size_summary.actions_accounting(oi, size, accounting_size);
+                size_summary.actions_accounting(oi, size, actual_size);
 
-                cumulative_size = cumulative_size.saturating_add(size);
+                cumulative_size += size;
             }
 
             self.alert_excessive_versions(object_infos.len(), cumulative_size);
@@ -665,108 +400,25 @@ impl ScannerItem {
         let mut to_delete_objs: Vec<ObjectToDelete> = Vec::new();
         let mut noncurrent_events: Vec<Event> = Vec::new();
         let mut noncurrent_accounting: Vec<PendingScannerAccounting<'_>> = Vec::new();
-        let mut noncurrent_unknown: Vec<&ObjectInfo> = Vec::new();
         let mut cumulative_size = 0;
         let mut remaining_versions = object_infos.len();
         'eventLoop: {
             for (i, event) in events.iter().enumerate() {
                 let oi = &object_infos[i];
-                let known_size = resolved_sizes[i].known_size();
-                if has_corrupt_size
-                    && matches!(
-                        event.action,
-                        IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction
-                    )
-                {
-                    // An all-version delete would also remove a corrupt
-                    // sibling that could not be reconciled safely.
-                    continue;
-                }
-                if !lifecycle_event_allowed(&resolved_sizes[i], event, &lifecycle) {
-                    // An unknown logical size must not make an otherwise
-                    // non-destructive scan disappear from heal/physical-tier
-                    // accounting. Size-filtered or deferred events remain
-                    // pending, so retain the version-only physical counters.
-                    if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
-                        self.heal_actions(oi, *physical, size_summary).await;
-                        size_summary.actions_accounting_unknown(oi);
-                    }
-                    continue;
-                }
-                let actual_size = match known_size {
-                    Some(size) => size,
-                    None => {
-                        match event.action {
-                            IlmAction::DeleteAction
-                            | IlmAction::DeleteRestoredAction
-                            | IlmAction::DeleteRestoredVersionAction
-                            | IlmAction::DeleteAllVersionsAction
-                            | IlmAction::DelMarkerDeleteAllVersionsAction => {
-                                let done_ilm = Metrics::time_ilm(event.action);
-                                let trace_started_at = trace_start_instant();
-                                let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
-                                emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
-                                if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
-                                    done_ilm(1)();
-                                    if event.action == IlmAction::DeleteAllVersionsAction
-                                        || event.action == IlmAction::DelMarkerDeleteAllVersionsAction
-                                    {
-                                        remaining_versions = 0;
-                                    }
-                                } else if matches!(
-                                    event.action,
-                                    IlmAction::DeleteAction
-                                        | IlmAction::DeleteRestoredAction
-                                        | IlmAction::DeleteRestoredVersionAction
-                                ) {
-                                    size_summary.actions_accounting_unknown(oi);
-                                } else {
-                                    size_summary.actions_accounting_unknown(oi);
-                                    for (j, retained) in object_infos.iter().enumerate().skip(i + 1) {
-                                        match &resolved_sizes[j] {
-                                            SizeResolution::Known { logical, .. } => PendingScannerAccounting {
-                                                object: retained,
-                                                retained_size: *logical,
-                                                expired_size: 0,
-                                            }
-                                            .apply(size_summary, &mut cumulative_size, false),
-                                            SizeResolution::Unknown { .. } => {
-                                                size_summary.actions_accounting_unknown(retained);
-                                            }
-                                            SizeResolution::Corrupt { .. } => {}
-                                        }
-                                    }
-                                }
-                            }
-                            IlmAction::DeleteVersionAction => {
-                                if let Some(opt) = object_opts.get(i) {
-                                    to_delete_objs.push(ObjectToDelete {
-                                        object_name: opt.name.clone(),
-                                        version_id: opt.version_id,
-                                        ..Default::default()
-                                    });
-                                    noncurrent_events.push(event.clone());
-                                    noncurrent_unknown.push(oi);
-                                }
-                            }
-                            IlmAction::TransitionAction | IlmAction::TransitionVersionAction => {
-                                let trace_started_at = trace_start_instant();
-                                let queued = apply_transition_rule(event, &LcEventSrc::Scanner, oi).await;
-                                emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
-                                if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
-                                    let done_ilm = Metrics::time_ilm(event.action);
-                                    done_ilm(1)();
-                                }
-                                size_summary.actions_accounting_unknown(oi);
-                            }
-                            IlmAction::NoneAction | IlmAction::ActionCount => {
-                                if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
-                                    self.heal_actions(oi, *physical, size_summary).await;
-                                }
-                                size_summary.actions_accounting_unknown(oi);
-                            }
-                        }
-                        continue;
+                let actual_size = match oi.get_actual_size() {
+                    Ok(size) => size,
+                    Err(_) => {
+                        warn!(
+                            target: "rustfs::scanner::folder",
+                            event = EVENT_SCANNER_LIFECYCLE_ACTION,
+                            component = LOG_COMPONENT_SCANNER,
+                            subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                            bucket = %self.bucket,
+                            object = %oi.name,
+                            state = "size_lookup_failed",
+                            "Scanner lifecycle action used fallback size"
+                        );
+                        0
                     }
                 };
 
@@ -794,24 +446,36 @@ impl ScannerItem {
                             done_ilm(1)();
                             remaining_versions = 0;
                         } else {
-                            if let Some(actual_size) = known_size {
+                            PendingScannerAccounting {
+                                object: oi,
+                                retained_size: actual_size,
+                                expired_size: 0,
+                            }
+                            .apply(size_summary, &mut cumulative_size, false);
+                            for retained in object_infos.iter().skip(i + 1) {
+                                let retained_size = match retained.get_actual_size() {
+                                    Ok(size) => size,
+                                    Err(_) => {
+                                        warn!(
+                                            target: "rustfs::scanner::folder",
+                                            event = EVENT_SCANNER_LIFECYCLE_ACTION,
+                                            component = LOG_COMPONENT_SCANNER,
+                                            subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                                            bucket = %self.bucket,
+                                            object = %retained.name,
+                                            state = "size_lookup_failed",
+                                            "Scanner lifecycle action used fallback size"
+                                        );
+                                        0
+                                    }
+                                };
                                 PendingScannerAccounting {
-                                    object: oi,
-                                    retained_size: actual_size,
+                                    object: retained,
+                                    retained_size,
                                     expired_size: 0,
                                 }
                                 .apply(size_summary, &mut cumulative_size, false);
                             }
-                            for (j, retained) in object_infos.iter().enumerate().skip(i + 1) {
-                                if let Some(retained_size) = resolved_sizes[j].known_size() {
-                                    PendingScannerAccounting {
-                                        object: retained,
-                                        retained_size,
-                                        expired_size: 0,
-                                    }
-                                    .apply(size_summary, &mut cumulative_size, false);
-                                }
-                            }
                         }
                         break 'eventLoop;
                     }
@@ -847,13 +511,11 @@ impl ScannerItem {
                                 version_id: opt.version_id,
                                 ..Default::default()
                             });
-                            if let Some(actual_size) = known_size {
-                                noncurrent_accounting.push(PendingScannerAccounting {
-                                    object: oi,
-                                    retained_size: actual_size,
-                                    expired_size: 0,
-                                });
-                            }
+                            noncurrent_accounting.push(PendingScannerAccounting {
+                                object: oi,
+                                retained_size: actual_size,
+                                expired_size: 0,
+                            });
                             account_now = false;
                         }
                         noncurrent_events.push(event.clone());
@@ -886,7 +548,7 @@ impl ScannerItem {
 
                 if account_now {
                     size_summary.actions_accounting(oi, size, actual_size);
-                    cumulative_size = cumulative_size.saturating_add(size);
+                    cumulative_size += size;
                 }
             }
         }
@@ -914,20 +576,11 @@ impl ScannerItem {
             }
             if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) {
                 done_ilm(count)();
-                remaining_versions = remaining_versions_after_queued_noncurrent(
-                    remaining_versions,
-                    noncurrent_accounting.len(),
-                    noncurrent_unknown.len(),
-                );
+                remaining_versions = remaining_versions.saturating_sub(noncurrent_accounting.len());
             }
             for pending in noncurrent_accounting {
                 pending.apply(size_summary, &mut cumulative_size, queued);
             }
-            if !queued {
-                for object in noncurrent_unknown {
-                    size_summary.actions_accounting_unknown(object);
-                }
-            }
         }
         self.alert_excessive_versions(remaining_versions, cumulative_size);
     }
@@ -1276,361 +929,4 @@ mod tests {
         assert_eq!(item.object_name, "object");
         assert_eq!(item.object_path(), "object");
     }
-
-    #[test]
-    fn size_resolution_rejects_negative_overflow_and_unknown_compression() {
-        let compressed = |actual_size: i64, declared: Option<&str>| {
-            let mut user_defined = HashMap::new();
-            rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
-            if let Some(declared) = declared {
-                rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, declared.to_string());
-            }
-            ObjectInfo {
-                size: 12,
-                actual_size,
-                user_defined: Arc::new(user_defined),
-                ..Default::default()
-            }
-        };
-
-        let normal = ObjectInfo {
-            size: 12,
-            actual_size: 10,
-            ..Default::default()
-        };
-        assert_eq!(
-            resolve_size(&normal),
-            SizeResolution::Known {
-                logical: 10,
-                physical: 12
-            }
-        );
-
-        let stale_declared_metadata = ObjectInfo {
-            size: 12,
-            actual_size: 10,
-            user_defined: Arc::new(HashMap::from([("x-rustfs-internal-actual-size".to_string(), "not-a-size".to_string())])),
-            parts: Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
-                actual_size: -2,
-                ..Default::default()
-            }]),
-            ..Default::default()
-        };
-        assert_eq!(
-            resolve_size(&stale_declared_metadata),
-            SizeResolution::Known {
-                logical: 10,
-                physical: 12
-            }
-        );
-
-        assert_eq!(
-            resolve_size(&compressed(0, Some("9"))),
-            SizeResolution::Known {
-                logical: 9,
-                physical: 12
-            }
-        );
-        assert_eq!(
-            resolve_size(&compressed(-1, None)),
-            SizeResolution::Unknown {
-                physical: 12,
-                reason: SizeResolutionReason::CompressedSizeUnknown,
-            }
-        );
-        assert!(matches!(
-            resolve_size(&compressed(0, Some("not-a-size"))),
-            SizeResolution::Corrupt {
-                reason: SizeResolutionReason::InvalidDeclaredSize,
-                ..
-            }
-        ));
-        assert!(matches!(
-            resolve_size(&ObjectInfo {
-                size: 12,
-                actual_size: -2,
-                ..Default::default()
-            }),
-            SizeResolution::Corrupt { .. }
-        ));
-        assert!(matches!(resolve_size(&compressed(0, Some("-1"))), SizeResolution::Corrupt { .. }));
-        assert!(matches!(resolve_size(&compressed(0, Some(""))), SizeResolution::Corrupt { .. }));
-
-        let unsupported = {
-            let mut object = compressed(0, None);
-            let mut metadata = (*object.user_defined).clone();
-            rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "unsupported".to_string());
-            object.user_defined = Arc::new(metadata);
-            object
-        };
-        assert!(matches!(resolve_size(&unsupported), SizeResolution::Corrupt { .. }));
-
-        let invalid_part = {
-            let mut object = compressed(0, None);
-            object.parts = Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
-                size: 12,
-                actual_size: -2,
-                ..Default::default()
-            }]);
-            object
-        };
-        assert!(matches!(resolve_size(&invalid_part), SizeResolution::Corrupt { .. }));
-
-        let overflow = {
-            let mut object = compressed(0, None);
-            object.parts = Arc::new(vec![
-                rustfs_filemeta::ObjectPartInfo {
-                    size: 1,
-                    actual_size: i64::MAX,
-                    ..Default::default()
-                },
-                rustfs_filemeta::ObjectPartInfo {
-                    size: 1,
-                    actual_size: 1,
-                    ..Default::default()
-                },
-            ]);
-            object
-        };
-        assert!(matches!(resolve_size(&overflow), SizeResolution::Corrupt { .. }));
-
-        let mismatch = compressed(0, None);
-        assert!(matches!(resolve_size(&mismatch), SizeResolution::Corrupt { .. }));
-        assert_eq!(
-            resolve_size(&ObjectInfo {
-                size: 0,
-                actual_size: 0,
-                ..Default::default()
-            }),
-            SizeResolution::Known { logical: 0, physical: 0 }
-        );
-    }
-
-    #[test]
-    fn size_resolution_records_and_replays_one_identity() {
-        let version_id = uuid::Uuid::new_v4();
-        let generation = uuid::Uuid::new_v4();
-        let mut metadata = HashMap::new();
-        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
-        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "not-a-number".to_string());
-        let corrupt = ObjectInfo {
-            bucket: "bucket".to_string(),
-            name: "object".to_string(),
-            size: 12,
-            version_id: Some(version_id),
-            data_dir: Some(generation),
-            user_defined: Arc::new(metadata),
-            ..Default::default()
-        };
-
-        let mut summary = SizeSummary::default();
-        let resolution = resolve_size(&corrupt);
-        record_size_resolution(&mut summary, &corrupt, &resolution);
-        record_size_resolution(&mut summary, &corrupt, &resolution);
-        assert_eq!(summary.size_reconciliation.len(), 1);
-        assert_eq!(summary.size_reconciliation[0].reason, "invalid_declared_size");
-        assert_eq!(summary.size_reconciliation[0].physical_size, Some(12));
-
-        let known = ObjectInfo {
-            actual_size: 12,
-            user_defined: Arc::new(HashMap::new()),
-            ..corrupt.clone()
-        };
-        record_size_resolution(&mut summary, &known, &resolve_size(&known));
-        summary.record_reconciliation_scope(&known.bucket, &known.name);
-        assert_eq!(summary.reconciliation_scopes.len(), 1);
-        assert_eq!(summary.reconciliation_scopes[0].bucket, "bucket");
-    }
-
-    #[test]
-    fn malformed_size_has_same_ilm_accounting() {
-        let mut metadata = HashMap::new();
-        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
-        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "invalid".to_string());
-        let object = ObjectInfo {
-            bucket: "bucket".to_string(),
-            name: "object".to_string(),
-            size: 12,
-            user_defined: Arc::new(metadata),
-            ..Default::default()
-        };
-        let resolution = resolve_size(&object);
-        let mut without_ilm = SizeSummary::default();
-        let mut with_ilm = SizeSummary::default();
-        record_size_resolution(&mut without_ilm, &object, &resolution);
-        record_size_resolution(&mut with_ilm, &object, &resolution);
-        assert_eq!(without_ilm.size_reconciliation, with_ilm.size_reconciliation);
-        assert_eq!(without_ilm.total_size, 0);
-        assert_eq!(with_ilm.total_size, 0);
-        assert!(without_ilm.tier_stats.is_empty());
-        assert!(with_ilm.tier_stats.is_empty());
-    }
-
-    #[test]
-    fn size_resolution_parses_once_per_version() {
-        let objects = vec![
-            ObjectInfo {
-                bucket: "bucket".to_string(),
-                name: "one".to_string(),
-                size: 1,
-                actual_size: 1,
-                ..Default::default()
-            },
-            ObjectInfo {
-                bucket: "bucket".to_string(),
-                name: "two".to_string(),
-                size: 2,
-                actual_size: -2,
-                ..Default::default()
-            },
-        ];
-        let resolutions = resolve_sizes(&objects);
-        assert_eq!(resolutions.len(), objects.len());
-        assert!(matches!(resolutions[0], SizeResolution::Known { logical: 1, .. }));
-        assert!(matches!(resolutions[1], SizeResolution::Corrupt { .. }));
-    }
-
-    #[test]
-    fn queued_unknown_noncurrent_versions_are_removed_from_alert_count() {
-        assert_eq!(remaining_versions_after_queued_noncurrent(3, 1, 2), 0);
-        assert_eq!(remaining_versions_after_queued_noncurrent(7, 2, 1), 4);
-        assert_eq!(remaining_versions_after_queued_noncurrent(usize::MAX, usize::MAX, usize::MAX), 0);
-    }
-
-    #[test]
-    fn malformed_size_blocks_size_dependent_transition_but_allows_time_only_expiry() {
-        let size_filtered = BucketLifecycleConfiguration {
-            rules: vec![s3s::dto::LifecycleRule {
-                status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
-                expiration: None,
-                abort_incomplete_multipart_upload: None,
-                del_marker_expiration: None,
-                id: Some("size".to_string()),
-                filter: Some(s3s::dto::LifecycleRuleFilter {
-                    object_size_greater_than: Some(1),
-                    ..Default::default()
-                }),
-                noncurrent_version_expiration: None,
-                noncurrent_version_transitions: None,
-                prefix: None,
-                transitions: None,
-            }],
-            ..Default::default()
-        };
-        let unknown = SizeResolution::Unknown {
-            physical: 12,
-            reason: SizeResolutionReason::CompressedSizeUnknown,
-        };
-        let size_event = Event {
-            action: IlmAction::DeleteAction,
-            rule_id: "size".to_string(),
-            ..Default::default()
-        };
-        assert!(!lifecycle_event_allowed(&unknown, &size_event, &size_filtered));
-        assert!(!lifecycle_event_allowed(
-            &unknown,
-            &Event {
-                action: IlmAction::TransitionAction,
-                rule_id: "size".to_string(),
-                ..Default::default()
-            },
-            &size_filtered
-        ));
-        let mixed_filters = BucketLifecycleConfiguration {
-            rules: vec![
-                size_filtered.rules[0].clone(),
-                s3s::dto::LifecycleRule {
-                    status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
-                    expiration: None,
-                    abort_incomplete_multipart_upload: None,
-                    del_marker_expiration: None,
-                    id: Some("time".to_string()),
-                    filter: None,
-                    noncurrent_version_expiration: None,
-                    noncurrent_version_transitions: None,
-                    prefix: None,
-                    transitions: None,
-                },
-            ],
-            ..Default::default()
-        };
-        assert!(lifecycle_event_allowed(
-            &unknown,
-            &Event {
-                action: IlmAction::DeleteAction,
-                rule_id: "time".to_string(),
-                ..Default::default()
-            },
-            &mixed_filters
-        ));
-        assert!(lifecycle_event_allowed(
-            &unknown,
-            &Event {
-                action: IlmAction::TransitionAction,
-                ..Default::default()
-            },
-            &BucketLifecycleConfiguration::default()
-        ));
-        assert!(!lifecycle_event_allowed(
-            &SizeResolution::Corrupt {
-                physical: 12,
-                reason: SizeResolutionReason::InvalidDeclaredSize,
-            },
-            &Event {
-                action: IlmAction::DeleteAction,
-                ..Default::default()
-            },
-            &BucketLifecycleConfiguration::default()
-        ));
-        assert!(lifecycle_event_allowed(
-            &SizeResolution::Known {
-                logical: 10,
-                physical: 12,
-            },
-            &Event {
-                action: IlmAction::DeleteAllVersionsAction,
-                ..Default::default()
-            },
-            &BucketLifecycleConfiguration::default()
-        ));
-        assert!(lifecycle_event_allowed(
-            &unknown,
-            &Event {
-                action: IlmAction::DeleteAction,
-                rule_id: "time-only".to_string(),
-                ..Default::default()
-            },
-            &BucketLifecycleConfiguration::default()
-        ));
-    }
-
-    #[tokio::test]
-    async fn long_object_size_reconciliation_scope_uses_bounded_identity() {
-        let object_name = "o".repeat(600);
-        let mut item = scanner_item_with_prefix("");
-        item.object_name = object_name.clone();
-
-        let mut metadata = HashMap::new();
-        rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
-        let object = ObjectInfo {
-            bucket: item.bucket.clone(),
-            name: object_name.clone(),
-            size: 12,
-            actual_size: -1,
-            version_id: Some(uuid::Uuid::new_v4()),
-            user_defined: Arc::new(metadata),
-            ..Default::default()
-        };
-        let mut summary = SizeSummary::default();
-        item.apply_actions(vec![object], None, VersioningConfiguration::default(), &mut summary)
-            .await;
-
-        let bounded_bucket = bounded_reconciliation_field(&item.bucket);
-        let bounded_object = bounded_reconciliation_field(&object_name);
-        assert_eq!(summary.reconciliation_scopes[0].bucket, bounded_bucket);
-        assert_eq!(summary.reconciliation_scopes[0].object, bounded_object);
-        assert_eq!(summary.size_reconciliation[0].object, bounded_object);
-        assert_eq!(summary.versions, 1);
-        assert_eq!(summary.total_size, 0);
-    }
 }

crates/scanner/src/scanner_folder/tests.rs

@@ -388,66 +388,6 @@ async fn test_record_failed_ttl_zero_noop() {
     assert!(!scanner.should_skip_failed("path2"));
 }
 
-#[tokio::test]
-async fn malformed_size_reconciliation_replays_after_restart() {
-    let (mut scanner, temp_dir) = build_test_scanner().await;
-    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
-
-    let entry = SizeReconciliationEntry {
-        key: "1:b|6:object|0:|0:".to_string(),
-        bucket: "b".to_string(),
-        object: "object".to_string(),
-        reason: "invalid_declared_size".to_string(),
-        physical_size: Some(12),
-        ..Default::default()
-    };
-    let mut summary = SizeSummary::default();
-    summary.record_size_reconciliation(entry.clone());
-    summary.record_reconciliation_scope("b", "object");
-    scanner.apply_size_reconciliation(&summary);
-    scanner.apply_size_reconciliation(&summary);
-
-    assert_eq!(scanner.new_cache.info.size_reconciliation.len(), 1);
-    assert_eq!(scanner.update_cache.info.size_reconciliation.len(), 1);
-    assert_eq!(scanner.new_cache.info.size_reconciliation[&entry.key].attempts, 2);
-
-    let encoded = rmp_serde::to_vec_named(&scanner.new_cache.info).expect("size ledger should encode");
-    let decoded: crate::data_usage_define::DataUsageCacheInfo =
-        rmp_serde::from_slice(&encoded).expect("size ledger should decode");
-    assert_eq!(decoded.size_reconciliation.len(), 1);
-    assert_eq!(decoded.size_reconciliation[&entry.key].reason, "invalid_declared_size");
-
-    let mut resolved = SizeSummary::default();
-    resolved.record_reconciliation_scope("b", "object");
-    scanner.apply_size_reconciliation(&resolved);
-    assert!(scanner.new_cache.info.size_reconciliation.is_empty());
-    assert!(scanner.update_cache.info.size_reconciliation.is_empty());
-}
-
-#[tokio::test]
-async fn malformed_size_reconciliation_clears_bounded_long_object_scope() {
-    let (mut scanner, temp_dir) = build_test_scanner().await;
-    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
-    let long_object = "o".repeat(600);
-    let bounded_object = item_actions::bounded_reconciliation_field(&long_object);
-    let entry = SizeReconciliationEntry {
-        key: "long-object-key".to_string(),
-        bucket: "b".to_string(),
-        object: bounded_object,
-        reason: "invalid_declared_size".to_string(),
-        ..Default::default()
-    };
-    let mut summary = SizeSummary::default();
-    summary.record_size_reconciliation(entry);
-    scanner.apply_size_reconciliation(&summary);
-    assert_eq!(scanner.new_cache.info.size_reconciliation.len(), 1);
-
-    let mut resolved = SizeSummary::default();
-    resolved.record_reconciliation_scope("b", &long_object);
-    scanner.apply_size_reconciliation(&resolved);
-    assert!(scanner.new_cache.info.size_reconciliation.is_empty());
-}
-
 #[test]
 fn test_classify_get_size_failure_marks_metadata_heal_object_path() {
     let temp_dir = std::env::temp_dir();

docs/architecture/decommission-compatibility.md

@@ -153,6 +153,91 @@ No migration step is required for these decisions because this note documents th
 current RustFS behavior. Changing either decision later requires an operator
 compatibility note and updated characterization tests.
 
+## Tier Free Versions During Decommission
+
+A tier free version is an internal xl.meta record (`rustfs_filemeta::FREE_VERSION`,
+flagged `XL_FLAG_FREE_VERSION`) shaped like a delete marker. It is created by
+`MetaObject::init_free_version` when a version whose remote transition completed is
+deleted locally: the visible version is removed and the record keeps the remote-tier
+identity (tier, object name, version id, state, destination id) needed for an
+idempotent remote delete. Free versions are not user-visible versions; `num_versions`
+and all listing/GET paths exclude them.
+
+### Lifecycle And Consumers
+
+Creation: any local delete that removes a version whose transition status is
+`complete` appends the record via `MetaObject::delete_version` →
+`init_free_version` (skipped only when `skip_tier_free_version` is set, as on
+data-movement copies). The same deletes also persist a durable tier-journal
+entry on every user-facing path: S3 single deletes (`execute_delete_object` →
+`delete_object_with_tier_delete_journal`), S3 batch deletes, lifecycle expiry,
+and lifecycle delete-all all prepare and commit a journal entry around the
+delete. A journal entry is omitted when the removed version's transition state
+decodes as `TransitionVersionState::Unknown`, or on internal journal-less
+delete paths that never touch transitioned user objects.
+
+Consumption while the record exists: the background recovery loop started by
+`init_background_expiry` (spawned by `spawn_tier_free_version_recovery_once`,
+enabled by default) scans disks for pending records and re-enqueues them; the
+usage scanner does the same; the lifecycle worker then deletes the remote tier
+object idempotently and only afterwards removes the local record. Heal walks
+include free-version records in metadata healing. Transition planning,
+replication, restore, GET, listings, and usage aggregation never depend on
+them.
+
+### Decommission Handling
+
+The exact decommission inventory loader (`load_file_info_versions_exact` via
+`get_all_file_info_versions`) keeps free-version records inline in `versions`; it
+never populates `free_versions`, so the source-cleanup preflight comparison of
+`free_versions` is vacuous for decommission. The migration loop then routes every
+record through the generic delete-marker handling:
+
+- a record that is the only remaining version without replication is skipped by the
+  empty-delete-marker rule and counted as done;
+- any other record is copied to the target pool as an ordinary delete marker with the
+  same version id and mod time.
+
+In both cases the free-version flag and its remote-tier identity are dropped:
+decommission neither preserves free-version semantics nor performs or reschedules the
+pending remote-tier delete. Source cleanup then removes the original records together
+with the source xl.meta.
+
+Allowed physical-delete timing: the source record may be removed once the migration
+loop has dispositioned it (copied as a plain marker or skipped as lone), which
+happens regardless of whether its remote-tier delete was ever performed.
+
+### Reference-Audit Result
+
+No cluster-local consumer resolves a free version after decommission finishes: GET,
+listing, transition planning, replication, restore, and heal operate either on
+user-visible versions or while the record still exists. The remote exposure is
+bounded:
+
+- On every user-facing delete path the remote-delete obligation is durably carried
+  by the committed tier-journal entry, which the tier sweeper processes
+  independently of xl.meta; the free-version record is an idempotent second
+  pointer, not the only one. Dropping it during decommission therefore does not
+  orphan the remote object.
+- Residual exposure: for records whose version state decoded as `Unknown` no
+  journal entry exists, so dropping the unconsumed record loses that cleanup hint
+  and the remote-tier object is orphaned. The same applies to any future internal
+  delete path that removes transitioned versions without a journal entry.
+
+Copying a pending record as an ordinary delete marker also adds a user-visible
+tombstone to the target pool's version history that the source never exposed.
+
+Because of the residual journal-less case, decommission must account for every
+free-version record instead of omitting it silently:
+
+- `decommission_free_versions_skipped` counts the records per decommission entry;
+- entries with a non-zero count log `state = "free_versions_skipped"` with reason
+  `tier_free_version_not_migrated`.
+
+Regression guard:
+
+- `decommission_free_version_accounting_reports_skipped_records`
+
 ## Regression Guard
 
 The queued multi-pool contract is guarded by:

rustfs/src/admin/handlers/replication.rs

@@ -73,6 +73,8 @@ enum TargetUpdateOp {
     /// Connection group: credentials plus endpoint, target bucket, and TLS settings.
     Credentials,
     Sync,
+    /// Per-target read-proxy opt-out (`disableProxy`).
+    Proxy,
     Bandwidth,
     Path,
 }
@@ -81,12 +83,13 @@ fn parse_remote_target_update_ops(queries: &HashMap<String, String>) -> S3Result
     const SUPPORTED_OPS: &[(&str, TargetUpdateOp)] = &[
         ("creds", TargetUpdateOp::Credentials),
         ("sync", TargetUpdateOp::Sync),
+        ("proxy", TargetUpdateOp::Proxy),
         ("bandwidth", TargetUpdateOp::Bandwidth),
         ("path", TargetUpdateOp::Path),
     ];
     // Present in the MinIO wire contract, but they drive target fields this
     // version rejects as unsupported — fail loudly instead of silently ignoring.
-    const UNSUPPORTED_OPS: &[&str] = &["proxy", "healthcheck", "edge", "edgeSyncBeforeExpiry"];
+    const UNSUPPORTED_OPS: &[&str] = &["healthcheck", "edge", "edgeSyncBeforeExpiry"];
 
     for key in UNSUPPORTED_OPS {
         if queries.get(*key).is_some_and(|value| value == "true") {
@@ -312,11 +315,10 @@ impl RemoteTargetRequest {
             ));
         }
 
-        for (unsupported, configured) in
-            REMOTE_TARGET_UNSUPPORTED_FIELDS
-                .iter()
-                .copied()
-                .zip([self.disable_proxy, self.edge, self.edge_sync_before_expiry])
+        for (unsupported, configured) in REMOTE_TARGET_UNSUPPORTED_FIELDS
+            .iter()
+            .copied()
+            .zip([self.edge, self.edge_sync_before_expiry])
         {
             if configured {
                 return Err(s3_error!(
@@ -702,6 +704,7 @@ impl Operation for SetRemoteTargetHandler {
                         target.deployment_id = remote_target.deployment_id.clone();
                     }
                     TargetUpdateOp::Sync => target.replication_sync = remote_target.replication_sync,
+                    TargetUpdateOp::Proxy => target.disable_proxy = remote_target.disable_proxy,
                     TargetUpdateOp::Bandwidth => target.bandwidth_limit = remote_target.bandwidth_limit,
                     TargetUpdateOp::Path => target.path = remote_target.path.clone(),
                 }
@@ -1520,6 +1523,7 @@ mod tests {
             ("update", "true"),
             ("creds", "true"),
             ("sync", "true"),
+            ("proxy", "true"),
             ("bandwidth", "true"),
             ("path", "true"),
         ]))
@@ -1529,6 +1533,7 @@ mod tests {
             vec![
                 TargetUpdateOp::Credentials,
                 TargetUpdateOp::Sync,
+                TargetUpdateOp::Proxy,
                 TargetUpdateOp::Bandwidth,
                 TargetUpdateOp::Path
             ]
@@ -2070,7 +2075,6 @@ mod tests {
             ("credentials.session_token", serde_json::json!("session-token")),
             ("credentials.expiration", serde_json::json!("2026-01-01T00:00:00Z")),
             ("api", serde_json::json!("s3v2")),
-            ("disableProxy", serde_json::json!(true)),
             ("edge", serde_json::json!(true)),
             ("edgeSyncBeforeExpiry", serde_json::json!(true)),
         ] {
@@ -2300,6 +2304,44 @@ mod tests {
         assert!(!REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"healthCheckDuration"));
     }
 
+    #[test]
+    fn remote_target_disable_proxy_is_declared_writable_edge_stays_unsupported() {
+        assert!(REMOTE_TARGET_WRITABLE_FIELDS.contains(&"disableProxy"));
+        assert!(!REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"disableProxy"));
+        // edge sync has no implementation behind it — it must stay rejected.
+        assert!(REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"edge"));
+        assert!(REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"edgeSyncBeforeExpiry"));
+    }
+
+    #[test]
+    fn remote_target_create_accepts_disable_proxy() {
+        let mut request = valid_remote_target_request();
+        request["disableProxy"] = serde_json::json!(true);
+
+        let target = serde_json::from_value::<RemoteTargetRequest>(request)
+            .expect("request should deserialize")
+            .into_bucket_target()
+            .expect("disableProxy is a supported per-target read-proxy opt-out");
+
+        assert!(target.disable_proxy);
+    }
+
+    #[test]
+    fn update_body_with_proxy_op_toggles_disable_proxy_without_credentials() {
+        // Mirrors the other partial-update groups: a proxy-only update body may
+        // omit the connection fields entirely.
+        let body = serde_json::json!({
+            "arn": "arn:rustfs:replication:us-east-1:dep:target",
+            "type": "replication",
+            "disableProxy": true
+        });
+        let request: RemoteTargetRequest = serde_json::from_value(body).expect("partial update body should deserialize");
+        let target = request
+            .into_update_bucket_target(&[TargetUpdateOp::Proxy])
+            .expect("proxy-only update must not require credentials");
+        assert!(target.disable_proxy);
+    }
+
     #[test]
     fn remote_target_capability_fields_do_not_overlap() {
         for field in REMOTE_TARGET_UNSUPPORTED_FIELDS {

rustfs/src/admin/handlers/system.rs

@@ -1262,7 +1262,9 @@ mod tests {
         assert_eq!(response.summary.manual_transition_jobs.state, CapabilityState::Supported);
         assert_eq!(response.replication.contract_version, 1);
         assert_eq!(response.replication.bucket_replication.contract_version, 1);
-        assert_eq!(response.replication.remote_targets.contract_version, 1);
+        // v2: disableProxy moved from unsupported to writable (per-target
+        // read-proxy opt-out reached the admin API).
+        assert_eq!(response.replication.remote_targets.contract_version, 2);
         assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported);
         assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported);
         assert_eq!(
@@ -1293,7 +1295,15 @@ mod tests {
                 .remote_targets
                 .fields
                 .iter()
-                .any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Unsupported)
+                .any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Supported)
+        );
+        assert!(
+            response
+                .replication
+                .remote_targets
+                .fields
+                .iter()
+                .any(|field| field.name == "edge" && field.state == super::ReplicationFieldState::Unsupported)
         );
         assert!(
             response
@@ -1364,7 +1374,7 @@ mod tests {
         assert_eq!(value["summary"]["manual_transition_jobs"]["state"], "supported");
         assert_eq!(value["replication"]["contract_version"], 1);
         assert_eq!(value["replication"]["bucket_replication"]["contract_version"], 1);
-        assert_eq!(value["replication"]["remote_targets"]["contract_version"], 1);
+        assert_eq!(value["replication"]["remote_targets"]["contract_version"], 2);
         assert_eq!(value["replication"]["bucket_replication"]["status"]["state"], "supported");
         assert_eq!(value["replication"]["remote_targets"]["status"]["state"], "supported");
         assert_eq!(
@@ -1383,7 +1393,14 @@ mod tests {
                 .as_array()
                 .expect("remote target fields should be an array")
                 .iter()
-                .any(|field| field["name"] == "disableProxy" && field["state"] == "unsupported")
+                .any(|field| field["name"] == "disableProxy" && field["state"] == "supported")
+        );
+        assert!(
+            value["replication"]["remote_targets"]["fields"]
+                .as_array()
+                .expect("remote target fields should be an array")
+                .iter()
+                .any(|field| field["name"] == "edge" && field["state"] == "unsupported")
         );
         assert!(
             value["replication"]["remote_targets"]["fields"]