test(heal): cover suspended owner production path

.github/workflows/nightly-gnu.yml

@@ -39,10 +39,11 @@ jobs:
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
     steps:
-      - name: Checkout repository
+      - name: Checkout main branch
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -88,10 +89,11 @@ jobs:
       # either casing.
       NO_PROXY: 127.0.0.1,localhost
     steps:
-      - name: Checkout repository
+      - name: Checkout main branch
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -176,10 +178,11 @@ jobs:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
       NO_PROXY: 127.0.0.1,localhost
     steps:
-      - name: Checkout repository
+      - name: Checkout main branch
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup

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

@@ -50,10 +50,10 @@ use rustfs_protos::evict_failed_connection;
 use rustfs_protos::proto_gen::node_service::RenamePartRequest;
 use rustfs_protos::proto_gen::node_service::{
     BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, DeletePathsRequest, DeleteRequest,
-    DeleteVersionRequest, DeleteVersionsRequest, DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest,
-    ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest,
-    ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest,
-    RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
+    DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest,
+    MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest,
+    ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
+    RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
     SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
     WriteMetadataRequest, node_service_client::NodeServiceClient,
 };
@@ -112,28 +112,6 @@ const EVENT_REMOTE_DISK_RPC: &str = "remote_disk_rpc";
 const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
 pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60);
 
-fn decode_delete_versions_errors(response: DeleteVersionsResponse, expected_len: usize) -> Vec<Option<Error>> {
-    if !response.item_errors.is_empty() {
-        if response.item_errors.len() != expected_len {
-            return vec![Some(Error::other("malformed delete_versions item errors")); expected_len];
-        }
-        return response
-            .item_errors
-            .into_iter()
-            .map(|error| (error.code != 0).then(|| error.into()))
-            .collect();
-    }
-
-    if response.errors.len() != expected_len {
-        return vec![Some(Error::other("malformed delete_versions errors")); expected_len];
-    }
-    response
-        .errors
-        .into_iter()
-        .map(|error| (!error.is_empty()).then(|| Error::other(error)))
-        .collect()
-}
-
 fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result<SnapshotLeaseToken> {
     if !response.success {
         return Err(response.error.unwrap_or_default().into());
@@ -2428,6 +2406,8 @@ impl DiskAPI for RemoteDisk {
             return errors;
         }
 
+        // TODO(backlog): replace string errors with typed `StorageError` variants
+
         let result = self
             .execute_with_timeout(
                 || async {
@@ -2459,7 +2439,17 @@ impl DiskAPI for RemoteDisk {
             }
             return errors;
         }
-        decode_delete_versions_errors(response, versions.len())
+        response
+            .errors
+            .iter()
+            .map(|error| {
+                if error.is_empty() {
+                    None
+                } else {
+                    Some(Error::other(error.to_string()))
+                }
+            })
+            .collect()
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
@@ -3770,63 +3760,6 @@ mod tests {
 
     static INIT: Once = Once::new();
 
-    #[test]
-    fn delete_versions_response_preserves_typed_item_errors() {
-        let errors = decode_delete_versions_errors(
-            DeleteVersionsResponse {
-                success: true,
-                errors: vec!["file not found".to_string(), String::new()],
-                error: None,
-                item_errors: vec![
-                    rustfs_protos::proto_gen::node_service::Error {
-                        code: DiskError::FileNotFound.to_u32(),
-                        error_info: "file not found".to_string(),
-                    },
-                    rustfs_protos::proto_gen::node_service::Error::default(),
-                ],
-            },
-            2,
-        );
-
-        assert!(matches!(errors.as_slice(), [Some(DiskError::FileNotFound), None]));
-    }
-
-    #[test]
-    fn delete_versions_response_accepts_legacy_string_errors() {
-        let errors = decode_delete_versions_errors(
-            DeleteVersionsResponse {
-                success: true,
-                errors: vec!["legacy error".to_string(), String::new()],
-                error: None,
-                item_errors: Vec::new(),
-            },
-            2,
-        );
-
-        assert_eq!(errors.len(), 2);
-        assert_eq!(errors[0].as_ref().map(ToString::to_string).as_deref(), Some("io error legacy error"));
-        assert!(errors[1].is_none());
-    }
-
-    #[test]
-    fn delete_versions_response_rejects_misaligned_item_errors() {
-        let errors = decode_delete_versions_errors(
-            DeleteVersionsResponse {
-                success: true,
-                errors: vec!["file not found".to_string()],
-                error: None,
-                item_errors: vec![rustfs_protos::proto_gen::node_service::Error {
-                    code: DiskError::FileNotFound.to_u32(),
-                    error_info: "file not found".to_string(),
-                }],
-            },
-            2,
-        );
-
-        assert_eq!(errors.len(), 2);
-        assert!(errors.iter().all(Option::is_some));
-    }
-
     #[test]
     fn disk_mutation_digest_marks_rolling_compatibility() {
         let mut request = Request::new(());

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/heal/src/heal/channel.rs

@@ -1640,60 +1640,6 @@ mod tests {
         assert_eq!(payload["items"].as_array().expect("items should be an array").len(), 0);
     }
 
-    #[tokio::test]
-    async fn test_process_query_request_reports_displaced_terminal_detail() {
-        let heal_manager = Arc::new(HealManager::new(
-            Arc::new(MockStorage),
-            Some(HealConfig {
-                queue_size: 1,
-                ..HealConfig::default()
-            }),
-        ));
-        let mut displaced = HealRequest::new(
-            HealType::Bucket {
-                bucket: "displaced-channel".to_string(),
-            },
-            HealOptions::default(),
-            HealPriority::Low,
-        );
-        displaced.id = "displaced-channel-task".to_string();
-        let displaced_id = displaced.id.clone();
-        heal_manager
-            .submit_heal_request(displaced)
-            .await
-            .expect("initial channel task should queue");
-        heal_manager
-            .submit_heal_request(HealRequest::new(
-                HealType::Bucket {
-                    bucket: "successor-channel".to_string(),
-                },
-                HealOptions::default(),
-                HealPriority::High,
-            ))
-            .await
-            .expect("successor channel task should displace the initial task");
-
-        let processor = HealChannelProcessor::new(heal_manager);
-        let (tx, rx) = oneshot::channel();
-        processor
-            .process_query_request("displaced-channel".to_string(), displaced_id, None, tx)
-            .await
-            .expect("displaced query should process");
-        let response = rx
-            .await
-            .expect("query response should be returned")
-            .expect("displaced query should remain successful");
-        let payload: serde_json::Value = serde_json::from_slice(response.data.as_deref().expect("status payload should exist"))
-            .expect("status payload should be json");
-        assert_eq!(payload["summary"], "stopped");
-        assert!(
-            response
-                .error
-                .as_deref()
-                .is_some_and(|detail| detail.contains("reason=displaced"))
-        );
-    }
-
     #[tokio::test]
     async fn test_process_query_request_reports_running_for_queued_task() {
         let heal_manager = create_test_heal_manager();

crates/heal/src/heal/manager.rs

@@ -40,7 +40,6 @@ use tracing::{debug, error, info, warn};
 use super::{DiskError, Endpoint, HealDiskExt as _, local_disk_map_read};
 
 const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
-const DISPLACED_HEAL_REASON: &str = "reason=displaced; retry_hint=submit_again";
 const LOG_COMPONENT_HEAL: &str = "heal";
 const LOG_SUBSYSTEM_DISK_SCANNER: &str = "disk_scanner";
 const LOG_SUBSYSTEM_MANAGER: &str = "manager";
@@ -121,30 +120,26 @@ struct MrfRepairNoticeTarget {
     version_id: Option<[u8; 16]>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 struct HealAdmissionDecision {
     result: HealAdmissionResult,
-    displaced_request: Option<HealRequest>,
+    displaced_task_id: Option<String>,
 }
 
 impl HealAdmissionDecision {
     const fn new(result: HealAdmissionResult) -> Self {
         Self {
             result,
-            displaced_request: None,
+            displaced_task_id: None,
         }
     }
 
-    fn accepted_with_displacement(displaced_request: HealRequest) -> Self {
+    fn accepted_with_displacement(displaced_task_id: String) -> Self {
         Self {
             result: HealAdmissionResult::Accepted,
-            displaced_request: Some(displaced_request),
+            displaced_task_id: Some(displaced_task_id),
         }
     }
-
-    fn displaced_task_id(&self) -> Option<&str> {
-        self.displaced_request.as_ref().map(|request| request.id.as_str())
-    }
 }
 
 fn lock_mrf_repair_notice_targets(
@@ -156,55 +151,6 @@ fn lock_mrf_repair_notice_targets(
     }
 }
 
-fn lock_displaced_terminals(
-    registry: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
-) -> StdMutexGuard<'_, HashMap<String, Arc<CompletedHealStatus>>> {
-    match registry.lock() {
-        Ok(guard) => guard,
-        Err(poisoned) => poisoned.into_inner(),
-    }
-}
-
-fn record_displaced_terminal(
-    registry: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
-    request: &HealRequest,
-) -> Arc<CompletedHealStatus> {
-    let terminal = Arc::new(CompletedHealStatus {
-        heal_type: request.heal_type.clone(),
-        status: HealTaskStatus::Failed {
-            error: format!("heal task displaced by a higher-priority request ({DISPLACED_HEAL_REASON})"),
-        },
-        result_items_truncated: false,
-        completed_at: SystemTime::now(),
-        seqed_items: Vec::new(),
-        next_seq: 0,
-        min_seq: 0,
-    });
-    let mut terminals = lock_displaced_terminals(registry);
-    prune_completed_heal_statuses(&mut terminals);
-    terminals.insert(request.id.clone(), Arc::clone(&terminal));
-    terminal
-}
-
-async fn remove_displaced_task_aliases(
-    aliases: &Arc<Mutex<HashMap<String, HealTaskAlias>>>,
-    terminals: &StdMutex<HashMap<String, Arc<CompletedHealStatus>>>,
-    task_id: &str,
-    terminal: &Arc<CompletedHealStatus>,
-) {
-    let mut aliases = aliases.lock().await;
-    let alias_ids = aliases
-        .iter()
-        .filter_map(|(alias_id, alias)| (alias.task_id == task_id).then_some(alias_id.clone()))
-        .collect::<Vec<_>>();
-    let mut displaced_terminals = lock_displaced_terminals(terminals);
-    prune_completed_heal_statuses(&mut displaced_terminals);
-    for alias_id in alias_ids {
-        displaced_terminals.insert(alias_id, Arc::clone(terminal));
-    }
-    aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
-}
-
 async fn remove_task_aliases_for_task(registry: &Arc<Mutex<HashMap<String, HealTaskAlias>>>, task_id: &str) {
     registry
         .lock()
@@ -672,14 +618,6 @@ pub struct HealManager {
     /// are shared so the lookup helper can hand a completed entry to a
     /// caller without cloning the retained result window.
     completed_heals: Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
-    /// Terminals for requests removed by priority displacement. An Accepted
-    /// task ID remains queryable for the same process lifetime and the normal
-    /// ten-minute status TTL; clients should treat `reason=displaced` as a
-    /// terminal result and submit a fresh request. This sidecar is synchronous
-    /// so admission can publish the terminal while the queue transition is
-    /// still under its lock, without awaiting another tokio lock. Queue state
-    /// is process-local, so this guarantee does not extend across restart.
-    displaced_terminals: Arc<StdMutex<HashMap<String, Arc<CompletedHealStatus>>>>,
     /// Client tokens merged into an existing task id.
     task_aliases: Arc<Mutex<HashMap<String, HealTaskAlias>>>,
     /// Heal tasks waiting for a retry backoff to expire.
@@ -721,7 +659,6 @@ struct HealQueueContext<'a> {
     heal_queue: &'a Arc<Mutex<PriorityHealQueue>>,
     active_heals: &'a Arc<Mutex<HashMap<String, Arc<HealTask>>>>,
     completed_heals: &'a Arc<Mutex<HashMap<String, Arc<CompletedHealStatus>>>>,
-    displaced_terminals: &'a Arc<StdMutex<HashMap<String, Arc<CompletedHealStatus>>>>,
     task_aliases: &'a Arc<Mutex<HashMap<String, HealTaskAlias>>>,
     retrying_heals: &'a Arc<Mutex<HashMap<String, RetryingHeal>>>,
     mrf_repair_notice_targets: &'a Arc<StdMutex<HashMap<String, Vec<MrfRepairNoticeTarget>>>>,
@@ -937,7 +874,7 @@ impl HealManager {
                         result = "accepted_by_displacement",
                         "Heal queue request accepted by displacement"
                     });
-                    return HealAdmissionDecision::accepted_with_displacement(displaced);
+                    return HealAdmissionDecision::accepted_with_displacement(displaced.id);
                 }
 
                 demote_to_debug_when!(per_object_request, warn, target: "rustfs::heal::manager", {
@@ -1168,7 +1105,6 @@ impl HealManager {
             active_heals: Arc::new(Mutex::new(HashMap::new())),
             heal_queue: Arc::new(Mutex::new(PriorityHealQueue::new())),
             completed_heals: Arc::new(Mutex::new(HashMap::new())),
-            displaced_terminals: Arc::new(StdMutex::new(HashMap::new())),
             task_aliases: Arc::new(Mutex::new(HashMap::new())),
             retrying_heals: Arc::new(Mutex::new(HashMap::new())),
             mrf_repair_notice_targets: Arc::new(StdMutex::new(HashMap::new())),
@@ -1273,10 +1209,6 @@ impl HealManager {
         active_heals.clear();
         publish_active_heal_count(&active_heals);
         self.completed_heals.lock().await.clear();
-        // Do not let the synchronous guard live across the following async lock.
-        {
-            lock_displaced_terminals(&self.displaced_terminals).clear();
-        }
         self.task_aliases.lock().await.clear();
         self.retrying_heals.lock().await.clear();
         lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets).clear();
@@ -1527,11 +1459,7 @@ impl HealManager {
             task_id = queued_id.to_owned();
         }
         let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
-        let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
-        let displaced_terminal = admission_decision
-            .displaced_request
-            .as_ref()
-            .map(|request| record_displaced_terminal(&self.displaced_terminals, request));
+        let displaced_task_id = admission_decision.displaced_task_id;
         if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged)
             && let Some(target) = mrf_notice_target
         {
@@ -1545,12 +1473,8 @@ impl HealManager {
         drop(queue);
         drop(active_heals);
 
-        if let (Some(displaced_task_id), Some(displaced_terminal)) = (displaced_task_id, displaced_terminal) {
-            // The queue has already removed the displaced request, so the
-            // synchronous terminal sidecar was published before aliases and
-            // MRF ownership are cleaned up.
-            remove_displaced_task_aliases(&self.task_aliases, &self.displaced_terminals, &displaced_task_id, &displaced_terminal)
-                .await;
+        if let Some(displaced_task_id) = displaced_task_id {
+            self.remove_aliases_for_task(&displaced_task_id).await;
         }
 
         if should_notify {
@@ -1625,15 +1549,6 @@ impl HealManager {
             }
         }
 
-        if terminal_completed.is_none() {
-            let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
-            prune_completed_heal_statuses(&mut displaced_terminals);
-            terminal_completed = displaced_terminals
-                .get(canonical_task_id)
-                .filter(|terminal| matches_path(&terminal.heal_type))
-                .cloned();
-        }
-
         match terminal_completed {
             Some(completed) => TaskStateLookup::Completed(completed),
             None => TaskStateLookup::NotFound,
@@ -1754,19 +1669,9 @@ impl HealManager {
 
         let mut completed_heals = self.completed_heals.lock().await;
         prune_completed_heal_statuses(&mut completed_heals);
-        if completed_heals
+        completed_heals
             .values()
             .any(|completed| heal_type_matches_path(&completed.heal_type, heal_path))
-        {
-            return true;
-        }
-        drop(completed_heals);
-
-        let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
-        prune_completed_heal_statuses(&mut displaced_terminals);
-        displaced_terminals
-            .values()
-            .any(|terminal| heal_type_matches_path(&terminal.heal_type, heal_path))
     }
 
     /// Get task progress

crates/heal/src/heal/manager/auto_scan.rs

@@ -21,7 +21,6 @@ impl HealManager {
         let heal_queue = self.heal_queue.clone();
         let active_heals = self.active_heals.clone();
         let task_aliases = self.task_aliases.clone();
-        let displaced_terminals = self.displaced_terminals.clone();
         let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
         let storage = self.storage.clone();
         let replacement_recovery_anchors = self.replacement_recovery_anchors.clone();
@@ -482,10 +481,6 @@ impl HealManager {
                             let admission = admission_decision.result;
                             let should_notify =
                                 matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
-                            let displaced_terminal = admission_decision
-                                .displaced_request
-                                .as_ref()
-                                .map(|request| record_displaced_terminal(&displaced_terminals, request));
                             if matches!(admission, HealAdmissionResult::Accepted)
                                 && let Some(anchor) = recovery_anchor
                             {
@@ -496,16 +491,8 @@ impl HealManager {
                             }
                             drop(queue);
                             drop(config);
-                            if let (Some(displaced_task_id), Some(displaced_terminal)) =
-                                (admission_decision.displaced_task_id().map(ToOwned::to_owned), displaced_terminal)
-                            {
-                                remove_displaced_task_aliases(
-                                    &task_aliases,
-                                    &displaced_terminals,
-                                    &displaced_task_id,
-                                    &displaced_terminal,
-                                )
-                                .await;
+                            if let Some(displaced_task_id) = admission_decision.displaced_task_id {
+                                remove_task_aliases_for_task(&task_aliases, &displaced_task_id).await;
                                 lock_mrf_repair_notice_targets(&mrf_repair_notice_targets).remove(&displaced_task_id);
                             }
                             if matches!(admission, HealAdmissionResult::Accepted) {

crates/heal/src/heal/manager/scheduler.rs

@@ -21,7 +21,6 @@ impl HealManager {
         let heal_queue = self.heal_queue.clone();
         let active_heals = self.active_heals.clone();
         let completed_heals = self.completed_heals.clone();
-        let displaced_terminals = self.displaced_terminals.clone();
         let task_aliases = self.task_aliases.clone();
         let retrying_heals = self.retrying_heals.clone();
         let mrf_repair_notice_targets = self.mrf_repair_notice_targets.clone();
@@ -54,7 +53,6 @@ impl HealManager {
                             heal_queue: &heal_queue,
                             active_heals: &active_heals,
                             completed_heals: &completed_heals,
-                            displaced_terminals: &displaced_terminals,
                             task_aliases: &task_aliases,
                             retrying_heals: &retrying_heals,
                             mrf_repair_notice_targets: &mrf_repair_notice_targets,
@@ -73,7 +71,6 @@ impl HealManager {
                             heal_queue: &heal_queue,
                             active_heals: &active_heals,
                             completed_heals: &completed_heals,
-                            displaced_terminals: &displaced_terminals,
                             task_aliases: &task_aliases,
                             retrying_heals: &retrying_heals,
                             mrf_repair_notice_targets: &mrf_repair_notice_targets,
@@ -101,7 +98,6 @@ impl HealManager {
             heal_queue,
             active_heals,
             completed_heals,
-            displaced_terminals,
             task_aliases,
             retrying_heals,
             mrf_repair_notice_targets,
@@ -187,7 +183,6 @@ impl HealManager {
                 let active_heals_clone = active_heals.clone();
                 let heal_queue_clone = heal_queue.clone();
                 let completed_heals_clone = completed_heals.clone();
-                let displaced_terminals_clone = displaced_terminals.clone();
                 let task_aliases_clone = task_aliases.clone();
                 let retrying_heals_clone = retrying_heals.clone();
                 let mrf_repair_notice_targets_clone = mrf_repair_notice_targets.clone();
@@ -368,7 +363,6 @@ impl HealManager {
                         let retry_heal_queue = heal_queue_clone.clone();
                         let retrying_heals_for_spawn = retrying_heals_clone.clone();
                         let retry_task_aliases = task_aliases_clone.clone();
-                        let retry_displaced_terminals = displaced_terminals_clone.clone();
                         let retry_mrf_repair_notice_targets = mrf_repair_notice_targets_clone.clone();
                         let retry_completed_heals = completed_heals_clone.clone();
                         let retry_notify = notify_clone.clone();
@@ -436,14 +430,6 @@ impl HealManager {
                                 let admission = admission_decision.result;
                                 let should_notify = matches!(admission, HealAdmissionResult::Accepted)
                                     && retry_config.event_driven_scheduler_enable;
-                                // Publish the terminal synchronously while the
-                                // queue transition is protected. The subsequent
-                                // queue -> retrying handoff retains the lock order
-                                // used by operations_snapshot.
-                                let displaced_terminal = admission_decision
-                                    .displaced_request
-                                    .as_ref()
-                                    .map(|request| record_displaced_terminal(&retry_displaced_terminals, request));
                                 match admission {
                                     HealAdmissionResult::Accepted => {
                                         // Transfer ownership while holding queue -> retrying,
@@ -451,18 +437,10 @@ impl HealManager {
                                         #[cfg(test)]
                                         pause_retry_ownership_transition(&retry_request_id, true).await;
                                         retrying_heals_for_spawn.lock().await.remove(&retry_request_id);
-                                        let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
+                                        let displaced_task_id = admission_decision.displaced_task_id;
                                         drop(queue);
-                                        if let (Some(displaced_task_id), Some(displaced_terminal)) =
-                                            (displaced_task_id, displaced_terminal)
-                                        {
-                                            remove_displaced_task_aliases(
-                                                &retry_task_aliases,
-                                                &retry_displaced_terminals,
-                                                &displaced_task_id,
-                                                &displaced_terminal,
-                                            )
-                                            .await;
+                                        if let Some(displaced_task_id) = displaced_task_id {
+                                            remove_task_aliases_for_task(&retry_task_aliases, &displaced_task_id).await;
                                             remove_mrf_repair_notice_targets(
                                                 &retry_mrf_repair_notice_targets,
                                                 &displaced_task_id,

crates/heal/src/heal/manager/tests.rs

@@ -84,7 +84,6 @@ async fn process_manager_queue_once(manager: &HealManager) {
         heal_queue: &manager.heal_queue,
         active_heals: &manager.active_heals,
         completed_heals: &manager.completed_heals,
-        displaced_terminals: &manager.displaced_terminals,
         task_aliases: &manager.task_aliases,
         retrying_heals: &manager.retrying_heals,
         mrf_repair_notice_targets: &manager.mrf_repair_notice_targets,
@@ -2779,10 +2778,7 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() {
         HealAdmissionResult::Accepted
     );
     assert_eq!(manager.get_queue_length().await, 1);
-    assert!(matches!(
-        manager.get_task_status(&low_id).await,
-        Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced")
-    ));
+    assert!(matches!(manager.get_task_status(&low_id).await, Err(Error::TaskNotFound { .. })));
     assert_eq!(
         manager
             .get_task_status(&high_id)
@@ -2792,263 +2788,6 @@ async fn test_high_priority_request_displaces_lower_priority_when_queue_full() {
     );
 }
 
-#[tokio::test]
-async fn displaced_task_remains_queryable() {
-    let manager = HealManager::new(
-        Arc::new(MockStorage),
-        Some(HealConfig {
-            queue_size: 1,
-            ..HealConfig::default()
-        }),
-    );
-    let mut displaced = HealRequest::new(
-        HealType::Bucket {
-            bucket: "displaced-bucket".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::Low,
-    );
-    displaced.id = "displaced-task".to_string();
-    let displaced_id = displaced.id.clone();
-    manager
-        .submit_heal_request(displaced)
-        .await
-        .expect("displaced request should queue");
-
-    let successor = HealRequest::new(
-        HealType::Bucket {
-            bucket: "successor-bucket".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::High,
-    );
-    manager
-        .submit_heal_request(successor)
-        .await
-        .expect("successor should displace low work");
-
-    let report = manager
-        .get_task_report(&displaced_id)
-        .await
-        .expect("displaced report should remain queryable");
-    assert!(matches!(report.status, HealTaskStatus::Failed { ref error } if error.contains("reason=displaced")));
-}
-
-#[tokio::test]
-async fn displaced_archive_failure_keeps_queryable_terminal() {
-    let manager = HealManager::new(Arc::new(MockStorage), None);
-    let mut request = HealRequest::new(
-        HealType::Bucket {
-            bucket: "archive-failure".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::Low,
-    );
-    request.id = "archive-failure-task".to_string();
-    let request_id = request.id.clone();
-    // The synchronous sidecar is the authoritative fallback when the normal
-    // completed-task archive has no entry (the failure window that must not
-    // turn an Accepted ID into NotFound).
-    record_displaced_terminal(&manager.displaced_terminals, &request);
-    assert!(manager.completed_heals.lock().await.is_empty());
-    assert!(matches!(
-        manager.get_task_status(&request_id).await,
-        Ok(HealTaskStatus::Failed { error }) if error.contains("reason=displaced")
-    ));
-}
-
-#[tokio::test]
-async fn scheduler_retry_displacement_keeps_evicted_task_queryable() {
-    let manager = Arc::new(HealManager::new(
-        Arc::new(MockStorage),
-        Some(HealConfig {
-            queue_size: 1,
-            event_driven_scheduler_enable: false,
-            ..HealConfig::default()
-        }),
-    ));
-    let mut retry_request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
-    retry_request.priority = HealPriority::High;
-    let retry_id = retry_request.id.clone();
-    manager
-        .submit_heal_request(retry_request)
-        .await
-        .expect("retry request should queue");
-
-    // Process exactly one queue cycle so the retry task is spawned without a
-    // background scheduler consuming the filler request before the retry wakes.
-    process_manager_queue_once(&manager).await;
-    tokio::time::timeout(Duration::from_secs(1), async {
-        loop {
-            if manager.retrying_heals.lock().await.contains_key(&retry_id) {
-                break;
-            }
-            tokio::task::yield_now().await;
-        }
-    })
-    .await
-    .expect("retry request should enter backoff");
-
-    let filler = HealRequest::new(
-        HealType::Bucket {
-            bucket: "retry-displaced-filler".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::Low,
-    );
-    let filler_id = filler.id.clone();
-    manager
-        .submit_heal_request(filler)
-        .await
-        .expect("filler request should occupy the queue");
-
-    tokio::time::timeout(Duration::from_secs(5), async {
-        loop {
-            if matches!(
-                manager.get_task_status(&filler_id).await,
-                Ok(HealTaskStatus::Failed { ref error }) if error.contains("reason=displaced")
-            ) {
-                break;
-            }
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("retry admission should displace the filler request");
-    assert_eq!(manager.get_queue_length().await, 1);
-    assert_eq!(
-        manager.get_task_status(&retry_id).await.expect("retry should be queued"),
-        HealTaskStatus::Pending
-    );
-}
-
-#[tokio::test]
-async fn concurrent_displacers_produce_one_terminal_generation() {
-    let manager = Arc::new(HealManager::new(
-        Arc::new(MockStorage),
-        Some(HealConfig {
-            queue_size: 1,
-            ..HealConfig::default()
-        }),
-    ));
-    let mut displaced = HealRequest::new(
-        HealType::Bucket {
-            bucket: "concurrent-displaced".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::Low,
-    );
-    displaced.id = "concurrent-displaced-task".to_string();
-    let displaced_id = displaced.id.clone();
-    manager
-        .submit_heal_request(displaced)
-        .await
-        .expect("initial request should queue");
-
-    let first = HealRequest::new(
-        HealType::Bucket {
-            bucket: "concurrent-successor-a".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::High,
-    );
-    let second = HealRequest::new(
-        HealType::Bucket {
-            bucket: "concurrent-successor-b".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::High,
-    );
-    let (first_result, second_result) = tokio::join!(manager.submit_heal_request(first), manager.submit_heal_request(second));
-    let accepted = [&first_result, &second_result]
-        .into_iter()
-        .filter(|result| matches!(result, Ok(HealAdmissionResult::Accepted)))
-        .count();
-    assert_eq!(accepted, 1, "exactly one concurrent displacer should win the full queue");
-    assert!(
-        first_result.is_ok() && second_result.is_ok(),
-        "the losing request should receive a typed Full result"
-    );
-    let terminals = lock_displaced_terminals(&manager.displaced_terminals);
-    assert_eq!(terminals.len(), 1);
-    assert!(terminals.contains_key(&displaced_id));
-}
-
-#[tokio::test]
-async fn successor_chain_is_bounded_and_authorized() {
-    let manager = HealManager::new(
-        Arc::new(MockStorage),
-        Some(HealConfig {
-            queue_size: 1,
-            ..HealConfig::default()
-        }),
-    );
-    let mut original = HealRequest::new(
-        HealType::Bucket {
-            bucket: "authorized-original".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::Low,
-    );
-    original.id = "authorized-original-task".to_string();
-    let original_id = original.id.clone();
-    manager.submit_heal_request(original).await.expect("original should queue");
-    let mut duplicate = HealRequest::new(
-        HealType::Bucket {
-            bucket: "authorized-original".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::Low,
-    );
-    duplicate.id = "authorized-duplicate-task".to_string();
-    let duplicate_id = duplicate.id.clone();
-    manager
-        .submit_heal_request(duplicate)
-        .await
-        .expect("same-target duplicate should merge");
-    let successor = HealRequest::new(
-        HealType::Bucket {
-            bucket: "authorized-successor".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::High,
-    );
-    let successor_id = successor.id.clone();
-    manager.submit_heal_request(successor).await.expect("successor should queue");
-    assert!(manager.task_aliases.lock().await.is_empty());
-    assert!(matches!(manager.get_task_status(&original_id).await, Ok(HealTaskStatus::Failed { .. })));
-    assert!(matches!(manager.get_task_status(&duplicate_id).await, Ok(HealTaskStatus::Failed { .. })));
-    assert_eq!(
-        manager
-            .get_task_status(&successor_id)
-            .await
-            .expect("successor should remain queued"),
-        HealTaskStatus::Pending
-    );
-}
-
-#[tokio::test]
-async fn displaced_terminal_expires_after_bounded_ttl() {
-    let manager = HealManager::new(Arc::new(MockStorage), None);
-    let mut request = HealRequest::new(
-        HealType::Bucket {
-            bucket: "expires".to_string(),
-        },
-        HealOptions::default(),
-        HealPriority::Low,
-    );
-    request.id = "expires-task".to_string();
-    let request_id = request.id.clone();
-    record_displaced_terminal(&manager.displaced_terminals, &request);
-    {
-        let mut terminals = lock_displaced_terminals(&manager.displaced_terminals);
-        let entry =
-            Arc::get_mut(terminals.get_mut(&request_id).expect("terminal should be retained")).expect("test owns terminal entry");
-        entry.completed_at = SystemTime::now() - KEEP_HEAL_TASK_STATUS_DURATION - Duration::from_secs(1);
-    }
-    assert!(matches!(manager.get_task_status(&request_id).await, Err(Error::TaskNotFound { .. })));
-}
-
 #[tokio::test]
 async fn test_displacing_registered_mrf_task_drops_notice_ownership() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/kms/tests/vault_ha_failover_live.rs

@@ -16,14 +16,14 @@
 //!
 //! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
 //! kills the active node while this test continuously decrypts through a
-//! surviving standby. KV2 and Transit must recover after the bounded circuit
-//! interval, use a bounded number of attempts, and leave the circuit and
-//! in-flight gauges at zero after a new leader is elected.
+//! surviving standby. KV2 and Transit requests must remain successful, use a
+//! bounded number of attempts, and leave the circuit and in-flight gauges at
+//! zero after a new leader is elected.
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use metrics_util::MetricKind;
@@ -43,11 +43,6 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
 const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
 const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
 const MAX_ATTEMPTS: u32 = 10;
-const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
-const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
-// The circuit remains open for 30s after five failed attempts.
-const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
-const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);
 
 type MetricEntry = (
     metrics_util::CompositeKey,
@@ -69,7 +64,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
         backend,
         backend_config,
         allow_insecure_dev_defaults: true,
-        timeout: ATTEMPT_TIMEOUT,
+        timeout: Duration::from_secs(2),
         retry_attempts: MAX_ATTEMPTS,
         enable_cache: false,
         ..KmsConfig::default()
@@ -169,31 +164,14 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
         .sum()
 }
 
-async fn wait_for_count(
-    counter: &AtomicU64,
-    failure: &Mutex<Option<String>>,
-    minimum: u64,
-    description: &str,
-    timeout: Duration,
-) {
-    tokio::time::timeout(timeout, async {
+async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
+    tokio::time::timeout(Duration::from_secs(20), async {
         while counter.load(Ordering::SeqCst) < minimum {
-            if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
-                panic!(
-                    "{description} worker failed after {} successful decrypts: {error}",
-                    counter.load(Ordering::SeqCst)
-                );
-            }
             tokio::time::sleep(Duration::from_millis(25)).await;
         }
     })
     .await
-    .unwrap_or_else(|_| {
-        panic!(
-            "timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
-            counter.load(Ordering::SeqCst)
-        )
-    });
+    .unwrap_or_else(|_| panic!("timed out waiting for {description}"));
 }
 
 async fn wait_for_file(path: &Path, description: &str) {
@@ -211,8 +189,7 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
     request: DecryptRequest,
     expected: Vec<u8>,
     completed: Arc<AtomicU64>,
-    allow_failover_errors: Arc<AtomicBool>,
-    failure: Arc<Mutex<Option<String>>>,
+    failed: Arc<AtomicBool>,
     stop: CancellationToken,
 ) {
     while !stop.is_cancelled() {
@@ -220,18 +197,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
             Ok(response) if response.plaintext == expected => {
                 completed.fetch_add(1, Ordering::SeqCst);
             }
-            Ok(_) => {
-                *failure.lock().expect("decrypt failure lock poisoned") =
-                    Some("decrypt returned unexpected plaintext".to_string());
-                return;
-            }
-            Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
-                if allow_failover_errors.load(Ordering::SeqCst) =>
-            {
-                tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
-            }
-            Err(error) => {
-                *failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
+            Ok(_) | Err(_) => {
+                failed.store(true, Ordering::SeqCst);
                 return;
             }
         }
@@ -329,9 +296,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
     );
 
     let stop = CancellationToken::new();
-    let allow_failover_errors = Arc::new(AtomicBool::new(false));
-    let kv2_failure = Arc::new(Mutex::new(None));
-    let transit_failure = Arc::new(Mutex::new(None));
+    let failed = Arc::new(AtomicBool::new(false));
     let kv2_completed = Arc::new(AtomicU64::new(0));
     let transit_completed = Arc::new(AtomicU64::new(0));
     let kv2_worker = tokio::spawn(decrypt_loop(
@@ -339,8 +304,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
         kv2_request,
         kv2_data_key.plaintext_key,
         Arc::clone(&kv2_completed),
-        Arc::clone(&allow_failover_errors),
-        Arc::clone(&kv2_failure),
+        Arc::clone(&failed),
         stop.clone(),
     ));
     let transit_worker = tokio::spawn(decrypt_loop(
@@ -348,21 +312,12 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
         transit_request,
         transit_data_key.plaintext_key,
         Arc::clone(&transit_completed),
-        Arc::clone(&allow_failover_errors),
-        Arc::clone(&transit_failure),
+        Arc::clone(&failed),
         stop.clone(),
     ));
 
-    wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
-    wait_for_count(
-        &transit_completed,
-        &transit_failure,
-        2,
-        "two healthy Transit decrypts",
-        HEALTHY_PROGRESS_TIMEOUT,
-    )
-    .await;
-    allow_failover_errors.store(true, Ordering::SeqCst);
+    wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
+    wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
     std::fs::write(&marker, b"ready").expect("publish failover readiness marker");
 
     wait_for_file(&elected, "the replacement Vault leader").await;
@@ -371,39 +326,18 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
 
     let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
     let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
-    wait_for_count(
-        &kv2_completed,
-        &kv2_failure,
-        kv2_after_election,
-        "post-failover KV2 decrypts",
-        POST_FAILOVER_PROGRESS_TIMEOUT,
-    )
-    .await;
-    wait_for_count(
-        &transit_completed,
-        &transit_failure,
-        transit_after_election,
-        "post-failover Transit decrypts",
-        POST_FAILOVER_PROGRESS_TIMEOUT,
-    )
-    .await;
+    wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
+    wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
 
     stop.cancel();
     kv2_worker.await.expect("KV2 decrypt worker must join");
     transit_worker.await.expect("Transit decrypt worker must join");
-    assert!(
-        kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
-        "no KV2 decrypt may fail or return different plaintext"
-    );
-    assert!(
-        transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
-        "no Transit decrypt may fail or return different plaintext"
-    );
+    assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
 }
 
 #[test]
 #[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
-fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
+fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
     let recorder = DebuggingRecorder::new();
     let snapshotter = recorder.snapshotter();
     metrics::with_local_recorder(&recorder, || {
@@ -415,6 +349,11 @@ fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
     });
     let snapshot = snapshotter.snapshot().into_vec();
 
+    assert_eq!(
+        counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
+        0,
+        "a bounded leader election must not open the circuit"
+    );
     assert_eq!(
         counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
         0,

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

@@ -722,10 +722,6 @@ pub struct DeleteVersionsResponse {
     pub errors: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
-    /// Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
-    /// when present and fall back to strings for peers that predate this field. Code zero means success.
-    #[prost(message, repeated, tag = "4")]
-    pub item_errors: ::prost::alloc::vec::Vec<Error>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct ReadMultipleRequest {

crates/protos/src/node.proto

@@ -493,9 +493,6 @@ message DeleteVersionsResponse {
   bool success = 1;
   repeated string errors = 2;
   optional Error error = 3;
-  // Senders dual-write the legacy strings and typed entries. Receivers prefer typed entries
-  // when present and fall back to strings for peers that predate this field. Code zero means success.
-  repeated Error item_errors = 4;
 }
 
 message ReadMultipleRequest {

crates/scanner/src/data_usage_define.rs

@@ -125,34 +125,6 @@ pub(crate) async fn read_config_with_revision<S: ScannerObjectIO>(
     }
 }
 
-/// Read only the object revision without materializing its body.
-pub(crate) async fn read_config_revision<S: ScannerObjectIO>(store: Arc<S>, path: &str) -> StorageResult<DataUsageCacheRevision> {
-    match store
-        .get_object_reader(
-            RUSTFS_META_BUCKET,
-            path,
-            None,
-            HeaderMap::new(),
-            &ObjectOptions {
-                no_lock: true,
-                ..Default::default()
-            },
-        )
-        .await
-    {
-        Ok(reader) => reader
-            .object_info
-            .etag
-            .filter(|etag| !etag.is_empty())
-            .map(DataUsageCacheRevision::Etag)
-            .ok_or_else(|| StorageError::other(format!("scanner config object {path} has no ETag"))),
-        Err(Error::FileNotFound | Error::VolumeNotFound | Error::ObjectNotFound(_, _) | Error::BucketNotFound(_)) => {
-            Ok(DataUsageCacheRevision::Missing)
-        }
-        Err(err) => Err(err),
-    }
-}
-
 #[derive(Clone, Debug)]
 pub(crate) struct DataUsageCacheRevisions {
     main: DataUsageCacheRevision,
@@ -174,11 +146,6 @@ pub static LEGACY_DATA_USAGE_OBJ_NAME_PATH: LazyLock<String> =
 pub static DATA_USAGE_BLOOM_NAME_PATH: LazyLock<String> =
     LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{DATA_USAGE_BLOOM_NAME}"));
 
-/// Durable companion object for a cycle-state object which cannot be decoded.
-/// The primary object is deliberately never replaced or deleted by recovery.
-pub static DATA_USAGE_BLOOM_RECOVERY_PATH: LazyLock<String> =
-    LazyLock::new(|| format!("{}.recovery-required.json", DATA_USAGE_BLOOM_NAME_PATH.as_str()));
-
 pub static BACKGROUND_HEAL_INFO_PATH: LazyLock<String> =
     LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}.background-heal.json"));
 

crates/scanner/src/data_usage_define/persistence.rs

@@ -74,7 +74,7 @@ impl DataUsageCache {
         let loaded = Self::load_cache(store.clone(), name).await?;
         let backup = match loaded.backup_revision {
             Some(revision) => Some(revision),
-            None => match read_config_revision(store, &backup_path).await {
+            None => match Self::revision_for_path(store, &backup_path).await {
                 Ok(revision) => Some(revision),
                 Err(err) => {
                     counter!(METRIC_CACHE_BACKUP_REVISION_FAILURE_TOTAL).increment(1);
@@ -336,6 +336,33 @@ impl DataUsageCache {
         }
     }
 
+    async fn revision_for_path<S: ScannerObjectIO>(store: Arc<S>, path: &str) -> StorageResult<DataUsageCacheRevision> {
+        match store
+            .get_object_reader(
+                RUSTFS_META_BUCKET,
+                path,
+                None,
+                HeaderMap::new(),
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+        {
+            Ok(reader) => reader
+                .object_info
+                .etag
+                .filter(|etag| !etag.is_empty())
+                .map(DataUsageCacheRevision::Etag)
+                .ok_or_else(|| StorageError::other(format!("scanner cache object {path} has no ETag"))),
+            Err(Error::FileNotFound | Error::VolumeNotFound | Error::ObjectNotFound(_, _) | Error::BucketNotFound(_)) => {
+                Ok(DataUsageCacheRevision::Missing)
+            }
+            Err(err) => Err(err),
+        }
+    }
+
     pub(super) fn cache_save_timeout() -> Duration {
         crate::runtime_config::scanner_cache_save_timeout()
     }

crates/scanner/src/lib.rs

@@ -75,10 +75,7 @@ pub use remote_scanner::{
 };
 pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_status, validate_scanner_runtime_config};
 pub use rustfs_common::last_minute;
-pub use scanner::{
-    ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerCycleScheduleStatus, init_data_scanner,
-    reset_scanner_cycle_recovery, scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_topology_digest,
-};
+pub use scanner::{ScannerCycleScheduleStatus, init_data_scanner, scanner_cycle_schedule_status, scanner_topology_digest};
 pub use scanner_io::{
     ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket,
     record_dirty_usage_bucket, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_state,

crates/scanner/src/scanner.rs

@@ -20,7 +20,7 @@ use std::sync::{Arc, LazyLock, RwLock};
 
 use crate::data_usage_define::{
     BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH, DATA_USAGE_OBSERVED_OBJ_NAME_PATH,
-    DataUsageCache, DataUsageCacheRevision, LEGACY_DATA_USAGE_OBJ_NAME_PATH, read_config_revision, read_config_with_revision,
+    DataUsageCache, DataUsageCacheRevision, LEGACY_DATA_USAGE_OBJ_NAME_PATH, read_config_with_revision,
 };
 use crate::runtime_config::{
     ScannerRuntimeConfig, ScannerRuntimeConfigSource, refresh_scanner_runtime_config_from_global, scanner_bitrot_cycle,
@@ -54,7 +54,9 @@ use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELA
 use rustfs_data_usage::observed_data_usage_is_newer;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest as _, Sha256};
-use tokio::sync::{Notify, mpsc};
+#[cfg(test)]
+use tokio::sync::Notify;
+use tokio::sync::mpsc;
 use tokio::time::{Duration, Instant};
 use tokio_util::sync::CancellationToken;
 use tokio_util::task::AbortOnDropHandle;
@@ -102,13 +104,6 @@ const CLEAN_IDLE_BACKOFF_FACTOR: u32 = 2;
 /// unavailable peer cannot drive a tight retry loop.
 const SCANNER_RETRY_BASE_INTERVAL: Duration = Duration::from_secs(5);
 const SCANNER_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30 * 60);
-/// A transient backend outage remains self-healing after the short retry
-/// budget is exhausted, but the probe is intentionally sparse until storage
-/// recovers or an operator reset wakes the scanner.
-const SCANNER_CYCLE_RECOVERY_PAUSED_INTERVAL: Duration = Duration::from_secs(5 * 60);
-/// Permanent recovery states still get a sparse status probe so a reset that
-/// races the wait registration cannot leave the scanner asleep forever.
-const SCANNER_CYCLE_RECOVERY_BLOCKED_PROBE_INTERVAL: Duration = Duration::from_secs(5 * 60);
 const SCANNER_LEADER_LOCK_POLL_INTERVAL: Duration = Duration::from_secs(1);
 #[cfg(not(test))]
 const SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
@@ -130,12 +125,6 @@ type ScannerCycleStatePersistTestHook = (u64, Arc<Notify>);
 static SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK: LazyLock<StdMutex<Option<ScannerCycleStatePersistTestHook>>> =
     LazyLock::new(|| StdMutex::new(None));
 
-static SCANNER_CYCLE_RECOVERY_WAKE: LazyLock<Notify> = LazyLock::new(Notify::new);
-
-pub(super) fn notify_scanner_cycle_recovery_wake() {
-    SCANNER_CYCLE_RECOVERY_WAKE.notify_one();
-}
-
 #[cfg(test)]
 struct ScannerCycleStatePersistTestHookGuard;
 
@@ -587,21 +576,19 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
             tokio::time::sleep(sleep_time).await;
         }
 
-        let mut transient_backoff = ScannerRetryBackoff::default();
-        let mut recovery_retry_count = 0_u32;
         loop {
             if ctx_clone.is_cancelled() {
                 break;
             }
 
-            let run_result = run_data_scanner_with_maintenance_state(
+            if let Err(e) = run_data_scanner_with_maintenance_state(
                 ctx_clone.clone(),
                 storeapi_clone.clone(),
                 startup_features,
                 startup_maintenance_generation,
             )
-            .await;
-            if let Err(e) = &run_result {
+            .await
+            {
                 error!(
                     target: "rustfs::scanner",
                     event = EVENT_SCANNER_CYCLE_STATE,
@@ -612,52 +599,11 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
                     "Scanner runtime iteration failed"
                 );
             }
-            let recovery_status = scanner_cycle_recovery_status();
-            if recovery_status.retryable {
-                recovery_retry_count = recovery_retry_count.saturating_add(1);
-                let _ = record_scanner_cycle_recovery_retry(recovery_retry_count);
-            } else {
-                recovery_retry_count = 0;
-            }
-
-            let recovery_status = scanner_cycle_recovery_status();
-            if recovery_status.state == "paused" {
-                transient_backoff.record_retryable_cycle(false);
-                tokio::select! {
-                    _ = ctx_clone.cancelled() => break,
-                    _ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {},
-                    _ = tokio::time::sleep(SCANNER_CYCLE_RECOVERY_PAUSED_INTERVAL) => {},
-                }
-                recovery_retry_count = 0;
-                continue;
-            }
-            if !recovery_status.retryable
-                && matches!(recovery_status.state.as_str(), "blocked" | "recovery-required" | "cleanup-pending")
-            {
-                transient_backoff.record_retryable_cycle(false);
-                tokio::select! {
-                    _ = ctx_clone.cancelled() => break,
-                    _ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {},
-                    _ = tokio::time::sleep(SCANNER_CYCLE_RECOVERY_BLOCKED_PROBE_INTERVAL) => {},
-                }
-                continue;
-            }
-
-            let retry_delay = if recovery_status.retryable || run_result.is_err() {
-                transient_backoff.record_retryable_cycle(true);
-                transient_backoff
-                    .retry_interval(scanner_cycle_interval())
-                    .unwrap_or(SCANNER_RETRY_BASE_INTERVAL)
-            } else {
-                transient_backoff.record_retryable_cycle(false);
-                randomized_cycle_delay()
-            };
             // Backoff before retrying after lock contention or scanner-level failures.
             // Keep this cancellation-aware so shutdown is not delayed by backoff sleep.
             tokio::select! {
                 _ = ctx_clone.cancelled() => break,
-                _ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {},
-                _ = tokio::time::sleep(retry_delay) => {}
+                _ = tokio::time::sleep(randomized_cycle_delay()) => {}
             }
         }
     });
@@ -1660,22 +1606,40 @@ async fn run_data_scanner_with_maintenance_state(
         observe_scanner_activity(&storeapi, distributed, &mut scanner_activity_seen).await;
     }
 
-    let (mut cycle_info, mut leader_epoch, mut cycle_revision) =
-        match load_scanner_cycle_state_for_startup(storeapi.clone()).await {
-            ScannerCycleStateStartup::Ready {
-                cycle,
-                leader_epoch,
-                revision,
-            } => (cycle, leader_epoch, revision),
-            ScannerCycleStateStartup::Blocked => {
-                global_metrics().set_cycle(None).await;
-                return Ok(());
-            }
-            ScannerCycleStateStartup::Transient(err) => {
-                global_metrics().set_cycle(None).await;
-                return Err(err);
-            }
-        };
+    let (buf, mut cycle_revision) = match read_config_with_revision(storeapi.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()).await {
+        Ok((buf, revision)) => (buf.unwrap_or_default(), revision),
+        Err(err) => {
+            error!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                path = %&*DATA_USAGE_BLOOM_NAME_PATH,
+                state = "revision_load_failed",
+                error = %err,
+                "Scanner cycle state revision load failed"
+            );
+            global_metrics().set_cycle(None).await;
+            return Ok(());
+        }
+    };
+    let (mut cycle_info, mut leader_epoch) = match decode_scanner_cycle_state_for_startup(&buf) {
+        Ok(state) => state,
+        Err(err) => {
+            error!(
+                target: "rustfs::scanner",
+                event = EVENT_SCANNER_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_RUNTIME,
+                path = %&*DATA_USAGE_BLOOM_NAME_PATH,
+                state = "cycle_decode_failed",
+                error = %err,
+                "Scanner stopped because persisted cycle state is invalid"
+            );
+            global_metrics().set_cycle(None).await;
+            return Ok(());
+        }
+    };
     let usage_floor = match persisted_usage_floor(storeapi.clone()).await {
         Ok(floor) => floor,
         Err(err) => {
@@ -2255,12 +2219,7 @@ pub(crate) use activity::{
 pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance};
 #[cfg(test)]
 pub(crate) use cycle_state::encode_scanner_cycle_fence_for_test;
-pub use cycle_state::{
-    ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, reset_scanner_cycle_recovery, scanner_cycle_recovery_status,
-};
-pub(crate) use cycle_state::{
-    current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence, load_scanner_cycle_state_for_startup,
-};
+pub(crate) use cycle_state::{current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence};
 pub use heal_info::{BackgroundHealInfo, read_background_heal_info, save_background_heal_info};
 pub use usage_store::store_data_usage_in_backend;
 

crates/scanner/src/scanner/leadership.rs

@@ -196,7 +196,7 @@ pub(super) async fn claim_scanner_leadership(
         if ctx.is_cancelled() {
             return false;
         }
-        let Some(claimed_epoch) = persisted_epoch.checked_add(1).filter(|epoch| *epoch < u64::MAX) else {
+        let Some(claimed_epoch) = persisted_epoch.checked_add(1) else {
             error!(
                 target: "rustfs::scanner",
                 event = EVENT_SCANNER_PERSIST_STATE,

crates/scanner/src/scanner/tests.rs

@@ -15,12 +15,11 @@
 use super::*;
 use crate::EcstoreResult;
 use crate::{
-    DATA_USAGE_BLOOM_RECOVERY_PATH, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints,
-    ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions,
-    ScannerPutObjReader as PutObjReader, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
-    init_local_disks_with_instance_ctx,
+    Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerGetObjectReader as GetObjectReader,
+    ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader,
+    init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx,
 };
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::io::Cursor;
 use std::task::Poll;
 use temp_env::{with_var, with_var_unset};
@@ -118,15 +117,6 @@ async fn scanner_cycle_lock_fence_bounds_uncooperative_shutdown() {
     assert!(cycle_ctx.is_cancelled());
 }
 
-#[tokio::test]
-async fn scanner_cycle_recovery_wake_survives_wait_registration_race() {
-    notify_scanner_cycle_recovery_wake();
-
-    tokio::time::timeout(Duration::from_secs(1), SCANNER_CYCLE_RECOVERY_WAKE.notified())
-        .await
-        .expect("recovery wake should retain a permit until the waiter registers");
-}
-
 struct ScannerDefaultSpeedGuard;
 
 impl ScannerDefaultSpeedGuard {
@@ -161,7 +151,6 @@ impl Drop for ScannerDefaultCycleGuard {
 struct MemoryConfigStore {
     objects: Mutex<HashMap<String, Vec<u8>>>,
     revisions: Mutex<HashMap<String, u64>>,
-    non_regular_objects: Mutex<HashSet<String>>,
     fail_put_number: Mutex<HashMap<String, usize>>,
     object_not_found_put_number: Mutex<HashMap<String, usize>>,
     error_after_commit_put_number: Mutex<HashMap<String, usize>>,
@@ -202,16 +191,12 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore {
             .get(&key)
             .cloned()
             .ok_or(EcstoreError::FileNotFound)?;
-        let data_len = i64::try_from(data.len()).expect("memory test object length should fit in i64");
-        let revision = *self.revisions.lock().await.entry(key.clone()).or_insert(1);
-        let is_dir = self.non_regular_objects.lock().await.contains(&key);
+        let revision = *self.revisions.lock().await.entry(key).or_insert(1);
 
         Ok(GetObjectReader {
             stream: Box::new(Cursor::new(data)),
             object_info: ObjectInfo {
                 etag: Some(format!("memory-{revision}")),
-                size: data_len,
-                is_dir,
                 ..Default::default()
             },
             buffered_body: None,
@@ -812,10 +797,6 @@ fn scanner_cycle_state_decodes_legacy_and_fenced_formats() {
     let (fenced_cycle, fenced_epoch) = decode_scanner_cycle_state(&fenced).expect("fenced cycle state should decode");
     assert_eq!(fenced_cycle.next, 13);
     assert_eq!(fenced_epoch, 7);
-
-    let mut trailing = fenced;
-    trailing.push(0);
-    assert!(decode_scanner_cycle_state(&trailing).is_err());
 }
 
 #[test]
@@ -842,840 +823,6 @@ fn scanner_startup_fails_closed_on_nonempty_corrupt_cycle_state() {
     assert!(encode_scanner_cycle_state(&exhausted, 7).is_err());
 }
 
-#[tokio::test]
-async fn corrupt_cycle_state_is_quarantined_once() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
-    store.objects.lock().await.insert(state_key.clone(), vec![1]);
-    store.revisions.lock().await.insert(state_key.clone(), 7);
-
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store.clone()).await,
-        ScannerCycleStateStartup::Blocked
-    ));
-    let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
-    let marker_data = store
-        .objects
-        .lock()
-        .await
-        .get(&marker_key)
-        .cloned()
-        .expect("corrupt state must leave a durable recovery marker");
-    let marker: ScannerCycleRecoveryMarker = serde_json::from_slice(&marker_data).expect("marker should be valid JSON");
-    assert_eq!(marker.primary_revision, "memory-7");
-    assert_eq!(marker.path, DATA_USAGE_BLOOM_NAME_PATH.as_str());
-    assert_eq!(marker.quarantine_path, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
-    assert_eq!(marker.classification, "corrupt");
-
-    // A second startup sees the matching marker before consuming the poison body.
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store.clone()).await,
-        ScannerCycleStateStartup::Blocked
-    ));
-
-    // Replacing the primary object advances its revision; the stale marker must
-    // not quarantine the newer, valid state.
-    let cycle = CurrentCycle {
-        next: 9,
-        ..Default::default()
-    };
-    let encoded = encode_scanner_cycle_state(&cycle, 3).expect("valid state should encode");
-    store.objects.lock().await.insert(state_key.clone(), encoded);
-    store.revisions.lock().await.insert(state_key, 8);
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store).await,
-        ScannerCycleStateStartup::Ready {
-            cycle: CurrentCycle { next: 9, .. },
-            leader_epoch: 3,
-            ..
-        }
-    ));
-}
-
-#[tokio::test]
-async fn empty_cycle_state_object_is_quarantined_as_corrupt() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
-    store.objects.lock().await.insert(state_key.clone(), Vec::new());
-    store.revisions.lock().await.insert(state_key, 6);
-
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store).await,
-        ScannerCycleStateStartup::Blocked
-    ));
-    assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("corrupt"));
-    assert!(
-        scanner_cycle_recovery_status()
-            .reason
-            .as_deref()
-            .is_some_and(|reason| reason.contains("empty"))
-    );
-}
-
-#[tokio::test]
-async fn future_cycle_state_schema_is_recovery_required() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
-    let mut future = 17_u64.to_le_bytes().to_vec();
-    future.extend_from_slice(b"RSCYC999");
-    future.extend_from_slice(&4_u64.to_le_bytes());
-    future.extend_from_slice(&[0x90]);
-    store.objects.lock().await.insert(state_key.clone(), future);
-    store.revisions.lock().await.insert(state_key, 13);
-
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store).await,
-        ScannerCycleStateStartup::Blocked
-    ));
-    assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("future_schema"));
-}
-
-#[tokio::test]
-async fn concurrent_leaders_cannot_quarantine_newer_cycle_state() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
-    store.objects.lock().await.insert(state_key.clone(), vec![1]);
-    store.revisions.lock().await.insert(state_key, 4);
-
-    let (first, second) = tokio::join!(
-        load_scanner_cycle_state_for_startup(store.clone()),
-        load_scanner_cycle_state_for_startup(store.clone()),
-    );
-    assert!(matches!(first, ScannerCycleStateStartup::Blocked));
-    assert!(matches!(second, ScannerCycleStateStartup::Blocked));
-
-    let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
-    let marker_data = store
-        .objects
-        .lock()
-        .await
-        .get(&marker_key)
-        .cloned()
-        .expect("one contender must publish the recovery marker");
-    let marker: ScannerCycleRecoveryMarker = serde_json::from_slice(&marker_data).expect("marker should decode");
-    assert_eq!(marker.primary_revision, "memory-4");
-}
-
-#[tokio::test]
-async fn cleanup_pending_marker_blocks_a_rewritten_primary_after_restart() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
-    let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
-    let encoded = encode_scanner_cycle_state(
-        &CurrentCycle {
-            next: 12,
-            ..Default::default()
-        },
-        8,
-    )
-    .expect("valid state should encode");
-    store.objects.lock().await.insert(state_key.clone(), encoded);
-    store.revisions.lock().await.insert(state_key, 22);
-    let marker = ScannerCycleRecoveryMarker {
-        schema_version: 1,
-        primary_revision: "memory-21".to_string(),
-        generation: 11,
-        leader_epoch: 7,
-        classification: "corrupt".to_string(),
-        first_detected_at_unix_secs: 1,
-        last_attempt_at_unix_secs: 2,
-        retry_count: 1,
-        reason: "reset in progress".to_string(),
-        path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
-        quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
-        state: "cleanup-pending".to_string(),
-    };
-    store
-        .objects
-        .lock()
-        .await
-        .insert(marker_key.clone(), serde_json::to_vec(&marker).expect("marker should encode"));
-    store.revisions.lock().await.insert(marker_key, 3);
-
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store).await,
-        ScannerCycleStateStartup::Blocked
-    ));
-    assert_eq!(scanner_cycle_recovery_status().state, "cleanup-pending");
-}
-
-#[test]
-fn full_rescan_reset_accepts_unknown_marker_fields_without_trusting_cursor() {
-    let marker = br#"{
-        "schema_version": 99,
-        "primary_revision": "memory-7",
-        "generation": 9000,
-        "leader_epoch": 9000,
-        "classification": "new-future-classification",
-        "first_detected_at_unix_secs": 1,
-        "last_attempt_at_unix_secs": 2,
-        "retry_count": 9,
-        "reason": "future marker",
-        "path": "buckets/.bloomcycle.bin",
-        "quarantine_path": "buckets/.bloomcycle.bin.recovery-required.json",
-        "future_field": {"cursor": "untrusted"}
-    }"#;
-    let decoded =
-        super::cycle_state::decode_recovery_marker_for_reset(marker, &DataUsageCacheRevision::Etag("memory-3".to_string()))
-            .expect("full-rescan compatibility decoder should accept additive fields");
-    assert_eq!(decoded.primary_revision, "memory-7");
-    assert_eq!(decoded.classification, "future_schema");
-    assert_eq!(decoded.generation, 0);
-    assert_eq!(decoded.leader_epoch, 0);
-    assert_eq!(decoded.state, "blocked");
-
-    let malformed =
-        super::cycle_state::decode_recovery_marker_for_reset(b"{not-json", &DataUsageCacheRevision::Etag("memory-4".to_string()))
-            .expect("a full-rescan reset must recover even when the marker is malformed");
-    assert!(malformed.primary_revision.is_empty());
-    assert_eq!(malformed.classification, "future_schema");
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rebuilds_after_malformed_marker_without_trusting_cursor() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0xff, 0x00, 0x01])
-        .await
-        .expect("corrupt cycle state should be persisted");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), br#"{not-json"#.to_vec())
-        .await
-        .expect("malformed marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("full-rescan reset should recover malformed marker");
-
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("rebuilt cycle state should remain durable");
-    let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
-    assert_eq!(cycle.next, 0, "reset must use the verified usage floor, not marker cursor");
-    assert_eq!(leader_epoch, 1);
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_ignores_epoch_from_malformed_future_primary() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    let mut future_primary = vec![0; 24];
-    future_primary[8..16].copy_from_slice(b"RSCY9999");
-    future_primary[16..24].copy_from_slice(&u64::MAX.to_le_bytes());
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), future_primary)
-        .await
-        .expect("future cycle state should be persisted");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), br#"{not-json"#.to_vec())
-        .await
-        .expect("malformed marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("full-rescan reset should recover malformed future state");
-
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("rebuilt cycle state should remain durable");
-    let (_, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
-    assert_eq!(leader_epoch, 1, "invalid persisted bytes must not raise the recovery epoch");
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn ecstore_exact_recovery_marker_delete_honors_etag() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"marker-v1".to_vec())
-        .await
-        .expect("initial recovery marker should be persisted");
-    let (_, stale_revision) = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
-        .await
-        .expect("initial marker revision should load");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"marker-v2".to_vec())
-        .await
-        .expect("replacement recovery marker should be persisted");
-
-    let delete_result = store
-        .delete_config_object(
-            RUSTFS_META_BUCKET,
-            DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
-            ObjectOptions {
-                http_preconditions: Some(stale_revision.preconditions()),
-                ..Default::default()
-            },
-        )
-        .await;
-    assert!(matches!(delete_result, Err(EcstoreError::PreconditionFailed)));
-    assert_eq!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
-            .await
-            .expect("replacement marker should remain durable"),
-        b"marker-v2"
-    );
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rejects_corrupt_primary_under_stale_blocked_marker() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    let corrupt_primary = vec![0xff, 0x00, 0x01];
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), corrupt_primary.clone())
-        .await
-        .expect("corrupt cycle state should be persisted");
-    let (_, primary_revision) = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("primary revision should load");
-    let marker = ScannerCycleRecoveryMarker {
-        schema_version: 1,
-        primary_revision: "memory-stale".to_string(),
-        generation: 1,
-        leader_epoch: 1,
-        classification: "corrupt".to_string(),
-        first_detected_at_unix_secs: 1,
-        last_attempt_at_unix_secs: 2,
-        retry_count: 1,
-        reason: "blocked primary changed".to_string(),
-        path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
-        quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
-        state: "blocked".to_string(),
-    };
-    let marker_data = serde_json::to_vec(&marker).expect("blocked marker should encode");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), marker_data.clone())
-        .await
-        .expect("blocked marker should be persisted");
-
-    assert!(
-        reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-            .await
-            .is_err(),
-        "a strict marker must fail closed when its primary revision changed"
-    );
-    assert_eq!(
-        read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-            .await
-            .expect("primary should remain readable"),
-        corrupt_primary
-    );
-    assert_eq!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
-            .await
-            .expect("blocked marker should remain durable"),
-        marker_data
-    );
-    assert!(!matches!(primary_revision, DataUsageCacheRevision::Missing));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_preserves_valid_primary_when_marker_is_malformed() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    let primary = CurrentCycle {
-        next: 42,
-        ..Default::default()
-    };
-    let old_primary_data = encode_scanner_cycle_state(&primary, 7).expect("valid cycle state should encode");
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), old_primary_data.clone())
-        .await
-        .expect("valid cycle state should be persisted");
-    let (_, old_primary_revision) = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("primary state revision should load");
-    let old_usage = DataUsageInfo {
-        scanner_epoch: Some(7),
-        scanner_cycle: Some(41),
-        ..Default::default()
-    };
-    let old_usage_data = serde_json::to_vec(&old_usage).expect("usage snapshot should encode");
-    save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), old_usage_data.clone())
-        .await
-        .expect("usage snapshot should be persisted");
-    let (_, old_usage_revision) = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
-        .await
-        .expect("usage snapshot revision should load");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
-        .await
-        .expect("malformed marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("reset should clear a stale malformed marker");
-
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("valid primary should remain durable");
-    let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("primary cycle state should decode");
-    assert_eq!(cycle.next, 42, "reset must not regress an independently fenced primary");
-    assert_eq!(leader_epoch, 8, "reset must advance the preserved primary epoch");
-    let stale_primary_save = save_config_with_preconditions(
-        store.clone(),
-        DATA_USAGE_BLOOM_NAME_PATH.as_str(),
-        old_primary_data,
-        old_primary_revision.preconditions(),
-    )
-    .await;
-    assert!(matches!(stale_primary_save, Err(EcstoreError::PreconditionFailed)));
-    let usage = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
-        .await
-        .expect("usage epoch fence should remain durable");
-    assert_eq!(
-        serde_json::from_slice::<DataUsageInfo>(&usage)
-            .expect("fenced usage should decode")
-            .scanner_epoch,
-        Some(8)
-    );
-    let stale_save = save_config_with_preconditions(
-        store.clone(),
-        DATA_USAGE_OBJ_NAME_PATH.as_str(),
-        old_usage_data,
-        old_usage_revision.preconditions(),
-    )
-    .await;
-    assert!(matches!(stale_save, Err(EcstoreError::PreconditionFailed)));
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_resumes_cleanup_pending_preserved_primary() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    let completed_at = Utc::now();
-    let primary = CurrentCycle {
-        current: 3,
-        next: 42,
-        cycle_completed: vec![completed_at],
-        started: completed_at,
-    };
-    save_config(
-        store.clone(),
-        DATA_USAGE_BLOOM_NAME_PATH.as_str(),
-        encode_scanner_cycle_state(&primary, 7).expect("valid cycle state should encode"),
-    )
-    .await
-    .expect("valid cycle state should be persisted");
-    let usage = DataUsageInfo {
-        scanner_epoch: Some(7),
-        scanner_cycle: Some(41),
-        ..Default::default()
-    };
-    save_config(
-        store.clone(),
-        DATA_USAGE_OBJ_NAME_PATH.as_str(),
-        serde_json::to_vec(&usage).expect("usage snapshot should encode"),
-    )
-    .await
-    .expect("usage snapshot should be persisted");
-    let marker = ScannerCycleRecoveryMarker {
-        schema_version: 1,
-        primary_revision: "memory-old".to_string(),
-        generation: 41,
-        leader_epoch: 7,
-        classification: "corrupt".to_string(),
-        first_detected_at_unix_secs: 1,
-        last_attempt_at_unix_secs: 2,
-        retry_count: 1,
-        reason: "reset in progress".to_string(),
-        path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
-        quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
-        state: "cleanup-pending".to_string(),
-    };
-    save_config(
-        store.clone(),
-        DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
-        serde_json::to_vec(&marker).expect("marker should encode"),
-    )
-    .await
-    .expect("cleanup marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("reset should resume a cleanup-pending preserved primary");
-
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("preserved cycle state should remain durable");
-    let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("cycle state should decode");
-    assert_eq!(cycle.current, 3, "cleanup retry must preserve the in-progress cursor");
-    assert_eq!(cycle.next, 42);
-    assert_eq!(cycle.cycle_completed, vec![completed_at]);
-    assert_eq!(cycle.started, completed_at);
-    assert_eq!(leader_epoch, 8);
-    let usage = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
-        .await
-        .expect("usage epoch fence should remain durable");
-    assert_eq!(
-        serde_json::from_slice::<DataUsageInfo>(&usage)
-            .expect("usage should decode")
-            .scanner_epoch,
-        Some(8)
-    );
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rebuilds_oversized_regular_primary_with_malformed_marker() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0; 1024 * 1024 + 1])
-        .await
-        .expect("oversized cycle state should be persisted");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
-        .await
-        .expect("malformed marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("explicit full-rescan reset should replace an oversized regular primary");
-
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("rebuilt cycle state should remain durable");
-    let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
-    assert_eq!(cycle.next, 0);
-    assert_eq!(leader_epoch, 1);
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rebuilds_oversized_primary_after_cleanup_marker() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0; 1024 * 1024 + 1])
-        .await
-        .expect("oversized cycle state should be persisted");
-    let (_, primary_revision) = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("primary revision should load");
-    let marker = ScannerCycleRecoveryMarker {
-        schema_version: 1,
-        primary_revision: match primary_revision {
-            DataUsageCacheRevision::Etag(etag) => etag,
-            DataUsageCacheRevision::Missing => panic!("primary revision should be present"),
-        },
-        generation: 1,
-        leader_epoch: 1,
-        classification: "corrupt".to_string(),
-        first_detected_at_unix_secs: 1,
-        last_attempt_at_unix_secs: 2,
-        retry_count: 1,
-        reason: "reset in progress".to_string(),
-        path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
-        quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
-        state: "cleanup-pending".to_string(),
-    };
-    save_config(
-        store.clone(),
-        DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
-        serde_json::to_vec(&marker).expect("cleanup marker should encode"),
-    )
-    .await
-    .expect("cleanup marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("cleanup retry should rebuild an oversized primary");
-
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("rebuilt cycle state should remain durable");
-    let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
-    assert_eq!(cycle.next, 0);
-    assert_eq!(leader_epoch, 1);
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rebuilds_with_oversized_marker() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0xff, 0x00, 0x01])
-        .await
-        .expect("corrupt cycle state should be persisted");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), vec![b'x'; 64 * 1024 + 1])
-        .await
-        .expect("oversized recovery marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("full-rescan reset should recover an oversized marker");
-
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("rebuilt cycle state should remain durable");
-    let (_, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
-    assert_eq!(leader_epoch, 1);
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rebuilds_with_empty_marker() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0xff, 0x00, 0x01])
-        .await
-        .expect("corrupt cycle state should be persisted");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), Vec::new())
-        .await
-        .expect("empty recovery marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("full-rescan reset should recover an empty marker");
-
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("rebuilt cycle state should remain durable");
-    let (_, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
-    assert_eq!(leader_epoch, 1);
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_keeps_cleanup_marker_when_preserved_epoch_is_exhausted() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    let primary = CurrentCycle {
-        next: 42,
-        ..Default::default()
-    };
-    save_config(
-        store.clone(),
-        DATA_USAGE_BLOOM_NAME_PATH.as_str(),
-        encode_scanner_cycle_state(&primary, u64::MAX).expect("valid cycle state should encode"),
-    )
-    .await
-    .expect("valid cycle state should be persisted");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
-        .await
-        .expect("malformed marker should be persisted");
-
-    assert!(
-        reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-            .await
-            .is_err()
-    );
-
-    let marker = read_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
-        .await
-        .expect("cleanup marker should remain durable");
-    assert_eq!(
-        serde_json::from_slice::<ScannerCycleRecoveryMarker>(&marker)
-            .expect("cleanup marker should decode")
-            .state,
-        "cleanup-pending"
-    );
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store).await,
-        ScannerCycleStateStartup::Blocked
-    ));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rejects_preserved_epoch_that_would_be_terminal() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    let primary = CurrentCycle {
-        next: 42,
-        ..Default::default()
-    };
-    save_config(
-        store.clone(),
-        DATA_USAGE_BLOOM_NAME_PATH.as_str(),
-        encode_scanner_cycle_state(&primary, u64::MAX - 1).expect("valid cycle state should encode"),
-    )
-    .await
-    .expect("valid cycle state should be persisted");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
-        .await
-        .expect("malformed marker should be persisted");
-
-    assert!(
-        reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-            .await
-            .is_err(),
-        "reset must not persist the terminal leader epoch"
-    );
-
-    let marker = read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
-        .await
-        .expect("cleanup marker should remain durable");
-    assert_eq!(
-        serde_json::from_slice::<ScannerCycleRecoveryMarker>(&marker)
-            .expect("cleanup marker should decode")
-            .state,
-        "cleanup-pending"
-    );
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rejects_usage_floor_that_would_be_terminal() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0xff, 0x00, 0x01])
-        .await
-        .expect("corrupt cycle state should be persisted");
-    save_config(
-        store.clone(),
-        DATA_USAGE_OBJ_NAME_PATH.as_str(),
-        serde_json::to_vec(&DataUsageInfo {
-            scanner_epoch: Some(u64::MAX - 1),
-            ..Default::default()
-        })
-        .expect("usage floor should encode"),
-    )
-    .await
-    .expect("usage floor should be persisted");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
-        .await
-        .expect("malformed marker should be persisted");
-
-    assert!(
-        reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-            .await
-            .is_err(),
-        "reset must not persist the terminal leader epoch"
-    );
-    assert_eq!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
-            .await
-            .expect("recovery marker should remain durable"),
-        b"{not-json"
-    );
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rebuilds_empty_primary_with_malformed_marker() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), Vec::new())
-        .await
-        .expect("empty cycle state should be persisted");
-    save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
-        .await
-        .expect("malformed marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("explicit full-rescan reset should replace an empty primary");
-
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("rebuilt cycle state should remain durable");
-    let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
-    assert_eq!(cycle.next, 0);
-    assert_eq!(leader_epoch, 1);
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn full_rescan_reset_rebuilds_when_primary_cycle_state_is_missing() {
-    let (_temp_dir, store) = setup_scanner_cycle_store().await;
-    let marker = ScannerCycleRecoveryMarker {
-        schema_version: 1,
-        primary_revision: "memory-missing".to_string(),
-        generation: u64::MAX,
-        leader_epoch: u64::MAX,
-        classification: "corrupt".to_string(),
-        first_detected_at_unix_secs: 1,
-        last_attempt_at_unix_secs: 2,
-        retry_count: 0,
-        reason: "missing primary".to_string(),
-        path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
-        quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
-        state: "blocked".to_string(),
-    };
-    save_config(
-        store.clone(),
-        DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
-        serde_json::to_vec(&marker).expect("marker should encode"),
-    )
-    .await
-    .expect("marker should be persisted");
-
-    reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
-        .await
-        .expect("full-rescan reset should recreate missing primary");
-    let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
-        .await
-        .expect("missing primary should be rebuilt");
-    let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
-    assert_eq!(cycle.next, 0);
-    assert_eq!(leader_epoch, 1);
-    assert!(matches!(
-        read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
-        Err(EcstoreError::ConfigNotFound)
-    ));
-}
-
-#[tokio::test]
-async fn corrupt_cycle_state_rename_or_marker_failure_stays_recovery_required() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
-    let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
-    store.objects.lock().await.insert(state_key.clone(), vec![1]);
-    store.revisions.lock().await.insert(state_key, 9);
-    store.fail_put_number.lock().await.insert(marker_key, 1);
-
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store.clone()).await,
-        ScannerCycleStateStartup::Transient(_)
-    ));
-    let status = scanner_cycle_recovery_status();
-    assert_eq!(status.state, "recovery-required");
-    assert!(status.retryable);
-    assert!(
-        store
-            .objects
-            .lock()
-            .await
-            .contains_key(&memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()))
-    );
-}
-
-#[tokio::test]
-async fn oversized_or_symlinked_cycle_state_is_rejected() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
-    store.objects.lock().await.insert(key.clone(), vec![0; 1024 * 1024 + 1]);
-    store.revisions.lock().await.insert(key.clone(), 11);
-
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store.clone()).await,
-        ScannerCycleStateStartup::Blocked
-    ));
-    assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("corrupt"));
-    assert!(
-        scanner_cycle_recovery_status()
-            .reason
-            .as_deref()
-            .is_some_and(|reason| reason.contains("oversized"))
-    );
-
-    let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
-    store.objects.lock().await.remove(&marker_key);
-    store.objects.lock().await.insert(key.clone(), vec![1]);
-    store.revisions.lock().await.insert(key.clone(), 12);
-    store.non_regular_objects.lock().await.insert(key);
-    // The object contract exposes a non-regular object as `is_dir`; local
-    // backends reject symlink/reparse entries before they become an object.
-    assert!(matches!(
-        load_scanner_cycle_state_for_startup(store).await,
-        ScannerCycleStateStartup::Blocked
-    ));
-}
-
 #[tokio::test]
 async fn scanner_startup_uses_primary_and_backup_usage_floor() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -1708,31 +855,6 @@ async fn scanner_startup_uses_primary_and_backup_usage_floor() {
     assert_eq!(epoch, 11);
 }
 
-#[tokio::test]
-async fn scanner_usage_floor_ignores_older_backup_after_primary_epoch_fence() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
-    for (path, epoch, cycle) in [(DATA_USAGE_OBJ_NAME_PATH.as_str(), 8, 100), (backup_path.as_str(), 7, 10_000)] {
-        store.objects.lock().await.insert(
-            memory_config_key(RUSTFS_META_BUCKET, path),
-            serde_json::to_vec(&DataUsageInfo {
-                scanner_epoch: Some(epoch),
-                scanner_cycle: Some(cycle),
-                ..Default::default()
-            })
-            .expect("usage snapshot should encode"),
-        );
-    }
-
-    assert_eq!(
-        persisted_usage_floor(store).await.expect("usage floor should load"),
-        PersistedUsageFloor {
-            next_cycle: 101,
-            leader_epoch: 8,
-        }
-    );
-}
-
 #[test]
 fn scanner_startup_treats_incomplete_usage_snapshot_as_cold() {
     let mut legacy = complete_usage_with_bucket_count(Some(std::time::SystemTime::now()), 1);
@@ -1865,15 +987,6 @@ async fn scanner_usage_floor_fails_closed_on_corrupt_or_exhausted_usage_state()
 
     assert!(persisted_usage_floor(store.clone()).await.is_err());
 
-    store.objects.lock().await.insert(
-        memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
-        br#"{}"#.to_vec(),
-    );
-    assert!(
-        persisted_usage_floor(store.clone()).await.is_err(),
-        "a structurally incomplete usage snapshot must not be treated as an empty floor"
-    );
-
     store.objects.lock().await.insert(
         memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
         serde_json::to_vec(&DataUsageInfo {
@@ -2131,22 +1244,6 @@ async fn test_leadership_claim_preserves_usage_epoch_floor_across_old_epoch_conf
     assert_eq!(store.put_counts.lock().await.get(&key), Some(&3));
 }
 
-#[tokio::test]
-async fn test_leadership_claim_rejects_terminal_epoch() {
-    let store = Arc::new(MemoryConfigStore::default());
-    let ctx = CancellationToken::new();
-    let mut revision = DataUsageCacheRevision::Missing;
-    let mut cycle = CurrentCycle {
-        next: 12,
-        ..Default::default()
-    };
-    let mut persisted_epoch = u64::MAX - 1;
-
-    assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch).await);
-    assert_eq!(persisted_epoch, u64::MAX - 1);
-    assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
-}
-
 #[tokio::test]
 async fn test_leadership_claim_confirms_commit_after_returned_error() {
     let store = Arc::new(MemoryConfigStore::default());
@@ -3878,24 +2975,6 @@ fn superseded_retry_backoff_grows_from_the_default_cycle() {
     }
 }
 
-#[tokio::test(start_paused = true)]
-async fn corrupt_cycle_state_backoff_uses_virtual_clock() {
-    let mut backoff = ScannerRetryBackoff::default();
-    backoff.record_retryable_cycle(true);
-    let first_delay = backoff
-        .retry_interval(Duration::from_secs(60))
-        .expect("the first recovery retry should be scheduled");
-    assert_eq!(first_delay, Duration::from_secs(5));
-
-    let deadline = Instant::now() + first_delay;
-    assert!(Instant::now() < deadline);
-    tokio::time::advance(first_delay).await;
-    assert!(Instant::now() >= deadline);
-
-    backoff.record_retryable_cycle(true);
-    assert_eq!(backoff.retry_interval(Duration::from_secs(60)), Some(Duration::from_secs(10)));
-}
-
 #[test]
 fn scanner_cycle_wait_plan_drives_growth_resets_and_bitrot_cap() {
     let runtime_config = ScannerRuntimeConfig {

rustfs/src/admin/handlers/mod.rs

@@ -126,7 +126,6 @@ mod tests {
         let _list_remote_target_handler = replication::ListRemoteTargetHandler {};
         let _remove_remote_target_handler = replication::RemoveRemoteTargetHandler {};
         let _scanner_status_handler = scanner::ScannerStatusHandler {};
-        let _scanner_cycle_state_reset_handler = scanner::ScannerCycleStateResetHandler {};
         let _ilm_expiry_status_handler = scanner::IlmExpiryStatusHandler {};
         let _manual_transition_handler = ilm_transition::ManualTransitionRunHandler {};
         let _manual_transition_status_handler = ilm_transition::ManualTransitionJobStatusHandler {};

rustfs/src/admin/handlers/scanner.rs

@@ -13,11 +13,8 @@
 // limitations under the License.
 
 use crate::admin::auth::authorize_admin_request;
-use crate::admin::handlers::supervise_admin_mutation;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
-use crate::admin::runtime_sources::{
-    app_context_from_req, current_object_store_handle_for_context, current_scanner_metrics_report,
-};
+use crate::admin::runtime_sources::current_scanner_metrics_report;
 use crate::module_switches::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
 use crate::server::ADMIN_PREFIX;
 use chrono::Utc;
@@ -25,13 +22,11 @@ use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};
 use matchit::Params;
 use rustfs_common::metrics::{ScannerLifecycleExpirySnapshot, ScannerMaintenanceControlSnapshot, ScannerMetricsReport};
-use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
 use rustfs_credentials::Credentials;
 use rustfs_policy::policy::action::{Action, AdminAction};
 use s3s::header::CONTENT_TYPE;
 use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
-use serde::{Deserialize, Serialize};
-use tokio_util::sync::CancellationToken;
+use serde::Serialize;
 
 const JSON_CONTENT_TYPE: &str = "application/json";
 
@@ -43,13 +38,6 @@ struct ScannerStatusResponse {
     metrics: ScannerMetricsReport,
     cycle_schedule: rustfs_scanner::ScannerCycleScheduleStatus,
     runtime_config: rustfs_scanner::runtime_config::ScannerRuntimeConfigStatus,
-    cycle_recovery: rustfs_scanner::ScannerCycleRecoveryStatus,
-}
-
-#[derive(Debug, Deserialize)]
-#[serde(deny_unknown_fields)]
-struct ScannerCycleResetRequest {
-    mode: String,
 }
 
 #[derive(Debug, Serialize)]
@@ -129,7 +117,6 @@ fn scanner_status_response(
         metrics,
         cycle_schedule,
         runtime_config,
-        cycle_recovery: rustfs_scanner::scanner::scanner_cycle_recovery_status(),
     }
 }
 
@@ -157,11 +144,6 @@ pub fn register_scanner_route(r: &mut S3Router<AdminOperation>) -> std::io::Resu
         format!("{ADMIN_PREFIX}/v3/scanner/status").as_str(),
         AdminOperation(&ScannerStatusHandler {}),
     )?;
-    r.insert(
-        Method::POST,
-        format!("{ADMIN_PREFIX}/v3/scanner/cycle-state/reset").as_str(),
-        AdminOperation(&ScannerCycleStateResetHandler {}),
-    )?;
     r.insert(
         Method::GET,
         format!("{ADMIN_PREFIX}/v3/ilm/expiry/status").as_str(),
@@ -181,13 +163,6 @@ async fn validate_scanner_status_request(req: &S3Request<Body>) -> S3Result<Cred
     authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await
 }
 
-async fn validate_scanner_reset_request(req: &S3Request<Body>) -> S3Result<Credentials> {
-    if req.credentials.is_none() {
-        return Err(s3_error!(InvalidRequest, "missing credentials"));
-    }
-    authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ConfigUpdateAdminAction)]).await
-}
-
 fn json_response(body: Vec<u8>) -> S3Result<S3Response<(StatusCode, Body)>> {
     let mut headers = HeaderMap::new();
     let content_type = HeaderValue::from_str(JSON_CONTENT_TYPE)
@@ -217,37 +192,6 @@ impl Operation for ScannerStatusHandler {
 
 pub struct IlmExpiryStatusHandler {}
 
-pub struct ScannerCycleStateResetHandler {}
-
-#[async_trait::async_trait]
-impl Operation for ScannerCycleStateResetHandler {
-    async fn call(&self, mut req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        let _cred = validate_scanner_reset_request(&req).await?;
-        let body = req
-            .input
-            .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
-            .await
-            .map_err(|err| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid reset request body: {err}")))?;
-        let reset = serde_json::from_slice::<ScannerCycleResetRequest>(&body)
-            .map_err(|err| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid reset request body: {err}")))?;
-        if reset.mode != "full-rescan" {
-            return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "reset mode must be full-rescan"));
-        }
-        let context = app_context_from_req(&req)
-            .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "storage layer not initialized"))?;
-        let store = current_object_store_handle_for_context(Some(context.as_ref()))
-            .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "storage layer not initialized"))?;
-        supervise_admin_mutation("scanner cycle state reset", async move {
-            rustfs_scanner::scanner::reset_scanner_cycle_recovery(CancellationToken::new(), store)
-                .await
-                .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, err.to_string()))?;
-            Ok::<_, S3Error>(())
-        })
-        .await?;
-        json_response(br#"{"status":"reset","mode":"full-rescan"}"#.to_vec())
-    }
-}
-
 #[async_trait::async_trait]
 impl Operation for IlmExpiryStatusHandler {
     async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -293,38 +237,6 @@ mod tests {
         assert_eq!(err.message(), Some("missing credentials"));
     }
 
-    #[tokio::test]
-    async fn scanner_reset_gate_rejects_missing_credentials() {
-        let req = S3Request {
-            input: Body::from(String::new()),
-            method: Method::POST,
-            uri: http::Uri::from_static("/rustfs/admin/v3/scanner/cycle-state/reset"),
-            headers: HeaderMap::new(),
-            extensions: http::Extensions::new(),
-            credentials: None,
-            region: None,
-            service: None,
-            trailing_headers: None,
-        };
-
-        let err = validate_scanner_reset_request(&req)
-            .await
-            .expect_err("a reset request without credentials must be rejected");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
-        assert_eq!(err.message(), Some("missing credentials"));
-    }
-
-    #[test]
-    fn admin_reset_requires_full_rescan_or_verified_cursor() {
-        let full_rescan: ScannerCycleResetRequest =
-            serde_json::from_str(r#"{"mode":"full-rescan"}"#).expect("full rescan must be accepted");
-        assert_eq!(full_rescan.mode, "full-rescan");
-        let cursor: ScannerCycleResetRequest =
-            serde_json::from_str(r#"{"mode":"cursor"}"#).expect("mode validation belongs to the handler");
-        assert_ne!(cursor.mode, "full-rescan");
-        assert!(serde_json::from_str::<ScannerCycleResetRequest>(r#"{"mode":"full-rescan","cursor":"untrusted"}"#).is_err());
-    }
-
     #[test]
     fn scanner_disabled_reason_reports_startup_env_key() {
         assert_eq!(scanner_disabled_reason(true), None);
@@ -392,11 +304,6 @@ mod tests {
         assert_eq!(encoded["cycle_schedule"]["effective_interval_seconds"], 0);
         assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_enabled"], false);
         assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_multiplier"], 1);
-        assert_eq!(encoded["cycle_recovery"]["state"], "healthy");
-        assert_eq!(
-            encoded["cycle_recovery"]["quarantine_path"],
-            rustfs_scanner::DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()
-        );
     }
 
     #[test]

rustfs/src/admin/route_policy.rs

@@ -428,12 +428,6 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
     admin(HttpMethod::Get, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High),
     admin(HttpMethod::Put, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High),
     admin(HttpMethod::Get, "/rustfs/admin/v3/scanner/status", SERVER_INFO, RouteRiskLevel::Sensitive),
-    admin(
-        HttpMethod::Post,
-        "/rustfs/admin/v3/scanner/cycle-state/reset",
-        CONFIG_UPDATE,
-        RouteRiskLevel::High,
-    ),
     admin(
         HttpMethod::Get,
         "/rustfs/admin/v3/ilm/expiry/status",
@@ -2026,12 +2020,6 @@ mod tests {
         assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/expiry/status", SET_TIER);
     }
 
-    #[test]
-    fn route_policy_requires_config_update_for_scanner_cycle_reset() {
-        assert_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/cycle-state/reset", CONFIG_UPDATE);
-        assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/cycle-state/reset", SERVER_INFO);
-    }
-
     #[test]
     fn route_policy_uses_tier_actions_for_transition_routes() {
         assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SET_TIER);

rustfs/src/admin/route_registration_test.rs

@@ -243,7 +243,6 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         admin_route(Method::GET, "/v3/config"),
         admin_route(Method::PUT, "/v3/config"),
         admin_route(Method::GET, "/v3/scanner/status"),
-        admin_route(Method::POST, "/v3/scanner/cycle-state/reset"),
         admin_route(Method::GET, "/v3/audit/target/list"),
         admin_route_sample(
             Method::PUT,
@@ -880,7 +879,6 @@ fn test_register_routes_cover_representative_admin_paths() {
     assert_route(&router, Method::GET, &admin_path("/v3/config"));
     assert_route(&router, Method::PUT, &admin_path("/v3/config"));
     assert_route(&router, Method::GET, &admin_path("/v3/scanner/status"));
-    assert_route(&router, Method::POST, &admin_path("/v3/scanner/cycle-state/reset"));
     assert_route(&router, Method::GET, &admin_path("/v3/ilm/expiry/status"));
     assert_route(&router, Method::POST, &admin_path("/v3/ilm/transition/run"));
     assert_route(
@@ -1369,7 +1367,6 @@ fn test_admin_alias_paths_match_existing_admin_routes() {
         (Method::GET, compat_admin_alias_path("/v3/config")),
         (Method::PUT, compat_admin_alias_path("/v3/config")),
         (Method::GET, compat_admin_alias_path("/v3/scanner/status")),
-        (Method::POST, compat_admin_alias_path("/v3/scanner/cycle-state/reset")),
         (Method::GET, compat_admin_alias_path("/v3/ilm/expiry/status")),
     ] {
         assert!(

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

@@ -146,29 +146,6 @@ fn encode_file_info_msgpack(value: &FileInfo) -> std::result::Result<Vec<u8>, Di
     encode_msgpack_with_capacity(value, "FileInfo", FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT)
 }
 
-fn encode_delete_versions_errors(disk_errors: Vec<Option<DiskError>>) -> (Vec<String>, Vec<Error>) {
-    let mut errors = Vec::with_capacity(disk_errors.len());
-    let mut item_errors = Vec::with_capacity(disk_errors.len());
-    for error in disk_errors {
-        match error {
-            Some(error) => {
-                let code = match &error {
-                    DiskError::Io(source) if source.kind() == std::io::ErrorKind::NotFound => DiskError::FileNotFound.to_u32(),
-                    _ => error.to_u32(),
-                };
-                let error_info = error.to_string();
-                errors.push(error_info.clone());
-                item_errors.push(Error { code, error_info });
-            }
-            None => {
-                errors.push(String::new());
-                item_errors.push(Error::default());
-            }
-        }
-    }
-    (errors, item_errors)
-}
-
 fn encode_msgpack_named<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, DiskError> {
     let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map();
     value
@@ -575,7 +552,6 @@ impl NodeService {
                             success: false,
                             errors: Vec::new(),
                             error: Some(DiskError::other(format!("decode FileInfoVersions failed: {err}")).into()),
-                            item_errors: Vec::new(),
                         }));
                     }
                 };
@@ -587,26 +563,30 @@ impl NodeService {
                         success: false,
                         errors: Vec::new(),
                         error: Some(DiskError::other(format!("decode DeleteOptions failed: {err}")).into()),
-                        item_errors: Vec::new(),
                     }));
                 }
             };
 
-            let (errors, item_errors) =
-                encode_delete_versions_errors(disk.delete_versions(&request.volume, versions, opts).await);
+            let errors = disk
+                .delete_versions(&request.volume, versions, opts)
+                .await
+                .into_iter()
+                .map(|error| match error {
+                    Some(e) => e.to_string(),
+                    None => "".to_string(),
+                })
+                .collect();
 
             Ok(Response::new(DeleteVersionsResponse {
                 success: true,
                 errors,
                 error: None,
-                item_errors,
             }))
         } else {
             Ok(Response::new(DeleteVersionsResponse {
                 success: false,
                 errors: Vec::new(),
                 error: Some(DiskError::other("cannot find disk".to_string()).into()),
-                item_errors: Vec::new(),
             }))
         }
     }
@@ -1632,8 +1612,8 @@ impl NodeService {
 mod tests {
     use super::{
         compat_response_json, decode_msgpack_or_json, decode_rename_data_request_file_info,
-        encode_batch_read_version_response_payloads, encode_delete_versions_errors, encode_file_info_msgpack, encode_msgpack,
-        encode_msgpack_named, encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
+        encode_batch_read_version_response_payloads, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named,
+        encode_read_multiple_response_payloads, encode_rename_data_response_payloads,
     };
     use crate::storage::rpc::node_service::make_server;
     use crate::storage::storage_api::ReadMultipleResp;
@@ -1652,18 +1632,6 @@ mod tests {
         count: u32,
     }
 
-    #[test]
-    fn delete_versions_response_dual_writes_typed_item_errors() {
-        let raw_not_found = super::DiskError::Io(std::io::Error::from(std::io::ErrorKind::NotFound));
-        let (errors, item_errors) = encode_delete_versions_errors(vec![Some(raw_not_found), None]);
-
-        assert!(errors[0].starts_with("io error "));
-        assert!(errors[1].is_empty());
-        assert_eq!(item_errors[0].code, super::DiskError::FileNotFound.to_u32());
-        assert_eq!(item_errors[0].error_info, errors[0]);
-        assert_eq!(item_errors[1].code, 0);
-    }
-
     #[tokio::test]
     #[serial]
     async fn handle_read_version_records_attribution_for_missing_disk() {

scripts/test/vault_ha_kms_live.sh

@@ -241,7 +241,7 @@ env \
   RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
   RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
   cargo test -p rustfs-kms --test vault_ha_failover_live \
-    vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
+    vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
     --ignored --nocapture --test-threads=1 &
 TEST_PID=$!