Merge branch 'main' into fix/split-functional-workflows-and-perf-report

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -122,6 +122,14 @@ fn control_plane_failure(op: &str, bucket: Option<&str>, error_code: Option<i32>
     if error_code == Some(rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode::ControlPlaneErrorNotInitialized as i32) {
         return Error::RemoteNotInitialized;
     }
+    if error_code == Some(rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32)
+    {
+        return Error::InvalidArgument(
+            "control-plane".to_string(),
+            op.to_string(),
+            error_info.unwrap_or_else(|| format!("{op}: peer rejected invalid argument without details")),
+        );
+    }
     match error_info {
         Some(msg) => Error::other(msg),
         None => peer_failure_without_details(op, bucket),
@@ -2335,6 +2343,29 @@ mod tests {
         use rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode;
         assert_eq!(ControlPlaneErrorCode::ControlPlaneErrorUnspecified as i32, 0);
         assert_eq!(ControlPlaneErrorCode::ControlPlaneErrorNotInitialized as i32, 1);
+        assert_eq!(ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32, 2);
+    }
+
+    #[test]
+    fn control_plane_failure_preserves_typed_invalid_argument_reason() {
+        use rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode;
+
+        let reason = "durable unresolved-entry recovery requires pool metadata V2 or V3";
+        let err = control_plane_failure(
+            "start_decommission",
+            None,
+            Some(ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32),
+            Some(reason.to_string()),
+        );
+
+        assert!(
+            matches!(
+                err,
+                Error::InvalidArgument(ref scope, ref operation, ref actual_reason)
+                    if scope == "control-plane" && operation == "start_decommission" && actual_reason == reason
+            ),
+            "forwarded validation failures must remain typed and actionable"
+        );
     }
 
     #[test]

crates/ecstore/src/core/pools.rs

@@ -176,6 +176,34 @@ fn pool_meta_v3_writer_enabled() -> bool {
     )
 }
 
+fn ensure_decommission_ledger_persistence_supported_for(
+    version: u16,
+    v2_writer_enabled: bool,
+    v3_writer_enabled: bool,
+) -> Result<()> {
+    if matches!(version, POOL_META_VERSION | POOL_META_GENERATION_VERSION) || v2_writer_enabled || v3_writer_enabled {
+        return Ok(());
+    }
+
+    Err(Error::InvalidArgument(
+        "decommission".to_string(),
+        "pool-metadata-version".to_string(),
+        format!(
+            "durable unresolved-entry recovery requires pool metadata V2 or V3; enable both {} and {} only after every reader and writer supports V2",
+            rustfs_config::ENV_POOL_META_V2_WRITE,
+            rustfs_config::ENV_POOL_META_V2_FLEET_CONFIRMED,
+        ),
+    ))
+}
+
+fn ensure_decommission_ledger_persistence_supported(pool_meta: &PoolMeta) -> Result<()> {
+    ensure_decommission_ledger_persistence_supported_for(
+        pool_meta.version,
+        pool_meta_v2_writer_enabled(),
+        pool_meta_v3_writer_enabled(),
+    )
+}
+
 #[derive(Clone, Debug)]
 pub struct DecommissionCanceler {
     operation: Arc<DecommissionOperation>,
@@ -2027,6 +2055,7 @@ pub(crate) async fn pause_pool_activation_after_durable_save<S>(pool: &Arc<S>, f
 #[cfg(test)]
 struct PoolActivationStartProbeState {
     kind: PoolActivationStartKind,
+    preflight_side_effect_attempted: std::sync::atomic::AtomicBool,
     attempted: std::sync::atomic::AtomicBool,
     notify: tokio::sync::Notify,
 }
@@ -2045,6 +2074,7 @@ impl PoolActivationStartProbe {
     pub(crate) fn install(kind: PoolActivationStartKind) -> Self {
         let state = Arc::new(PoolActivationStartProbeState {
             kind,
+            preflight_side_effect_attempted: std::sync::atomic::AtomicBool::new(false),
             attempted: std::sync::atomic::AtomicBool::new(false),
             notify: tokio::sync::Notify::new(),
         });
@@ -2061,6 +2091,14 @@ impl PoolActivationStartProbe {
             self.state.notify.notified().await;
         }
     }
+
+    pub(crate) fn preflight_side_effect_was_attempted(&self) -> bool {
+        self.state.preflight_side_effect_attempted.load(Ordering::Acquire)
+    }
+
+    pub(crate) fn activation_was_attempted(&self) -> bool {
+        self.state.attempted.load(Ordering::Acquire)
+    }
 }
 
 #[cfg(test)]
@@ -2090,6 +2128,21 @@ pub(crate) fn observe_pool_activation_start_attempt(kind: PoolActivationStartKin
     }
 }
 
+#[cfg(test)]
+fn observe_pool_activation_preflight_side_effect_attempt(kind: PoolActivationStartKind) {
+    let probes = POOL_ACTIVATION_START_PROBES
+        .get_or_init(|| std::sync::Mutex::new(Vec::new()))
+        .lock()
+        .expect("pool activation start probe should not be poisoned")
+        .iter()
+        .filter(|state| state.kind == kind)
+        .cloned()
+        .collect::<Vec<_>>();
+    for state in probes {
+        state.preflight_side_effect_attempted.store(true, Ordering::Release);
+    }
+}
+
 fn rollback_decommission_pool_meta(pool_meta: &mut PoolMeta, previous_pool_meta: &PoolMeta, indices: &[usize]) {
     publish_pool_meta_updates(pool_meta, previous_pool_meta, indices);
 }
@@ -6189,6 +6242,7 @@ impl ECStore {
     ) -> Result<()> {
         {
             let mut pool_meta = self.pool_meta.write().await;
+            ensure_decommission_ledger_persistence_supported(&pool_meta)?;
             record_decommission_unresolved_entry(&mut pool_meta, idx, generation, entry)?;
         }
         self.save_current_pool_meta(&[idx])
@@ -6339,6 +6393,7 @@ impl ECStore {
         }
 
         ensure_decommission_start_pool_states(&latest_pool_meta, indices)?;
+        ensure_decommission_ledger_persistence_supported(&latest_pool_meta)?;
 
         let previous_pool_meta = latest_pool_meta.clone();
         let first_idx = indices.first().copied();
@@ -7040,10 +7095,14 @@ impl ECStore {
         save_guard.ensure_write_safe("decommission cannot be scheduled while pool metadata requires recovery")?;
         let indices = {
             let pool_meta = self.pool_meta.read().await;
-            resumable_decommission_queue_indices(&pool_meta)
+            let indices = resumable_decommission_queue_indices(&pool_meta)
                 .into_iter()
                 .filter(|idx| indices.contains(idx))
-                .collect::<Vec<_>>()
+                .collect::<Vec<_>>();
+            if !indices.is_empty() {
+                ensure_decommission_ledger_persistence_supported(&pool_meta)?;
+            }
+            indices
         };
         if indices.is_empty() {
             return Ok(Vec::new());
@@ -9299,8 +9358,11 @@ impl ECStore {
         {
             let pool_meta = self.pool_meta.read().await;
             ensure_decommission_start_pool_states(&pool_meta, &indices)?;
+            ensure_decommission_ledger_persistence_supported(&pool_meta)?;
         }
 
+        #[cfg(test)]
+        observe_pool_activation_preflight_side_effect_attempt(PoolActivationStartKind::Decommission);
         let decom_buckets = self.get_buckets_to_decommission().await?;
 
         let mut healed_buckets = HashSet::with_capacity(decom_buckets.len());
@@ -10856,10 +10918,98 @@ mod tests {
         assert!(!is_pool_activation_fleet_proof_error(&Error::ConfigNotFound));
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn decommission_v1_start_preflights_reject_before_metadata_writes() {
+        let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
+        let baseline = load_pool_meta_replicas(store.pools.clone(), true)
+            .await
+            .expect("baseline pool metadata should be readable");
+        assert_eq!(baseline.meta.version, POOL_META_V1_VERSION);
+        *store.pool_meta.write().await = baseline.meta.clone();
+        let start_probe = PoolActivationStartProbe::install(PoolActivationStartKind::Decommission);
+        let err = store
+            .start_decommission(vec![0])
+            .await
+            .expect_err("the initial V1 start preflight must reject before side effects");
+        assert!(matches!(err, Error::InvalidArgument(..)));
+        assert!(
+            !start_probe.preflight_side_effect_was_attempted(),
+            "V1 rejection must precede bucket listing, healing, and metadata-bucket creation"
+        );
+        assert!(
+            !start_probe.activation_was_attempted(),
+            "V1 rejection must not enter the authoritative activation save"
+        );
+        let after_early_rejection = load_pool_meta_replicas(store.pools.clone(), true)
+            .await
+            .expect("early rejection must leave durable pool metadata readable");
+        assert_eq!(after_early_rejection.canonical, baseline.canonical);
+        assert!(
+            store
+                .pool_meta
+                .read()
+                .await
+                .pools
+                .iter()
+                .all(|pool| pool.decommission.is_none())
+        );
+        assert!(store.decommission_cancelers.read().await.iter().all(Option::is_none));
+        store
+            .ensure_pool_meta_side_effects_safe("V1 start preflight")
+            .await
+            .expect("a deterministic start rejection must not latch recovery");
+        drop(start_probe);
+
+        let err = store
+            .save_current_pool_meta_for_decommission_start(
+                &[0],
+                vec![(
+                    0,
+                    PoolSpaceInfo {
+                        free: 50,
+                        total: 100,
+                        used: 50,
+                    },
+                )],
+                Vec::new(),
+            )
+            .await
+            .expect_err("the authoritative V1 start preflight must reject before saving");
+        assert!(matches!(err, Error::InvalidArgument(..)));
+
+        let after_authoritative_rejection = load_pool_meta_replicas(store.pools.clone(), true)
+            .await
+            .expect("authoritative rejection must leave durable pool metadata readable");
+        assert_eq!(after_authoritative_rejection.canonical, baseline.canonical);
+        assert!(
+            after_authoritative_rejection
+                .meta
+                .pools
+                .iter()
+                .all(|pool| pool.decommission.is_none())
+        );
+        assert!(
+            store
+                .pool_meta
+                .read()
+                .await
+                .pools
+                .iter()
+                .all(|pool| pool.decommission.is_none())
+        );
+        assert!(store.decommission_cancelers.read().await.iter().all(Option::is_none));
+        store
+            .ensure_pool_meta_side_effects_safe("authoritative V1 start preflight")
+            .await
+            .expect("an authoritative capability rejection must not latch recovery");
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn decommission_activation_fence_loss_after_durable_save_blocks_publication() {
         let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&store).await;
         let barrier = PoolActivationDurableSaveBarrier::install(&store.pools[0]);
         let start_store = Arc::clone(&store);
         let start_task = tokio::spawn(async move {
@@ -10914,6 +11064,7 @@ mod tests {
     #[serial_test::serial]
     async fn decommission_activation_adopts_canonical_commit_after_replica_failure() {
         let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&store).await;
         let barrier = PoolActivationDurableSaveBarrier::install(&store.pools[0]);
         let start_store = Arc::clone(&store);
         let start_task = tokio::spawn(async move {
@@ -11226,6 +11377,35 @@ mod tests {
         assert!(pool_meta_v3_writer_enabled_for(true, true));
     }
 
+    #[test]
+    fn decommission_ledger_persistence_requires_an_observed_or_confirmed_format() {
+        for (version, v2_enabled, v3_enabled, expected) in [
+            (POOL_META_V1_VERSION, false, false, false),
+            (POOL_META_V1_VERSION, true, false, true),
+            (POOL_META_V1_VERSION, false, true, true),
+            (POOL_META_VERSION, false, false, true),
+            (super::POOL_META_GENERATION_VERSION, false, false, true),
+        ] {
+            let result = super::ensure_decommission_ledger_persistence_supported_for(version, v2_enabled, v3_enabled);
+            assert_eq!(
+                result.is_ok(),
+                expected,
+                "unexpected capability result for pool metadata version {version}"
+            );
+        }
+
+        let half_confirmed_v2 = pool_meta_v2_writer_enabled_for(true, false);
+        let half_confirmed_v3 = pool_meta_v3_writer_enabled_for(false, true);
+        let err = super::ensure_decommission_ledger_persistence_supported_for(
+            POOL_META_V1_VERSION,
+            half_confirmed_v2,
+            half_confirmed_v3,
+        )
+        .expect_err("half-enabled rollout gates must not admit decommission");
+        assert!(matches!(err, Error::InvalidArgument(..)));
+        assert!(err.to_string().contains("durable unresolved-entry recovery"));
+    }
+
     #[test]
     fn pool_meta_stale_write_rejection_metric_is_countable() {
         let recorder = metrics_util::debugging::DebuggingRecorder::new();
@@ -14494,6 +14674,113 @@ mod pools_tests {
         assert!(store.decommission_cancelers.read().await[0].is_none());
     }
 
+    #[tokio::test]
+    async fn test_v1_unresolved_ledger_rejection_keeps_live_state_and_write_gate_safe() {
+        let generation = OffsetDateTime::UNIX_EPOCH;
+        let status = decommission_test_pool_status(
+            0,
+            Some(PoolDecommissionInfo {
+                start_time: Some(generation),
+                ..Default::default()
+            }),
+        );
+        let last_update = status.last_update;
+        let store = decommission_worker_test_store(
+            PoolMeta {
+                version: POOL_META_V1_VERSION,
+                pools: vec![status],
+                ..Default::default()
+            },
+            vec![None],
+        );
+        let entry = DecommissionUnresolvedEntry {
+            bucket: "bucket-a".to_string(),
+            object: "directory/".to_string(),
+            pool_index: 0,
+            set_index: 0,
+            source_generation: generation,
+            candidate_count: 1,
+            disk_error_count: 0,
+            observed_at: generation,
+            reason: "metadata_resolution_failed".to_string(),
+        };
+
+        let err = store
+            .persist_decommission_unresolved_entry(0, generation, entry)
+            .await
+            .expect_err("V1 must reject the ledger before changing live state");
+
+        assert!(matches!(err, Error::InvalidArgument(..)));
+        let pool_meta = store.pool_meta.read().await;
+        let status = &pool_meta.pools[0];
+        assert_eq!(status.last_update, last_update);
+        assert!(
+            status
+                .decommission
+                .as_ref()
+                .expect("active decommission metadata should remain present")
+                .unresolved_entries
+                .is_empty()
+        );
+        drop(pool_meta);
+        store
+            .pool_meta_save_gate
+            .lock()
+            .await
+            .ensure_write_safe("V1 unresolved-entry preflight")
+            .expect("a deterministic capability rejection must not latch recovery");
+    }
+
+    #[tokio::test]
+    async fn test_v1_runtime_recovery_rejects_worker_but_keeps_cancel_persistable() {
+        let generation = OffsetDateTime::UNIX_EPOCH;
+        let store = decommission_worker_test_store(
+            PoolMeta {
+                version: POOL_META_V1_VERSION,
+                pools: vec![decommission_test_pool_status(
+                    0,
+                    Some(PoolDecommissionInfo {
+                        start_time: Some(generation),
+                        ..Default::default()
+                    }),
+                )],
+                ..Default::default()
+            },
+            vec![None],
+        );
+
+        let err = store
+            .reserve_decommission_routines(&CancellationToken::new(), &[0])
+            .await
+            .err()
+            .expect("V1 recovery must not install a worker that cannot persist an unresolved ledger");
+        assert!(matches!(err, Error::InvalidArgument(..)));
+        assert!(store.decommission_cancelers.read().await[0].is_none());
+
+        let save_called = Arc::new(AtomicBool::new(false));
+        store
+            .decommission_cancel_with_owner_and_save(0, None, {
+                let save_called = save_called.clone();
+                move |snapshot, _| async move {
+                    snapshot.encode_config_data_for_v2_gate(false)?;
+                    save_called.store(true, Ordering::SeqCst);
+                    Ok(())
+                }
+            })
+            .await
+            .expect("a rejected V1 recovery must remain cancelable without restart");
+
+        assert!(save_called.load(Ordering::SeqCst));
+        let pool_meta = store.pool_meta.read().await;
+        let info = pool_meta.pools[0]
+            .decommission
+            .as_ref()
+            .expect("cancel metadata should remain present");
+        assert!(info.canceled);
+        assert!(!info.failed);
+        assert!(!info.complete);
+    }
+
     #[tokio::test]
     async fn test_decommission_transition_waits_without_registered_canceler() {
         let store = decommission_worker_test_store(PoolMeta::default(), vec![None]);
@@ -17487,6 +17774,7 @@ mod pools_tests {
     #[tokio::test]
     async fn test_runtime_recovery_reserves_the_startup_resumable_queue() {
         let meta = PoolMeta {
+            version: super::POOL_META_VERSION,
             pools: vec![
                 decommission_test_pool_status(
                     0,
@@ -17544,6 +17832,7 @@ mod pools_tests {
     #[tokio::test]
     async fn test_runtime_recovery_does_not_reserve_behind_active_predecessor() {
         let meta = PoolMeta {
+            version: super::POOL_META_VERSION,
             pools: vec![
                 decommission_test_pool_status(
                     0,

crates/ecstore/src/disk/local.rs

@@ -3148,9 +3148,10 @@ impl LocalIoBackend for StdBackend {
                 direct_read_copy_fault_delta: MmapPageFaultDelta,
                 blocking_task_duration: StdDuration,
                 used_direct_io: bool,
-                /// The descriptor opened by THIS call (None on a cache hit), handed
-                /// back so the async caller can index it in the fd cache.
-                opened_fd: Option<Arc<std::fs::File>>,
+                /// The descriptor and size snapshot opened by THIS call (None on a
+                /// cache hit), handed back so the async caller can index it in the
+                /// fd cache.
+                opened_fd: Option<Arc<FdCacheEntry>>,
             }
 
             enum MmapCopyReadError {
@@ -3197,12 +3198,12 @@ impl LocalIoBackend for StdBackend {
                 (cache, key, gen_at_open)
             });
             #[cfg(target_os = "linux")]
-            let cached_fd: Option<Arc<std::fs::File>> = match &fd_lookup {
+            let cached_fd: Option<Arc<FdCacheEntry>> = match &fd_lookup {
                 Some((cache, key, _)) => cache.get(key).await,
                 None => None,
             };
             #[cfg(not(target_os = "linux"))]
-            let cached_fd: Option<Arc<std::fs::File>> = None;
+            let cached_fd: Option<Arc<FdCacheEntry>> = None;
 
             let blocking_wait_start = metrics_enabled.then(std::time::Instant::now);
             let read_result = tokio::task::spawn_blocking(move || {
@@ -3224,8 +3225,15 @@ impl LocalIoBackend for StdBackend {
                 // the read below is positioned (mmap offset argument / `read_exact_at`)
                 // and never depends on the descriptor's current offset. `cached_fd` being
                 // None also marks this call as a miss for the cache-insert side-channel.
-                let (file, access_check_duration) = if let Some(cached) = cached_fd.as_ref() {
-                    (cached.as_ref().try_clone().map_err(DiskError::from)?, StdDuration::ZERO)
+                // The cached length is the metadata snapshot captured at open time;
+                // all in-place/replacement writers invalidate this entry before
+                // publishing a mutation, so cache hits avoid a redundant fstat.
+                let (file, cached_len, access_check_duration) = if let Some(cached) = cached_fd.as_ref() {
+                    (
+                        cached.file.as_ref().try_clone().map_err(DiskError::from)?,
+                        Some(cached.len),
+                        StdDuration::ZERO,
+                    )
                 } else {
                     // Measure the volume access probe only — the part-path resolution
                     // above is accounted in `path_resolve_duration` (rustfs/backlog#1801).
@@ -3236,20 +3244,27 @@ impl LocalIoBackend for StdBackend {
                             .map_err(|e| DiskError::from(to_access_error(e, DiskError::VolumeAccessDenied)))?;
                     }
                     let access_check_duration = access_check_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
-                    (std::fs::File::open(&file_path).map_err(DiskError::from)?, access_check_duration)
+                    (std::fs::File::open(&file_path).map_err(DiskError::from)?, None, access_check_duration)
                 };
                 let file_open_duration = file_open_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
 
-                let metadata_lookup_start = metrics_enabled.then(StdInstant::now);
-                // On a cache hit this fstats the cached descriptor — the inode it was
-                // opened against, which invalidation keeps current for live entries. EC
-                // shards are fixed-length, so a still-cached pre-heal length is benign.
-                let meta = file.metadata().map_err(DiskError::from)?;
-                let metadata_lookup_duration = metadata_lookup_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
+                let (metadata_len, metadata_lookup_duration) = if let Some(len) = cached_len {
+                    // Reuse the open-time metadata snapshot on a cache hit. The
+                    // generation fence and mutation invalidation keep this value
+                    // tied to the inode held by `file`.
+                    (len, StdDuration::ZERO)
+                } else {
+                    let metadata_lookup_start = metrics_enabled.then(StdInstant::now);
+                    let meta = file.metadata().map_err(DiskError::from)?;
+                    let duration = metadata_lookup_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
+                    (meta.len(), duration)
+                };
 
                 let metadata_validate_start = metrics_enabled.then(StdInstant::now);
-                if meta.len() < end_offset_u64 {
-                    return Err(MmapCopyReadError::OutOfBounds { actual_size: meta.len() });
+                if metadata_len < end_offset_u64 {
+                    return Err(MmapCopyReadError::OutOfBounds {
+                        actual_size: metadata_len,
+                    });
                 }
                 let metadata_validate_duration =
                     metadata_validate_start.map_or(StdDuration::ZERO, |started_at| started_at.elapsed());
@@ -3395,9 +3410,14 @@ impl LocalIoBackend for StdBackend {
                 // Arc; `cached_fd.is_none()` is true exactly when this call did the open.
                 // Non-Linux has no fd cache, so skip the Arc allocation there.
                 #[cfg(target_os = "linux")]
-                let opened_fd: Option<Arc<std::fs::File>> = cached_fd.is_none().then(|| Arc::new(file));
+                let opened_fd: Option<Arc<FdCacheEntry>> = cached_fd.is_none().then(|| {
+                    Arc::new(FdCacheEntry {
+                        file: Arc::new(file),
+                        len: metadata_len,
+                    })
+                });
                 #[cfg(not(target_os = "linux"))]
-                let opened_fd: Option<Arc<std::fs::File>> = None;
+                let opened_fd: Option<Arc<FdCacheEntry>> = None;
 
                 Ok::<MmapCopyReadResult, MmapCopyReadError>(MmapCopyReadResult {
                     bytes,
@@ -3520,7 +3540,7 @@ impl LocalIoBackend for StdBackend {
                     }
                 }
             }
-            // Index the freshly opened descriptor for future cache hits
+            // Index the freshly opened descriptor and metadata snapshot for future cache hits
             // (rustfs/backlog#1801). `insert_if_fresh` refuses to cache if an
             // invalidation (heal/delete/rename) bumped the generation between the
             // open snapshot and now, so a stale pre-mutation inode is never served
@@ -3872,6 +3892,18 @@ struct FdKey {
     direct: bool,
 }
 
+/// Descriptor and immutable size snapshot retained for one cached shard inode.
+///
+/// The generation fence and explicit mutation invalidation keep the snapshot
+/// tied to the inode held by `file`, allowing cache hits to avoid a repeated
+/// metadata syscall without weakening replacement/heal semantics.
+struct FdCacheEntry {
+    /// An independently cloneable descriptor for the immutable shard inode.
+    file: Arc<std::fs::File>,
+    /// File length captured together with the descriptor.
+    len: u64,
+}
+
 /// Per-disk cache of open descriptors for io_uring reads (backlog#1145).
 ///
 /// Why this exists: `pread_uring` opened the file on the blocking pool for every
@@ -3901,7 +3933,7 @@ struct FdKey {
 /// the descriptor once no in-flight read still holds it.
 #[cfg(target_os = "linux")]
 struct FdCache {
-    cache: moka::future::Cache<FdKey, Arc<std::fs::File>>,
+    cache: moka::future::Cache<FdKey, Arc<FdCacheEntry>>,
     /// Bumped by every invalidation. A miss-path open snapshots this before it
     /// opens and refuses to insert if it moved, so an fd opened before a
     /// heal/delete commit can never be resurrected into the cache after the
@@ -3931,7 +3963,7 @@ impl FdCache {
         }
     }
 
-    async fn get(&self, key: &FdKey) -> Option<Arc<std::fs::File>> {
+    async fn get(&self, key: &FdKey) -> Option<Arc<FdCacheEntry>> {
         self.cache.get(key).await
     }
 
@@ -3946,11 +3978,11 @@ impl FdCache {
     /// open bumped the generation, so a stale pre-heal/pre-delete inode is never
     /// cached. The post-insert re-check closes the tiny window where an
     /// invalidate races the insert itself, by removing the entry we just added.
-    async fn insert_if_fresh(&self, key: FdKey, file: Arc<std::fs::File>, gen_at_open: u64) {
+    async fn insert_if_fresh(&self, key: FdKey, entry: Arc<FdCacheEntry>, gen_at_open: u64) {
         if self.generation.load(Ordering::Acquire) != gen_at_open {
             return;
         }
-        self.cache.insert(key.clone(), file).await;
+        self.cache.insert(key.clone(), entry).await;
         if self.generation.load(Ordering::Acquire) != gen_at_open {
             self.cache.invalidate(&key).await;
         }
@@ -3986,7 +4018,7 @@ impl FdCache {
         self.generation.fetch_add(1, Ordering::AcqRel);
         let volume = volume.to_owned();
         let prefix = prefix.trim_end_matches('/').to_owned();
-        let matches = move |k: &FdKey, _: &Arc<std::fs::File>| {
+        let matches = move |k: &FdKey, _: &Arc<FdCacheEntry>| {
             k.volume == volume && (k.path == prefix || k.path.strip_prefix(&prefix).is_some_and(|r| r.starts_with('/')))
         };
         if self.cache.invalidate_entries_if(matches).is_err() {
@@ -4002,7 +4034,7 @@ impl FdCache {
     fn invalidate_volume(&self, volume: &str) {
         self.generation.fetch_add(1, Ordering::AcqRel);
         let volume = volume.to_owned();
-        let matches = move |k: &FdKey, _: &Arc<std::fs::File>| k.volume == volume;
+        let matches = move |k: &FdKey, _: &Arc<FdCacheEntry>| k.volume == volume;
         if self.cache.invalidate_entries_if(matches).is_err() {
             self.cache.invalidate_all();
         }
@@ -4020,7 +4052,8 @@ impl FdCache {
     /// tests that drive the cache directly.
     #[cfg(test)]
     async fn insert(&self, key: FdKey, file: Arc<std::fs::File>) {
-        self.cache.insert(key, file).await;
+        let len = file.metadata().map(|metadata| metadata.len()).unwrap_or_default();
+        self.cache.insert(key, Arc::new(FdCacheEntry { file, len })).await;
     }
 
     #[cfg(test)]
@@ -4399,7 +4432,12 @@ impl UringBackend {
         };
 
         let file = match cached {
-            Some(file) => file,
+            Some(entry) => {
+                if entry.len < u64::try_from(end_offset).map_err(|_| DiskError::FileCorrupt)? {
+                    return Err(DiskError::FileCorrupt);
+                }
+                Arc::clone(&entry.file)
+            }
             None => {
                 // Snapshot the cache generation BEFORE opening (rustfs/backlog#1176):
                 // if a heal/delete invalidation runs while this open is in flight,
@@ -4409,7 +4447,7 @@ impl UringBackend {
                 let root = self.root.clone();
                 let volume_owned = volume.to_owned();
                 let path_owned = path.to_owned();
-                let file = tokio::task::spawn_blocking(move || -> Result<std::fs::File> {
+                let (file, len) = tokio::task::spawn_blocking(move || -> Result<(std::fs::File, u64)> {
                     let file_path = resolve_uring_object_path(&root, &volume_owned, &path_owned)?;
                     let file = std::fs::File::open(&file_path).map_err(DiskError::from)?;
                     let meta = file.metadata().map_err(DiskError::from)?;
@@ -4417,30 +4455,22 @@ impl UringBackend {
                     if meta.len() < end_offset_u64 {
                         return Err(DiskError::FileCorrupt);
                     }
-                    Ok(file)
+                    Ok((file, meta.len()))
                 })
                 .await
                 .map_err(|e| DiskError::other(format!("uring pread join error: {e}")))??;
-                let file = Arc::new(file);
+                let file = Arc::new(FdCacheEntry {
+                    file: Arc::new(file),
+                    len,
+                });
                 if let (Some((cache, key)), Some(gen_at_open)) = (cache_entry, gen_at_open) {
                     cache.insert_if_fresh(key, Arc::clone(&file), gen_at_open).await;
                 }
-                file
+                file.file.clone()
             }
         };
 
         if length == 0 {
-            // Parity with StdBackend and the miss path (rustfs/backlog#1173): a
-            // zero-length read still rejects an offset past EOF. The miss path
-            // validated `meta.len() < end_offset` (end_offset == offset here), but
-            // a cache hit skipped it — so fstat the descriptor and match. This is
-            // a rare path (callers do not issue zero-length reads), so the one
-            // extra fstat is negligible.
-            match file.metadata() {
-                Ok(meta) if offset_u64 > meta.len() => return Err(DiskError::FileCorrupt),
-                Ok(_) => {}
-                Err(e) => return Err(DiskError::from(e)),
-            }
             return Ok(Bytes::new());
         }
 
@@ -21407,11 +21437,10 @@ mod test {
 
     /// Zero-length read bounds parity on the cache-HIT path (backlog#1173/#1180).
     /// A `length == 0` read past EOF must be rejected identically whether the
-    /// descriptor is freshly opened (miss path) or served from the cache: the
-    /// cache-hit branch fstats the descriptor to reproduce the miss path's
-    /// `offset > len` check instead of returning empty unconditionally. Seeds
-    /// the cache with a normal read so the zero-length reads are hits, then pins
-    /// that UringBackend and StdBackend agree on every case.
+    /// descriptor is freshly opened (miss path) or served from the cache. Seeds
+    /// the cache with a normal read so the zero-length reads reuse the same
+    /// open-time size snapshot, then pins that UringBackend and StdBackend agree
+    /// on every case.
     #[cfg(target_os = "linux")]
     #[tokio::test(flavor = "multi_thread")]
     async fn uring_zero_length_read_bounds_match_std_on_cache_hit() {

crates/ecstore/src/services/rebalance/control.rs

@@ -1667,6 +1667,8 @@ mod tests {
     async fn assert_real_activation_start_race(paused_kind: PoolActivationStartKind) {
         let (_temp_dirs, rebalance_store, decommission_store) =
             crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(None).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&rebalance_store).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&decommission_store).await;
         let disk_stats = vec![
             DiskStat {
                 total_space: 100,

crates/ecstore/src/services/rebalance/mod.rs

@@ -111,6 +111,17 @@ pub(crate) async fn test_two_pool_stores_with_isolated_node_contexts(
     test_two_pool_stores_with_contexts(rebalance_meta, true).await
 }
 
+#[cfg(test)]
+pub(crate) async fn promote_test_pool_meta_to_v2(store: &std::sync::Arc<crate::store::ECStore>) {
+    let mut pool_meta = store.pool_meta.read().await.clone();
+    pool_meta.version = crate::core::pools::POOL_META_VERSION;
+    pool_meta
+        .save(store.pools.clone())
+        .await
+        .expect("test pool metadata should be promoted to V2");
+    *store.pool_meta.write().await = pool_meta;
+}
+
 #[cfg(test)]
 async fn test_two_pool_stores_with_contexts(
     rebalance_meta: Option<RebalanceMeta>,

crates/protos/src/generated/proto_gen/node_service.rs

@@ -1553,6 +1553,9 @@ pub enum ControlPlaneErrorCode {
     /// The peer answered but its storage/IAM layer is not initialized yet
     /// (legacy string form: "errServerNotInitialized").
     ControlPlaneErrorNotInitialized = 1,
+    /// The peer rejected a control-plane request before changing durable state.
+    /// error_info carries the actionable validation reason.
+    ControlPlaneErrorInvalidArgument = 2,
 }
 impl ControlPlaneErrorCode {
     /// String value of the enum field names used in the ProtoBuf definition.
@@ -1563,6 +1566,7 @@ impl ControlPlaneErrorCode {
         match self {
             Self::ControlPlaneErrorUnspecified => "CONTROL_PLANE_ERROR_UNSPECIFIED",
             Self::ControlPlaneErrorNotInitialized => "CONTROL_PLANE_ERROR_NOT_INITIALIZED",
+            Self::ControlPlaneErrorInvalidArgument => "CONTROL_PLANE_ERROR_INVALID_ARGUMENT",
         }
     }
     /// Creates an enum from field names used in the ProtoBuf definition.
@@ -1570,6 +1574,7 @@ impl ControlPlaneErrorCode {
         match value {
             "CONTROL_PLANE_ERROR_UNSPECIFIED" => Some(Self::ControlPlaneErrorUnspecified),
             "CONTROL_PLANE_ERROR_NOT_INITIALIZED" => Some(Self::ControlPlaneErrorNotInitialized),
+            "CONTROL_PLANE_ERROR_INVALID_ARGUMENT" => Some(Self::ControlPlaneErrorInvalidArgument),
             _ => None,
         }
     }

crates/protos/src/node.proto

@@ -30,6 +30,9 @@ enum ControlPlaneErrorCode {
   // The peer answered but its storage/IAM layer is not initialized yet
   // (legacy string form: "errServerNotInitialized").
   CONTROL_PLANE_ERROR_NOT_INITIALIZED = 1;
+  // The peer rejected a control-plane request before changing durable state.
+  // error_info carries the actionable validation reason.
+  CONTROL_PLANE_ERROR_INVALID_ARGUMENT = 2;
 }
 
 message PingRequest {

rustfs/src/storage/rpc/node_service.rs

@@ -123,6 +123,21 @@ fn verify_node_mutation_body<T: CanonicalMutationBody>(request: &Request<T>, ope
         .map_err(|err| Status::permission_denied(format!("{operation} authentication failed: {err}")))
 }
 
+fn start_decommission_failure_response(err: Error) -> StartDecommissionResponse {
+    match err {
+        Error::InvalidArgument(_, _, reason) => StartDecommissionResponse {
+            success: false,
+            error_info: Some(reason),
+            error_code: Some(ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32),
+        },
+        err => StartDecommissionResponse {
+            success: false,
+            error_info: Some(err.to_string()),
+            error_code: None,
+        },
+    }
+}
+
 fn supports_dynamic_config_rpc(sub_system: &str) -> bool {
     NOTIFY_SUB_SYSTEMS.contains(&sub_system)
         || matches!(
@@ -2334,11 +2349,7 @@ impl Node for NodeService {
                 success: true,
                 error_info: None,
             })),
-            Err(err) => Ok(Response::new(StartDecommissionResponse {
-                error_code: None,
-                success: false,
-                error_info: Some(err.to_string()),
-            })),
+            Err(err) => Ok(Response::new(start_decommission_failure_response(err))),
         }
     }
 
@@ -2456,7 +2467,7 @@ mod tests {
         initialize_heal_topology_fingerprint, initialize_heal_topology_fingerprint_with_probe, legacy_scanner_activity_response,
         make_heal_control_server, make_heal_control_server_with_cache, make_server, make_server_for_context,
         make_tier_mutation_control_server_for_context, previous_scanner_activity_response, remove_heal_control_replay,
-        scanner_activity_response_v7, stop_rebalance_response,
+        scanner_activity_response_v7, start_decommission_failure_response, stop_rebalance_response,
     };
     use crate::storage::rpc::node_service::heal::heal_topology_fingerprint;
     use crate::storage::storage_api::rpc_consumer::node_service::{DiskError, HealBucketInfo};
@@ -2479,14 +2490,14 @@ mod tests {
     use rustfs_protos::models::PingBodyBuilder;
     use rustfs_protos::proto_gen::node_service::{
         BackgroundHealStatusRequest, BatchGenerallyLockRequest, CancelDecommissionRequest, CheckPartsRequest,
-        ClearDecommissionRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest, DeletePolicyRequest,
-        DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest, DeleteVersionsRequest,
-        DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest, GetAllBucketStatsRequest,
-        GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest, GetMetacacheListingRequest,
-        GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest,
-        GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest, HealBucketRequest, HealControlRequest,
-        ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest, LoadGroupRequest,
-        LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest,
+        ClearDecommissionRequest, ControlPlaneErrorCode, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest,
+        DeletePolicyRequest, DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest,
+        DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest,
+        GetAllBucketStatsRequest, GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest,
+        GetMetacacheListingRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest,
+        GetProcInfoRequest, GetSeLinuxInfoRequest, GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest,
+        HealBucketRequest, HealControlRequest, ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest,
+        LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest,
         LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest,
         MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest,
         ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest,
@@ -2576,6 +2587,20 @@ mod tests {
         }
     }
 
+    #[test]
+    fn start_decommission_failure_response_preserves_invalid_argument_reason() {
+        let reason = "durable unresolved-entry recovery requires pool metadata V2 or V3";
+        let response = start_decommission_failure_response(Error::InvalidArgument(
+            "decommission".to_string(),
+            "pool-metadata-version".to_string(),
+            reason.to_string(),
+        ));
+
+        assert!(!response.success);
+        assert_eq!(response.error_info.as_deref(), Some(reason));
+        assert_eq!(response.error_code, Some(ControlPlaneErrorCode::ControlPlaneErrorInvalidArgument as i32));
+    }
+
     struct HealControlMockStorage;
 
     #[async_trait::async_trait]