test(ecstore): use replication status types

.github/workflows/e2e-s3tests.yml

@@ -90,10 +90,6 @@ on:
         description: "Optional pytest -m expression"
         required: false
         default: ""
-      testexpr:
-        description: "Optional pytest -k expression"
-        required: false
-        default: ""
   schedule:
     # Weekly full sweep (Sunday 02:00 UTC): full suite, run against BOTH the
     # single-node and the 4-node distributed topologies (matrix below).
@@ -120,7 +116,6 @@ env:
   XDIST: ${{ github.event.inputs.xdist || '4' }}
   MAXFAIL: ${{ github.event.inputs.maxfail || '0' }}
   MARKEXPR: ${{ github.event.inputs.markexpr || '' }}
-  TESTEXPR: ${{ github.event.inputs.testexpr || '' }}
   S3_SHARD_COUNT: ${{ github.event_name == 'schedule' && '4' || github.event.inputs.shard-count || '1' }}
   TEST_TIMEOUT: "300"
 
@@ -274,20 +269,14 @@ jobs:
           EOF
 
           cat > haproxy.cfg <<'EOF'
-          global
-            log stdout format raw local0 info
-
           defaults
             mode http
-            log global
-            log-format '%ci:%cp [%tr] %ft %b/%s %TR/%Tw/%Tc/%Tr/%Ta %ST %B %tsc %HM %HP'
             timeout connect 5s
             timeout client  30s
             timeout server  30s
 
           frontend fe_s3
             bind *:9000
-            option http-buffer-request
             default_backend be_s3
 
           backend be_s3
@@ -325,7 +314,6 @@ jobs:
           XDIST="${XDIST}" \
           MAXFAIL="${MAXFAIL}" \
           MARKEXPR="${MARKEXPR}" \
-          TESTEXPR="${TESTEXPR}" \
           ./scripts/s3-tests/run.sh
 
       - name: Publish compatibility report

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/rebalance.rs

@@ -859,6 +859,7 @@ fn lifecycle_delete_all_test_failure(phase: crate::object_api::LifecycleDeleteAl
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::bucket::replication::{ReplicationStatusType, VersionPurgeStatusType};
     use crate::config::storageclass::{CLASS_RRS, CLASS_STANDARD, lookup_config_for_pools_without_env};
     use crate::disk::error::DiskError;
     use crate::layout::endpoint::Endpoint;
@@ -1423,6 +1424,14 @@ mod tests {
         }
     }
 
+    fn object_info_with_identity(unix_ts: i64, delete_marker: bool, version_id: Uuid, etag: Option<String>) -> ObjectInfo {
+        ObjectInfo {
+            version_id: Some(version_id),
+            etag,
+            ..object_info_with_mod_time(unix_ts, delete_marker)
+        }
+    }
+
     #[test]
     fn resolve_latest_object_info_candidates_returns_latest_delete_marker() {
         let candidates = vec![
@@ -1446,7 +1455,7 @@ mod tests {
     }
 
     #[test]
-    fn resolve_latest_object_info_candidates_prefers_higher_pool_idx_on_equal_mod_time() {
+    fn resolve_latest_object_info_candidates_prefers_higher_pool_idx_on_equal_mod_time_for_equivalent_candidates() {
         let candidates = vec![
             LatestObjectInfoCandidate {
                 info: Some(object_info_with_mod_time(10, false)),
@@ -1466,6 +1475,333 @@ mod tests {
         assert_eq!(idx, 1);
     }
 
+    #[test]
+    fn resolve_latest_object_info_candidates_keeps_index_fallback_for_fully_equivalent_identities() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 2,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 7,
+                err: None,
+            },
+        ];
+
+        let (info, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect("equivalent replicas must resolve deterministically");
+
+        assert_eq!(idx, 7);
+        assert_eq!(info.version_id, Some(Uuid::from_u128(1)));
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_equal_time_version_id_conflict() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(2), Some("etag-a".to_string()))),
+                idx: 1,
+                err: None,
+            },
+        ];
+
+        let err = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect_err("divergent version ids must not silently resolve to the higher pool index");
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_equal_time_etag_conflict() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-old".to_string()))),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-new".to_string()))),
+                idx: 1,
+                err: None,
+            },
+        ];
+
+        let err = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect_err("divergent etags must not silently resolve to the higher pool index");
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_equal_time_delete_marker_conflict() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), None)),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, true, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 1,
+                err: None,
+            },
+        ];
+
+        let err = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect_err("a delete marker tied with a live version must not be masked by the pool index");
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
+    fn assert_equal_time_identity_conflict(left: ObjectInfo, right: ObjectInfo) {
+        let err = resolve_latest_object_info_candidates(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(left),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(right),
+                    idx: 1,
+                    err: None,
+                },
+            ],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect_err("equal-time identity divergence must fail closed");
+
+        assert_eq!(err, Error::ErasureReadQuorum);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_equal_time_payload_identity_conflicts() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+
+        let mut data_dir = base.clone();
+        data_dir.data_dir = Some(Uuid::from_u128(2));
+        assert_equal_time_identity_conflict(base.clone(), data_dir);
+
+        let mut size = base.clone();
+        size.size = 1;
+        assert_equal_time_identity_conflict(base.clone(), size);
+
+        let mut actual_size = base.clone();
+        actual_size.actual_size = 1;
+        assert_equal_time_identity_conflict(base.clone(), actual_size);
+
+        let mut checksum = base.clone();
+        checksum.checksum = Some(bytes::Bytes::from_static(b"checksum"));
+        assert_equal_time_identity_conflict(base.clone(), checksum);
+
+        let mut parts = base.clone();
+        parts.parts = std::sync::Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
+            etag: "part-etag".to_string(),
+            number: 1,
+            size: 1,
+            ..Default::default()
+        }]);
+        assert_equal_time_identity_conflict(base.clone(), parts);
+
+        let mut transition = base;
+        transition.transitioned_object.tier = "tier-a".to_string();
+        assert_equal_time_identity_conflict(
+            object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())),
+            transition,
+        );
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_accepts_internal_metadata_aliases() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+        let mut rustfs_alias = base.clone();
+        rustfs_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "x-rustfs-internal-compression".to_string(),
+            "zstd".to_string(),
+        )]));
+        let mut minio_alias = base.clone();
+        minio_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "X-MINIO-INTERNAL-COMPRESSION".to_string(),
+            "zstd".to_string(),
+        )]));
+
+        let (_, idx) = resolve_latest_object_info_candidates(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(rustfs_alias),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(minio_alias),
+                    idx: 1,
+                    err: None,
+                },
+            ],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect("same-value internal aliases should resolve");
+        assert_eq!(idx, 1);
+
+        let mut dual_alias = base.clone();
+        dual_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([
+            ("x-rustfs-internal-compression".to_string(), "zstd".to_string()),
+            ("x-minio-internal-compression".to_string(), "zstd".to_string()),
+        ]));
+        let mut single_alias = base;
+        single_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "x-rustfs-internal-compression".to_string(),
+            "zstd".to_string(),
+        )]));
+
+        let (_, idx) = resolve_latest_object_info_candidates(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(dual_alias),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(single_alias),
+                    idx: 1,
+                    err: None,
+                },
+            ],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect("dual-key and single-key internal metadata should resolve");
+        assert_eq!(idx, 1);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_different_internal_metadata_alias_values() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+        let mut rustfs_alias = base.clone();
+        rustfs_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "x-rustfs-internal-compression".to_string(),
+            "zstd".to_string(),
+        )]));
+        let mut minio_alias = base;
+        minio_alias.user_defined = std::sync::Arc::new(std::collections::HashMap::from([(
+            "x-minio-internal-compression".to_string(),
+            "snappy".to_string(),
+        )]));
+
+        assert_equal_time_identity_conflict(rustfs_alias, minio_alias);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_conflicting_internal_metadata_aliases_in_one_candidate() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+        let mut first = base.clone();
+        first.user_defined = std::sync::Arc::new(std::collections::HashMap::from([
+            ("x-rustfs-internal-compression".to_string(), "zstd".to_string()),
+            ("x-minio-internal-compression".to_string(), "snappy".to_string()),
+        ]));
+        let mut second = base;
+        second.user_defined = first.user_defined.clone();
+
+        assert_equal_time_identity_conflict(first, second);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_replication_identity_conflict() {
+        let base = object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+
+        let mut replication = base.clone();
+        replication.replication_status_internal = Some("PENDING".to_string());
+        replication.replication_status = ReplicationStatusType::Pending;
+        assert_equal_time_identity_conflict(base.clone(), replication);
+
+        let mut purge = base.clone();
+        purge.version_purge_status_internal = Some("PENDING".to_string());
+        purge.version_purge_status = VersionPurgeStatusType::Pending;
+        assert_equal_time_identity_conflict(base.clone(), purge);
+
+        let mut decision = base;
+        decision.replication_decision = "replicate".to_string();
+        assert_equal_time_identity_conflict(
+            object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string())),
+            decision,
+        );
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_rejects_none_vs_unix_epoch_mod_time() {
+        let mut without_mod_time = object_info_with_identity(0, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+        without_mod_time.mod_time = None;
+        let with_unix_epoch = object_info_with_identity(0, false, Uuid::from_u128(1), Some("etag-a".to_string()));
+
+        assert_equal_time_identity_conflict(without_mod_time, with_unix_epoch);
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_ignores_older_identity_conflicts() {
+        let latest = object_info_with_identity(20, false, Uuid::from_u128(1), Some("etag-latest".to_string()));
+        let mut older = object_info_with_identity(10, true, Uuid::from_u128(2), Some("etag-old".to_string()));
+        older.data_dir = Some(Uuid::from_u128(2));
+
+        let (info, idx) = resolve_latest_object_info_candidates(
+            vec![
+                LatestObjectInfoCandidate {
+                    info: Some(latest),
+                    idx: 0,
+                    err: None,
+                },
+                LatestObjectInfoCandidate {
+                    info: Some(older),
+                    idx: 9,
+                    err: None,
+                },
+            ],
+            "bucket",
+            "object",
+            &ObjectOptions::default(),
+        )
+        .expect("older identity divergence must not affect the latest candidate");
+
+        assert_eq!(idx, 0);
+        assert_eq!(
+            info.mod_time,
+            Some(OffsetDateTime::from_unix_timestamp(20).expect("operation should succeed"))
+        );
+    }
+
+    #[test]
+    fn resolve_latest_object_info_candidates_ignores_not_found_pools_when_resolving() {
+        let candidates = vec![
+            LatestObjectInfoCandidate {
+                info: Some(object_info_with_identity(10, false, Uuid::from_u128(1), Some("etag-a".to_string()))),
+                idx: 0,
+                err: None,
+            },
+            LatestObjectInfoCandidate {
+                info: None,
+                idx: 1,
+                err: Some(Error::ObjectNotFound("bucket".to_string(), "object".to_string())),
+            },
+        ];
+
+        let (info, idx) = resolve_latest_object_info_candidates(candidates, "bucket", "object", &ObjectOptions::default())
+            .expect("not-found pools must not block resolution of found candidates");
+
+        assert_eq!(idx, 0);
+        assert_eq!(info.version_id, Some(Uuid::from_u128(1)));
+    }
+
     #[test]
     fn resolve_latest_object_info_candidates_returns_non_not_found_error() {
         let err = resolve_latest_object_info_candidates(

crates/ecstore/src/store/rebalance/support.rs

@@ -12,10 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::cmp::Ordering;
+use std::collections::HashMap;
 
 use crate::error::{Error, Result, StorageError, is_err_object_not_found, is_err_version_not_found};
 use crate::object_api::{ObjectInfo, ObjectOptions};
+use rustfs_utils::http::metadata_compat::strip_internal_prefix;
 use rustfs_utils::path::decode_dir_object;
 use time::OffsetDateTime;
 
@@ -137,37 +138,140 @@ pub(super) fn rebalance_disk_set_lookup_error(pool_idx: usize, set_idx: usize, p
     ))
 }
 
+fn latest_candidate_mod_time(candidate: &LatestObjectInfoCandidate) -> Option<OffsetDateTime> {
+    candidate
+        .info
+        .as_ref()
+        .map(|info| info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH))
+}
+
+fn same_transition_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool {
+    left.transition_version_state == right.transition_version_state
+        && left.transitioned_object.name == right.transitioned_object.name
+        && left.transitioned_object.version_id == right.transitioned_object.version_id
+        && left.transitioned_object.tier == right.transitioned_object.tier
+        && left.transitioned_object.free_version == right.transitioned_object.free_version
+        && left.transitioned_object.status == right.transitioned_object.status
+}
+
+#[derive(PartialEq, Eq)]
+struct LatestUserDefinedIdentity {
+    internal: HashMap<String, String>,
+    other: HashMap<String, String>,
+}
+
+fn normalize_user_defined_identity(user_defined: &HashMap<String, String>) -> Option<LatestUserDefinedIdentity> {
+    let mut identity = LatestUserDefinedIdentity {
+        internal: HashMap::with_capacity(user_defined.len()),
+        other: HashMap::with_capacity(user_defined.len()),
+    };
+
+    for (key, value) in user_defined {
+        if let Some(suffix) = strip_internal_prefix(key) {
+            if identity
+                .internal
+                .insert(suffix, value.clone())
+                .is_some_and(|previous| previous != *value)
+            {
+                return None;
+            }
+        } else {
+            identity.other.insert(key.clone(), value.clone());
+        }
+    }
+
+    Some(identity)
+}
+
+fn same_user_defined_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool {
+    match (
+        normalize_user_defined_identity(&left.user_defined),
+        normalize_user_defined_identity(&right.user_defined),
+    ) {
+        (Some(left), Some(right)) => left == right,
+        _ => false,
+    }
+}
+
+/// Pool-specific erasure geometry is intentionally excluded: `get_object_info`
+/// returns each pool's own `data_blocks`/`parity_blocks`, so those values can
+/// differ for the same object version while the selected winner still carries
+/// the chosen pool's layout. `put_object_reader` is also intentionally
+/// excluded because it is a transient request handle that `ObjectInfo::clone`
+/// drops. Every other ObjectInfo field is part of the production-visible
+/// identity and must agree before the pool index can provide a deterministic
+/// tie-break.
+fn same_latest_object_info_identity(left: &ObjectInfo, right: &ObjectInfo) -> bool {
+    left.bucket == right.bucket
+        && left.name == right.name
+        && left.storage_class == right.storage_class
+        && left.mod_time == right.mod_time
+        && left.size == right.size
+        && left.actual_size == right.actual_size
+        && left.is_dir == right.is_dir
+        && same_user_defined_identity(left, right)
+        && left.user_tags == right.user_tags
+        && left.version_id == right.version_id
+        && left.data_dir == right.data_dir
+        && left.delete_marker == right.delete_marker
+        && same_transition_identity(left, right)
+        && left.restore_ongoing == right.restore_ongoing
+        && left.restore_expires == right.restore_expires
+        && left.parts == right.parts
+        && left.is_latest == right.is_latest
+        && left.content_type == right.content_type
+        && left.content_encoding == right.content_encoding
+        && left.expires == right.expires
+        && left.num_versions == right.num_versions
+        && left.successor_mod_time == right.successor_mod_time
+        && left.etag == right.etag
+        && left.inlined == right.inlined
+        && left.metadata_only == right.metadata_only
+        && left.version_only == right.version_only
+        && left.replication_status_internal == right.replication_status_internal
+        && left.replication_status == right.replication_status
+        && left.version_purge_status_internal == right.version_purge_status_internal
+        && left.version_purge_status == right.version_purge_status
+        && left.replication_decision == right.replication_decision
+        && left.checksum == right.checksum
+}
+
 pub(super) fn resolve_latest_object_info_candidates(
-    mut candidates: Vec<LatestObjectInfoCandidate>,
+    candidates: Vec<LatestObjectInfoCandidate>,
     bucket: &str,
     object: &str,
     opts: &ObjectOptions,
 ) -> Result<(ObjectInfo, usize)> {
-    candidates.sort_by(|a, b| {
-        let a_mod = if let Some(info) = &a.info {
-            info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-        } else {
-            OffsetDateTime::UNIX_EPOCH
+    let latest_mod_time = candidates.iter().filter_map(latest_candidate_mod_time).max();
+
+    if let Some(latest_mod_time) = latest_mod_time {
+        let mut latest_candidates = candidates
+            .into_iter()
+            .filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
+            .collect::<Vec<_>>();
+
+        latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx));
+
+        let Some(winner) = latest_candidates.first() else {
+            return Err(Error::ErasureReadQuorum);
+        };
+        let Some(winner_info) = winner.info.as_ref() else {
+            return Err(Error::ErasureReadQuorum);
         };
 
-        let b_mod = if let Some(info) = &b.info {
-            info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)
-        } else {
-            OffsetDateTime::UNIX_EPOCH
-        };
-
-        if a_mod == b_mod {
-            return if a.idx < b.idx { Ordering::Greater } else { Ordering::Less };
+        if latest_candidates.iter().skip(1).any(|candidate| {
+            candidate
+                .info
+                .as_ref()
+                .is_none_or(|info| !same_latest_object_info_identity(winner_info, info))
+        }) {
+            return Err(Error::ErasureReadQuorum);
         }
 
-        b_mod.cmp(&a_mod)
-    });
+        return Ok((winner_info.clone(), winner.idx));
+    }
 
     for candidate in candidates {
-        if let Some(info) = candidate.info {
-            return Ok((info, candidate.idx));
-        }
-
         if let Some(err) = candidate.err
             && !is_err_object_not_found(&err)
             && !is_err_version_not_found(&err)

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() {