ci(keystone): inherit workspace lint policy

* fix(ilm): preserve restore source version mode

* test(ilm): cover suspended null-version restore

* fix(ilm): reconcile worker results from journal

crates/data-usage/src/data_usage.rs

@@ -506,8 +506,22 @@ pub struct ReplicationStats {
 }
 
 impl ReplicationStats {
+    pub fn is_empty(&self) -> bool {
+        self.pending_size == 0
+            && self.replicated_size == 0
+            && self.failed_size == 0
+            && self.failed_count == 0
+            && self.pending_count == 0
+            && self.missed_threshold_size == 0
+            && self.after_threshold_size == 0
+            && self.missed_threshold_count == 0
+            && self.after_threshold_count == 0
+            && self.replicated_count == 0
+    }
+
+    #[deprecated(note = "use is_empty instead")]
     pub fn empty(&self) -> bool {
-        self.replicated_size == 0 && self.failed_size == 0 && self.failed_count == 0
+        self.is_empty()
     }
 }
 
@@ -520,16 +534,13 @@ pub struct ReplicationAllStats {
 }
 
 impl ReplicationAllStats {
+    pub fn is_empty(&self) -> bool {
+        self.replica_size == 0 && self.replica_count == 0 && self.targets.values().all(ReplicationStats::is_empty)
+    }
+
+    #[deprecated(note = "use is_empty instead")]
     pub fn empty(&self) -> bool {
-        if self.replica_size != 0 && self.replica_count != 0 {
-            return false;
-        }
-        for v in self.targets.values() {
-            if !v.empty() {
-                return false;
-            }
-        }
-        true
+        self.is_empty()
     }
 }
 
@@ -783,7 +794,7 @@ impl DataUsageCache {
                     return Some(root);
                 }
                 let mut flat = self.flatten(&root);
-                if flat.replication_stats.as_ref().is_some_and(|stats| stats.empty()) {
+                if flat.replication_stats.as_ref().is_some_and(ReplicationAllStats::is_empty) {
                     flat.replication_stats = None;
                 }
                 Some(flat)
@@ -1582,6 +1593,128 @@ mod tests {
         assert_eq!(map["BETWEEN_1024B_AND_1_MB"], u64::MAX);
     }
 
+    #[test]
+    #[allow(deprecated)]
+    fn replication_stats_empty_checks_every_field() {
+        type SetField = fn(&mut ReplicationStats);
+
+        let cases: [(&str, SetField); 10] = [
+            ("pending_size", |stats| stats.pending_size = 1),
+            ("replicated_size", |stats| stats.replicated_size = 1),
+            ("failed_size", |stats| stats.failed_size = 1),
+            ("failed_count", |stats| stats.failed_count = 1),
+            ("pending_count", |stats| stats.pending_count = 1),
+            ("missed_threshold_size", |stats| stats.missed_threshold_size = 1),
+            ("after_threshold_size", |stats| stats.after_threshold_size = 1),
+            ("missed_threshold_count", |stats| stats.missed_threshold_count = 1),
+            ("after_threshold_count", |stats| stats.after_threshold_count = 1),
+            ("replicated_count", |stats| stats.replicated_count = 1),
+        ];
+
+        assert!(ReplicationStats::default().empty());
+        for (field, set_nonzero) in cases {
+            let mut stats = ReplicationStats::default();
+            set_nonzero(&mut stats);
+            assert!(!stats.empty(), "{field} must make replication stats non-empty");
+        }
+    }
+
+    #[test]
+    #[allow(deprecated)]
+    fn replication_all_stats_empty_checks_aggregate_fields_independently() {
+        let cases = [
+            (
+                "replica_size",
+                ReplicationAllStats {
+                    replica_size: 1,
+                    ..Default::default()
+                },
+            ),
+            (
+                "replica_count",
+                ReplicationAllStats {
+                    replica_count: 1,
+                    ..Default::default()
+                },
+            ),
+        ];
+
+        assert!(ReplicationAllStats::default().empty());
+        for (field, stats) in cases {
+            assert!(!stats.empty(), "{field} must make aggregate replication stats non-empty");
+        }
+
+        let empty_targets = ReplicationAllStats {
+            targets: HashMap::from([("arn:test:empty".to_string(), ReplicationStats::default())]),
+            ..Default::default()
+        };
+        assert!(empty_targets.empty(), "all-empty targets must keep aggregate stats empty");
+
+        let stats = ReplicationAllStats {
+            targets: HashMap::from([
+                ("arn:test:empty".to_string(), ReplicationStats::default()),
+                (
+                    "arn:test:non-empty".to_string(),
+                    ReplicationStats {
+                        pending_count: 1,
+                        ..Default::default()
+                    },
+                ),
+            ]),
+            ..Default::default()
+        };
+        assert!(!stats.empty(), "a non-empty target must make aggregate replication stats non-empty");
+    }
+
+    #[test]
+    fn size_recursive_prunes_empty_and_preserves_pending_replication_stats() {
+        let root = hash_path("bucket");
+        let child = hash_path("bucket/child");
+        let mut cache = DataUsageCache::default();
+        cache.replace_hashed(&root, &None, &DataUsageEntry::default());
+        cache.replace_hashed(
+            &child,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                replication_stats: Some(ReplicationAllStats::default()),
+                ..Default::default()
+            },
+        );
+
+        assert!(
+            cache
+                .size_recursive("bucket")
+                .expect("bucket usage should flatten")
+                .replication_stats
+                .is_none()
+        );
+
+        cache.replace_hashed(
+            &child,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                replication_stats: Some(ReplicationAllStats {
+                    targets: HashMap::from([(
+                        "arn:test:pending".to_string(),
+                        ReplicationStats {
+                            pending_count: 1,
+                            ..Default::default()
+                        },
+                    )]),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            },
+        );
+
+        let flattened = cache.size_recursive("bucket").expect("bucket usage should flatten");
+        let replication = flattened
+            .replication_stats
+            .expect("pending-only replication stats must survive pruning");
+
+        assert_eq!(replication.targets["arn:test:pending"].pending_count, 1);
+    }
+
     #[test]
     fn test_data_usage_cache_merge_adds_missing_child() {
         let mut base = DataUsageCache::default();

crates/ecstore/src/api/mod.rs

@@ -127,8 +127,8 @@ pub mod bucket {
             get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config,
             get_object_lock_config, get_public_access_block_config, get_quota_config, get_replication_config,
             get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config, get_website_config,
-            init_bucket_metadata_sys, list_bucket_targets, remove_bucket_metadata, set_bucket_metadata, update,
-            update_bucket_targets_under_transaction_lock, update_config_with,
+            init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata,
+            update, update_bucket_targets_under_transaction_lock, update_config_with,
         };
     }
 

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -8974,7 +8974,7 @@ mod tests {
         .await
         .expect("first worker result should persist");
         assert_eq!(first.state, ManualTransitionJobState::Running);
-        assert_eq!(first.report.transition_completed, 1);
+        assert_eq!(first.report.transition_completed, 0);
         assert_eq!(first.report.transition_failed, 0);
 
         let duplicate = record_manual_transition_worker_result(
@@ -8987,11 +8987,11 @@ mod tests {
         .await
         .expect("duplicate worker result should be idempotent");
         assert_eq!(duplicate.state, ManualTransitionJobState::Running);
-        assert_eq!(duplicate.report.transition_completed, 1);
+        assert_eq!(duplicate.report.transition_completed, 0);
         assert_eq!(duplicate.report.transition_failed, 0);
 
         let second_key = manual_transition_worker_result_task_key(&bucket, "logs/b", None);
-        let final_record = record_manual_transition_worker_result(
+        let pending_record = record_manual_transition_worker_result(
             ecstore.clone(),
             job_id,
             &second_key,
@@ -9000,7 +9000,14 @@ mod tests {
         )
         .await
         .expect("second distinct worker result should persist");
+        assert_eq!(pending_record.state, ManualTransitionJobState::Running);
+        assert_eq!(pending_record.report.transition_completed, 0);
+        assert_eq!(pending_record.report.transition_failed, 0);
 
+        let final_record =
+            reconcile_manual_transition_worker_results(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
+                .await
+                .expect("worker result journal should reconcile");
         assert_eq!(final_record.state, ManualTransitionJobState::Partial);
         assert_eq!(final_record.report.transition_completed, 1);
         assert_eq!(final_record.report.transition_failed, 1);
@@ -9035,7 +9042,7 @@ mod tests {
             .expect("worker result job record should save");
 
         let task_key = manual_transition_worker_result_task_key(&bucket, "logs/fail", None);
-        let final_record = record_manual_transition_worker_result_with_reason(
+        let pending_record = record_manual_transition_worker_result_with_reason(
             ecstore.clone(),
             job_id,
             &task_key,
@@ -9045,7 +9052,12 @@ mod tests {
         )
         .await
         .expect("worker result with failure reason should persist");
+        assert!(pending_record.report.tier_failure_by_reason.is_empty());
+        assert_eq!(pending_record.report.transition_failed, 0);
 
+        let final_record = reconcile_manual_transition_worker_results(ecstore, job_id, ManualTransitionQueueSnapshot::default())
+            .await
+            .expect("worker failure reason should reconcile");
         assert_eq!(
             final_record
                 .report

crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs

@@ -1646,40 +1646,14 @@ pub async fn record_manual_transition_worker_result_with_reason(
     job_id: Uuid,
     task_key: &str,
     result: ManualTransitionWorkerResult,
-    queue_snapshot: ManualTransitionQueueSnapshot,
+    _queue_snapshot: ManualTransitionQueueSnapshot,
     failure_reason: Option<ManualTransitionWorkerFailureReason>,
 ) -> EcstoreResult<ManualTransitionJobRecord> {
     let result_record = ManualTransitionWorkerResultRecord::new_with_reason(job_id, task_key, result, failure_reason);
     if !save_manual_transition_worker_result_if_absent(api.clone(), &result_record).await? {
         return load_manual_transition_job_record(api, job_id).await;
     }
-
-    for _ in 0..4 {
-        let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
-        if record.is_terminal() {
-            return Ok(record);
-        }
-        record.record_worker_result_with_reason(result, queue_snapshot, failure_reason);
-        match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await {
-            Ok(()) => {
-                if record.is_terminal() {
-                    delete_manual_transition_scope_admission_if_current(
-                        api.clone(),
-                        &record.scope_key,
-                        record.job_id,
-                        record.lease_id,
-                    )
-                    .await?;
-                } else {
-                    renew_manual_transition_scope_admission_from_job(api, &record).await?;
-                }
-                return Ok(record);
-            }
-            Err(Error::PreconditionFailed) => continue,
-            Err(err) => return Err(err),
-        }
-    }
-    Err(Error::PreconditionFailed)
+    load_manual_transition_job_record(api, job_id).await
 }
 
 pub async fn renew_manual_transition_job_lease(

crates/ecstore/src/bucket/metadata_sys.rs

@@ -91,6 +91,20 @@ pub async fn set_bucket_metadata(bucket: String, bm: BucketMetadata) -> Result<(
     Ok(())
 }
 
+/// Peer LoadBucketMetadata entry point; see
+/// [`BucketMetadataSys::reload_from_store`] for the caching contract.
+///
+/// The outer write guard spans the disk load, mirroring [`update`]: every
+/// other cache installer holds this lock (read or write), so the snapshot
+/// read here can never land after — and roll back — a newer concurrent
+/// install, and the install-plus-registry-sync sequence stays atomic
+/// against concurrent removes and reloads.
+pub async fn reload_bucket_metadata(bucket: &str) -> Result<()> {
+    let sys = get_bucket_metadata_sys()?;
+    let lock = sys.write().await;
+    lock.reload_from_store(bucket).await
+}
+
 /// Drop a bucket's cached metadata from the in-memory map.
 ///
 /// This is the counterpart to [`set_bucket_metadata`] and is invoked when a
@@ -659,6 +673,43 @@ impl BucketMetadataSys {
         }
     }
 
+    /// Reload `bucket`'s metadata from this system's own store and cache it,
+    /// refusing to treat a load miss as authoritative (the peer
+    /// LoadBucketMetadata notification path, [`reload_bucket_metadata`]).
+    ///
+    /// Only metadata actually read from persisted storage reaches the cache.
+    /// On a miss the fabricated default is discarded and an error is
+    /// returned: installing it would let a transient ConfigNotFound during
+    /// the notification overwrite a lock-enabled bucket's cached metadata
+    /// with an authoritative "no Object Lock" default, disabling the
+    /// batch-delete retention gate (`object_lock_delete_check_required`) on
+    /// this node until the next refresh. A miss is also not treated as
+    /// deletion: bucket deletion propagates through the dedicated
+    /// DeleteBucketMetadata notification ([`remove_bucket_metadata`]), which
+    /// is best-effort — a reload racing it can still re-install a just
+    /// deleted bucket's entry (pre-existing, bounded by the next delete or
+    /// restart) — but a reload miss removing entries would turn every
+    /// transient quorum dip into dropped metadata and spurious
+    /// target/durability teardown.
+    ///
+    /// The peer-visible error text is deliberately fixed: the notifying peer
+    /// matches error strings against network-failure needles
+    /// (`is_network_like_error`), so interpolating a caller-controlled
+    /// bucket name here could mark a healthy peer offline.
+    ///
+    /// Lock order: the caller holds the outer metadata-sys guard, and the
+    /// load acquires the namespace lock on the bucket's metadata config
+    /// object — the same `outer guard → meta-config namespace lock` order
+    /// `update`'s load takes; no path acquires these in reverse.
+    pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
+        let (bm, persisted) = load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true).await?;
+        if !persisted {
+            return Err(Error::other("no persisted bucket metadata readable; peer cache left unchanged"));
+        }
+        self.set(bucket.to_string(), Arc::new(bm)).await;
+        Ok(())
+    }
+
     /// Remove a bucket's cached metadata from the in-memory map.
     ///
     /// Returns `true` if an entry was present. Reserved meta buckets are ignored.
@@ -1431,6 +1482,71 @@ mod tests {
         assert_eq!(tags.tag_set.len(), WRITERS, "every concurrent rewrite must survive: {tags:?}");
     }
 
+    /// Pins the peer reload-notification contract (`reload_from_store`, the
+    /// LoadBucketMetadata RPC path): only metadata actually read from
+    /// persisted storage enters the cache. A load miss errors out and leaves
+    /// the cache untouched — it must neither install a fabricated default
+    /// for an unknown bucket nor replace an existing entry, since a
+    /// transient ConfigNotFound during the notification would otherwise
+    /// downgrade a lock-enabled bucket to an authoritative "no Object Lock"
+    /// default and disable the batch-delete retention gate on this peer.
+    #[tokio::test]
+    async fn peer_reload_never_caches_fabricated_defaults_as_authoritative() {
+        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let sys = BucketMetadataSys::new(ecstore.clone());
+
+        // (a) Miss with no cached entry: the reload fails and installs nothing.
+        let err = sys
+            .reload_from_store("reload-bucket")
+            .await
+            .expect_err("a reload miss must be reported to the notifying peer");
+        assert!(
+            err.to_string().contains("no persisted bucket metadata readable"),
+            "the miss must surface through the dedicated non-persisted branch, got: {err}"
+        );
+        assert!(
+            sys.get("reload-bucket").await.is_err(),
+            "a reload miss must not install a fabricated default"
+        );
+
+        // (b) Miss with an existing entry: the reload fails and the entry
+        // (standing in for a lock-enabled bucket's metadata) survives intact.
+        let mut kept = BucketMetadata::new("reload-bucket");
+        kept.object_lock_config_xml = b"<ObjectLockConfiguration/>".to_vec();
+        sys.set("reload-bucket".to_string(), Arc::new(kept)).await;
+        assert!(sys.reload_from_store("reload-bucket").await.is_err());
+        let cached = sys
+            .get("reload-bucket")
+            .await
+            .expect("existing entry must survive a reload miss");
+        assert_eq!(
+            cached.object_lock_config_xml,
+            b"<ObjectLockConfiguration/>".to_vec(),
+            "a reload miss must not replace the cached entry with a fabricated default"
+        );
+
+        // (c) Persisted metadata reloads over a stale cached entry: the
+        // reload converges the cache to disk truth.
+        let mut persisted = BucketMetadata::new("reload-bucket");
+        persisted.policy_config_json = b"persisted-marker".to_vec();
+        sys.persist_and_set(persisted).await.expect("metadata should persist");
+        let mut stale = BucketMetadata::new("reload-bucket");
+        stale.policy_config_json = b"stale-cache-marker".to_vec();
+        sys.set("reload-bucket".to_string(), Arc::new(stale)).await;
+        sys.reload_from_store("reload-bucket")
+            .await
+            .expect("persisted metadata should reload");
+        let cached = sys
+            .get("reload-bucket")
+            .await
+            .expect("reloaded persisted metadata must be cached");
+        assert_eq!(
+            cached.policy_config_json,
+            b"persisted-marker".to_vec(),
+            "a reload must converge the cache to the persisted disk state"
+        );
+    }
+
     fn target(bucket: &str, id: &str) -> BucketTarget {
         BucketTarget {
             source_bucket: bucket.to_string(),

crates/keystone/Cargo.toml

@@ -25,6 +25,9 @@ keywords = ["rustfs", "openstack", "keystone", "authentication", "s3"]
 categories = ["authentication", "web-programming"]
 authors.workspace = true
 
+[lints]
+workspace = true
+
 [dependencies]
 tokio = { workspace = true, features = ["rt", "sync"] }
 reqwest = { workspace = true, features = ["json"] }

crates/scanner/src/data_usage_define.rs

@@ -698,7 +698,7 @@ impl DataUsageCache {
                 let mut visited = HashSet::new();
                 visited.insert(hash_path(path).key());
                 let mut flat = self.flatten_with_guard(root, &mut visited, 0);
-                if flat.replication_stats.as_ref().is_some_and(|stats| stats.empty()) {
+                if flat.replication_stats.as_ref().is_some_and(|stats| stats.is_empty()) {
                     flat.replication_stats = None;
                 }
                 Some(flat)
@@ -1574,6 +1574,7 @@ mod tests {
     use super::*;
     use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO};
     use crate::{ScannerGetObjectReader, ScannerPutObjReader};
+    use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
     use serde_json::Value;
     use std::io::Cursor;
     use std::pin::Pin;
@@ -2767,6 +2768,55 @@ mod tests {
         assert!(flat.children.is_empty());
     }
 
+    #[test]
+    fn size_recursive_prunes_empty_and_preserves_threshold_replication_stats() {
+        let root = hash_path("bucket");
+        let child = hash_path("bucket/child");
+        let mut cache = DataUsageCache::default();
+        cache.replace_hashed(&root, &None, &DataUsageEntry::default());
+        cache.replace_hashed(
+            &child,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                replication_stats: Some(ReplicationAllStats::default()),
+                ..Default::default()
+            },
+        );
+
+        assert!(
+            cache
+                .size_recursive("bucket")
+                .expect("scanner bucket usage should flatten")
+                .replication_stats
+                .is_none()
+        );
+
+        cache.replace_hashed(
+            &child,
+            &Some(root.clone()),
+            &DataUsageEntry {
+                replication_stats: Some(ReplicationAllStats {
+                    targets: HashMap::from([(
+                        "arn:test:threshold".to_string(),
+                        ReplicationStats {
+                            after_threshold_count: 1,
+                            ..Default::default()
+                        },
+                    )]),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            },
+        );
+
+        let flattened = cache.size_recursive("bucket").expect("scanner bucket usage should flatten");
+        let replication = flattened
+            .replication_stats
+            .expect("threshold-only replication stats must survive pruning");
+
+        assert_eq!(replication.targets["arn:test:threshold"].after_threshold_count, 1);
+    }
+
     #[test]
     fn checked_flatten_rejects_dangling_child() {
         let root_key = hash_path("bucket").key();

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -60,6 +60,7 @@ use uuid::Uuid;
 static GLOBAL_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
 static INIT: Once = Once::new();
 const TRANSITION_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
+const RESTORE_SUSPENDED_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
 const ENV_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE";
 const ENV_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOUT";
 const ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED";
@@ -161,6 +162,23 @@ async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
         .expect("Failed to create test bucket");
 }
 
+async fn suspend_test_bucket(bucket: &str) {
+    DefaultBucketUsecase::from_global()
+        .execute_put_bucket_versioning(build_request(
+            PutBucketVersioningInput::builder()
+                .bucket(bucket.to_string())
+                .versioning_configuration(VersioningConfiguration {
+                    status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED)),
+                    ..Default::default()
+                })
+                .build()
+                .expect("suspended versioning request should build"),
+            Method::PUT,
+        ))
+        .await
+        .expect("bucket versioning should be suspended");
+}
+
 async fn upload_test_object(ecstore: &Arc<ECStore>, bucket: &str, object: &str, data: &[u8]) -> ObjectInfo {
     let mut reader = PutObjReader::from_vec(data.to_vec());
     (**ecstore)
@@ -338,6 +356,45 @@ async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str,
     }
 }
 
+async fn wait_for_restore_completion(
+    ecstore: &Arc<ECStore>,
+    backend: &MockWarmBackend,
+    bucket: &str,
+    object: &str,
+    timeout: Duration,
+) -> Result<ObjectInfo, String> {
+    let deadline = tokio::time::Instant::now() + timeout;
+    let mut last_state = None;
+
+    loop {
+        if tokio::time::Instant::now() >= deadline {
+            let tier_gets = backend.get_count().await;
+            let op_log = backend.op_log().await;
+            return Err(format!(
+                "restore copy-back should complete within {timeout:?}; tier_gets={tier_gets}, op_log={op_log:?}; last observed state: {}",
+                last_state.unwrap_or_else(|| "no object info observed".to_string())
+            ));
+        }
+
+        match (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
+            Ok(info) => {
+                if !info.restore_ongoing && info.restore_expires.is_some() {
+                    return Ok(info);
+                }
+                last_state = Some(format!(
+                    "restore_ongoing={}, restore_expires={:?}, transitioned_status={}",
+                    info.restore_ongoing, info.restore_expires, info.transitioned_object.status
+                ));
+            }
+            Err(err) => {
+                last_state = Some(format!("get_object_info failed: {err}"));
+            }
+        }
+
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    }
+}
+
 // SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio
 // runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the
 // `env::set_var` / `env::remove_var` window.
@@ -2084,9 +2141,9 @@ async fn restore_object_usecase_reports_ongoing_conflict() {
 
     create_test_bucket(&ecstore, bucket.as_str()).await;
     let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
+    assert!(uploaded.version_id.is_none(), "fixture must create the null version");
     let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
     backend.clear_op_log().await;
-
     let get_barrier = backend.arm_get_barrier().await;
 
     let restore_request = || RestoreRequest {
@@ -2138,6 +2195,73 @@ async fn restore_object_usecase_reports_ongoing_conflict() {
     get_barrier.release();
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[serial]
+#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#4879"]
+async fn restore_object_usecase_completes_suspended_null_version_in_place() {
+    let (_disk_paths, ecstore) = setup_test_env().await;
+    let usecase = DefaultObjectUsecase::from_global();
+    let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+    let backend = register_mock_tier(&tier_name).await;
+    let bucket = format!("test-api-restore-suspended-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let object = "test/restore/suspended-null.bin";
+    let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();
+
+    create_test_bucket(&ecstore, bucket.as_str()).await;
+    let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
+    assert!(uploaded.version_id.is_none(), "fixture must create the null version");
+    let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
+    suspend_test_bucket(bucket.as_str()).await;
+    backend.clear_op_log().await;
+    let tier_gets_before_restore = backend.get_count().await;
+    let get_barrier = backend.arm_get_barrier().await;
+
+    Box::pin(
+        usecase.execute_restore_object(build_request(
+            RestoreObjectInput::builder()
+                .bucket(bucket.clone())
+                .key(object.to_string())
+                .restore_request(Some(RestoreRequest {
+                    days: Some(1),
+                    description: None,
+                    glacier_job_parameters: None,
+                    output_location: None,
+                    select_parameters: None,
+                    tier: None,
+                    type_: None,
+                }))
+                .build()
+                .expect("restore request should build"),
+            Method::POST,
+        )),
+    )
+    .await
+    .expect("suspended null-version restore should be accepted");
+
+    get_barrier.wait_until_paused().await;
+    get_barrier.release();
+    let completed = wait_for_restore_completion(&ecstore, &backend, bucket.as_str(), object, RESTORE_SUSPENDED_WAIT_TIMEOUT)
+        .await
+        .unwrap_or_else(|err| panic!("{err}"));
+
+    assert!(!completed.restore_ongoing, "the original null version must complete in place");
+    assert!(completed.restore_expires.is_some(), "completed restore must carry an expiry");
+    assert!(
+        completed.version_id.is_none() || completed.version_id.is_some_and(|version_id| version_id.is_nil()),
+        "suspended restore must remain on the null version"
+    );
+    assert_eq!(
+        live_object_version_count(&ecstore, bucket.as_str(), object).await,
+        1,
+        "suspended restore must not create a UUID version"
+    );
+    assert_eq!(
+        backend.get_count().await - tier_gets_before_restore,
+        1,
+        "suspended restore copy-back must fetch the tier exactly once"
+    );
+}
+
 /// backlog#1304: the restore-accept compare-and-set itself, under real
 /// concurrency. Two POST ?restore for the same transitioned object race each
 /// other from separate tasks; the accept guard must let exactly one through —

rustfs/src/app/object_usecase.rs

@@ -7783,7 +7783,12 @@ impl DefaultObjectUsecase {
         let bucket_clone = bucket.clone();
         let object_clone = object.clone();
         let rreq_clone = rreq.clone();
-        let version_id_clone = obj_info_.version_id.map(|v| v.to_string());
+        let version_id_clone = obj_info_
+            .version_id
+            .map(|v| v.to_string())
+            .or_else(|| (opts.versioned || opts.version_suspended).then(|| Uuid::nil().to_string()));
+        let versioned = opts.versioned;
+        let version_suspended = opts.version_suspended;
         let mut restore_operation_metadata = HashMap::new();
         if let Some(id) = restore_operation_id {
             insert_str(&mut restore_operation_metadata, SUFFIX_RESTORE_OPERATION_ID, id.to_string());
@@ -7797,6 +7802,8 @@ impl DefaultObjectUsecase {
                     ..Default::default()
                 },
                 version_id: version_id_clone,
+                versioned,
+                version_suspended,
                 user_defined: restore_operation_metadata,
                 ..Default::default()
             };

rustfs/src/storage/mod.rs

@@ -69,9 +69,9 @@ pub(crate) use storage_api::{
     get_lock_acquire_timeout, get_public_access_block_config, head_prefix_consumer, helper_consumer, init_background_replication,
     init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx, init_lock_clients,
     is_all_buckets_not_found, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class,
-    load_bucket_metadata, options_consumer, prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy,
-    rpc_consumer, runtime_sources_consumer, s3_api_consumer, serialize, set_bucket_metadata, table_catalog_path_hash,
-    to_s3s_etag, topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, try_migrate_iam_config,
+    options_consumer, prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, rpc_consumer,
+    runtime_sources_consumer, s3_api_consumer, serialize, table_catalog_path_hash, to_s3s_etag,
+    topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, try_migrate_iam_config,
     try_migrate_server_config, update_bucket_metadata_config, verify_rpc_signature, wrap_reader,
 };
 

rustfs/src/storage/rpc/node_service.rs

@@ -4439,6 +4439,31 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_load_bucket_metadata_failure_skips_scanner_maintenance() {
+        let service = create_test_node_service();
+        let maintenance_generation = rustfs_scanner::scanner_maintenance_generation();
+
+        let request = Request::new(LoadBucketMetadataRequest {
+            bucket: "reload-miss-scanner-guard-bucket".to_string(),
+            scanner_maintenance_change: true,
+        });
+
+        let response = service.load_bucket_metadata(request).await.expect("rpc should reply");
+        let load_response = response.into_inner();
+
+        // Whether the reload fails on missing server state or on the absent
+        // persisted metadata, a failed reload must report failure and must
+        // not tell the scanner a maintenance change landed.
+        assert!(!load_response.success);
+        assert!(load_response.error_info.is_some());
+        assert_eq!(
+            rustfs_scanner::scanner_maintenance_generation(),
+            maintenance_generation,
+            "a failed metadata reload must not advance scanner maintenance activity"
+        );
+    }
+
     #[tokio::test]
     #[ignore = "requires isolated global object layer state"]
     async fn test_load_bucket_metadata_no_object_layer() {

rustfs/src/storage/rpc/node_service/bucket.rs

@@ -17,7 +17,7 @@ use crate::storage::storage_api::rpc_consumer::node_service::contract::bucket::{
     BucketOptions, DeleteBucketOptions, MakeBucketOptions,
 };
 use crate::storage::storage_api::rpc_consumer::node_service::{
-    DiskError, StoragePeerS3ClientExt as _, load_bucket_metadata, remove_bucket_metadata, set_bucket_metadata,
+    DiskError, StoragePeerS3ClientExt as _, reload_bucket_metadata, remove_bucket_metadata,
 };
 use rustfs_common::heal_channel::HealOpts;
 use rustfs_protos::proto_gen::node_service::*;
@@ -66,21 +66,15 @@ impl NodeService {
             }));
         }
 
-        let Some(store) = self.resolve_object_store() else {
+        let Some(_store) = self.resolve_object_store() else {
             return Ok(Response::new(LoadBucketMetadataResponse {
                 success: false,
                 error_info: Some("errServerNotInitialized".to_string()),
             }));
         };
 
-        match load_bucket_metadata(store, &bucket).await {
-            Ok(meta) => {
-                if let Err(err) = set_bucket_metadata(bucket.clone(), meta).await {
-                    return Ok(Response::new(LoadBucketMetadataResponse {
-                        success: false,
-                        error_info: Some(err.to_string()),
-                    }));
-                };
+        match reload_bucket_metadata(&bucket).await {
+            Ok(()) => {
                 if scanner_maintenance_change {
                     rustfs_scanner::record_scanner_maintenance_change(&bucket);
                 }

rustfs/src/storage/storage_api.rs

@@ -236,8 +236,8 @@ pub(crate) mod rpc_consumer {
             ECStore, Error, FileInfoVersions, LocalPeerS3Client, MetricType, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS,
             ReadMultipleReq, ReadMultipleResp, ReadOptions, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC,
             StorageDiskRpcExt, StoragePeerS3ClientExt, UpdateMetadataOpts, all_local_disk_path, collect_local_metrics,
-            find_local_disk_by_ref, get_local_server_property, load_bucket_metadata, reload_transition_tier_config,
-            remove_bucket_metadata, set_bucket_metadata, validate_batch_read_version_item_count,
+            find_local_disk_by_ref, get_local_server_property, reload_bucket_metadata, reload_transition_tier_config,
+            remove_bucket_metadata, validate_batch_read_version_item_count,
         };
         pub(crate) type StorageResult<T> = super::super::Result<T>;
 
@@ -1365,10 +1365,6 @@ impl StoragePeerS3ClientExt for LocalPeerS3Client {
     }
 }
 
-pub(crate) async fn load_bucket_metadata(api: Arc<ECStore>, bucket: &str) -> Result<BucketMetadata> {
-    ecstore_bucket::metadata::load_bucket_metadata(api, bucket).await
-}
-
 #[cfg(test)]
 pub(crate) fn bucket_metadata_sys_initialized() -> bool {
     ecstore_bucket::metadata_sys::get_global_bucket_metadata_sys().is_some()
@@ -1441,10 +1437,15 @@ pub(crate) async fn get_bucket_website_config(bucket: &str) -> Result<(s3s::dto:
     ecstore_bucket::metadata_sys::get_website_config(bucket).await
 }
 
+#[cfg(test)]
 pub(crate) async fn set_bucket_metadata(bucket: String, bm: BucketMetadata) -> Result<()> {
     ecstore_bucket::metadata_sys::set_bucket_metadata(bucket, bm).await
 }
 
+pub(crate) async fn reload_bucket_metadata(bucket: &str) -> Result<()> {
+    ecstore_bucket::metadata_sys::reload_bucket_metadata(bucket).await
+}
+
 pub(crate) async fn remove_bucket_metadata(bucket: &str) -> Result<bool> {
     ecstore_bucket::metadata_sys::remove_bucket_metadata(bucket).await
 }