ci: refresh Linux full E2E membership after test additions

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
 sha256-darwin=a881fd7d3f5cb94654221ca85b8b30cce1b95e608824a55a15339cbc294e6d34
-sha256-linux=e9a8d64e73f627c4d26c236dbbba690c9ee03a9e26d42a4244515b4439365535
+sha256-linux=a2933d83dfe74ffa03410a0959333a1c48288b8469ca9f17273d449d7510c24b

crates/ecstore/src/api/mod.rs

@@ -479,11 +479,9 @@ pub mod notification {
     #[cfg(any(test, feature = "test-util"))]
     pub use crate::services::notification_sys::rotate_cross_pool_fence_fleet_proof_for_test;
     pub use crate::services::notification_sys::{
-        ClusterTierDailyStats, CrossPoolFenceFleetProofToken, LegacyTransitionStateReconcileFleetProofToken, NotificationPeerErr,
-        NotificationSys, ScannerPublicationLeaseGrant, acquire_cross_pool_fence_fleet_proof,
-        acquire_legacy_transition_state_reconcile_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
-        legacy_transition_state_reconcile_fleet_proof_matches, new_global_notification_sys,
-        scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
+        ClusterTierDailyStats, CrossPoolFenceFleetProofToken, NotificationPeerErr, NotificationSys, ScannerPublicationLeaseGrant,
+        acquire_cross_pool_fence_fleet_proof, cross_pool_fence_fleet_proof_matches, get_global_notification_sys,
+        new_global_notification_sys, scanner_peer_transport_error_message_is_retryable, start_remote_version_state_fleet_probe,
     };
 }
 

crates/ecstore/src/bucket/on_demand_migration/source_client.rs

@@ -14,9 +14,9 @@
 
 //! Outbound client for an on-demand migration source bucket.
 //!
-//! `SourceClient` wraps an `aws_sdk_s3::Client` built through the shared
-//! remote builder and exposes the read-only surface the migration path
-//! needs (HEAD, ranged streaming GET, ListObjectsV2, GetObjectTagging, a
+//! `SourceClient` maps local keys onto a read-only `SourceBackend`. The
+//! S3 backend uses the shared remote builder and exposes the surface the
+//! migration path needs (HEAD, ranged streaming GET, ListObjectsV2, GetObjectTagging, a
 //! probe for admin validation). Every request carries the
 //! `source-proxy-request` anti-loop marker in both the `x-rustfs-` and
 //! `x-minio-` prefixes so a RustFS/MinIO source answers locally instead of
@@ -511,7 +511,7 @@ pub struct SourceObject {
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SourcePage {
     pub objects: Vec<SourceObject>,
-    /// Rolled-up prefixes, in the local namespace; always empty when the
+    /// Rolled-up prefixes, in the same namespace as `objects`; always empty when the
     /// request carried no delimiter.
     pub common_prefixes: Vec<String>,
     pub is_truncated: bool,
@@ -519,7 +519,8 @@ pub struct SourcePage {
 }
 
 /// One `ListObjectsV2` page request against the source. Keys are given in the
-/// local namespace; `SourceClient` maps them through `source_prefix`.
+/// local namespace at `SourceClient`, and in the source namespace at
+/// `SourceBackend`; `SourceClient` maps them through `source_prefix`.
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SourceListRequest<'a> {
     pub prefix: Option<&'a str>,
@@ -575,8 +576,30 @@ impl Intercept for SourceProxyMarkerInterceptor {
     }
 }
 
-pub struct SourceClient {
+/// Read-only provider operations in the source bucket namespace.
+///
+/// Implementations must preserve streaming, honor the requested range and
+/// pagination cursor, and classify failures without including credentials.
+/// `SourceClient` owns prefix mapping so every provider shares the same local
+/// namespace. Continuation tokens are opaque and must never be prefix-mapped.
+#[async_trait::async_trait]
+pub trait SourceBackend: Send + Sync {
+    async fn head(&self, key: &str) -> Result<SourceHead, SourceError>;
+    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError>;
+    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError>;
+    async fn tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError>;
+    /// Verify bucket access; `SourceClient` separately probes a filtered listing.
+    async fn probe(&self) -> Result<(), SourceError>;
+}
+
+/// S3-compatible implementation, including request signing and anti-loop headers.
+pub struct S3SourceBackend {
     client: S3Client,
+    bucket: String,
+}
+
+pub struct SourceClient {
+    backend: Box<dyn SourceBackend>,
     endpoint: String,
     bucket: String,
     source_prefix: Option<String>,
@@ -609,7 +632,10 @@ impl SourceClient {
     fn from_config_builder(config: aws_sdk_s3::config::Builder, endpoint: String, spec: &SourceClientSpec) -> Self {
         let client = S3Client::from_conf(config.interceptor(SourceProxyMarkerInterceptor::new()).build());
         Self {
-            client,
+            backend: Box::new(S3SourceBackend {
+                client,
+                bucket: spec.bucket.clone(),
+            }),
             endpoint,
             bucket: spec.bucket.clone(),
             source_prefix: spec.source_prefix.clone().filter(|prefix| !prefix.is_empty()),
@@ -652,35 +678,15 @@ impl SourceClient {
     }
 
     pub async fn head_object(&self, key: &str) -> Result<SourceHead, SourceError> {
-        let output = self
-            .client
-            .head_object()
-            .bucket(&self.bucket)
-            .key(self.source_key(key))
-            .send()
-            .await
-            .map_err(classify_sdk_error)?;
-        source_head_from_head_output(output)
+        self.backend.head(&self.source_key(key)).await
     }
 
-    /// Streams the object; `range` is passed through as an HTTP `Range`
-    /// header and omitted entirely when `None`.
+    /// Streams the object, preserving an optional HTTP byte range.
     pub async fn get_object(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
-        let range = range.map(range_header_value).transpose()?;
-        let output = self
-            .client
-            .get_object()
-            .bucket(&self.bucket)
-            .key(self.source_key(key))
-            .set_range(range)
-            .send()
-            .await
-            .map_err(classify_sdk_error)?;
-        source_get_from_output(output)
+        self.backend.get(&self.source_key(key), range).await
     }
 
-    /// Lists one page under the local `prefix`. Keys are returned in the
-    /// local namespace; entries outside `source_prefix` are skipped.
+    /// Lists one page under the local prefix.
     pub async fn list_objects_v2(
         &self,
         prefix: Option<&str>,
@@ -696,9 +702,81 @@ impl SourceClient {
         .await
     }
 
-    /// [`Self::list_objects_v2`] with the delimiter and start-after the
-    /// list-through merge needs (rustfs/backlog#2164).
+    /// Maps keys and common prefixes while leaving opaque cursors untouched.
     pub async fn list_page(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
+        let prefix = self.source_key(request.prefix.unwrap_or_default());
+        let start_after = request.start_after.map(|key| self.source_key(key));
+        let mut page = self
+            .backend
+            .list(&SourceListRequest {
+                prefix: Some(&prefix),
+                start_after: start_after.as_deref(),
+                ..*request
+            })
+            .await?;
+        page.objects = page
+            .objects
+            .into_iter()
+            .filter_map(|object| self.local_object(object))
+            .collect();
+        page.common_prefixes = page
+            .common_prefixes
+            .into_iter()
+            .filter_map(|prefix| self.local_key(&prefix).map(str::to_string))
+            .collect();
+        Ok(page)
+    }
+
+    fn local_object(&self, mut object: SourceObject) -> Option<SourceObject> {
+        object.key = self.local_key(&object.key)?.to_string();
+        Some(object)
+    }
+
+    pub async fn get_object_tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
+        self.backend.tagging(&self.source_key(key)).await
+    }
+
+    pub async fn probe(&self) -> Result<SourceProbe, SourceError> {
+        self.backend.probe().await?;
+        let page = self.list_objects_v2(None, None, 1).await?;
+        Ok(SourceProbe {
+            sample_object: page.objects.into_iter().next(),
+            has_more_objects: page.is_truncated,
+        })
+    }
+}
+
+#[async_trait::async_trait]
+impl SourceBackend for S3SourceBackend {
+    async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
+        let output = self
+            .client
+            .head_object()
+            .bucket(&self.bucket)
+            .key(key)
+            .send()
+            .await
+            .map_err(classify_sdk_error)?;
+        source_head_from_head_output(output)
+    }
+
+    /// Streams the object; `range` is passed through as an HTTP `Range`
+    /// header and omitted entirely when `None`.
+    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
+        let range = range.map(range_header_value).transpose()?;
+        let output = self
+            .client
+            .get_object()
+            .bucket(&self.bucket)
+            .key(key)
+            .set_range(range)
+            .send()
+            .await
+            .map_err(classify_sdk_error)?;
+        source_get_from_output(output)
+    }
+
+    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
         // `start_after` is silently ignored by S3 once a continuation token is
         // present; refuse the ambiguous pair rather than list from the wrong
         // position.
@@ -711,9 +789,9 @@ impl SourceClient {
             .client
             .list_objects_v2()
             .bucket(&self.bucket)
-            .prefix(self.source_key(request.prefix.unwrap_or_default()))
+            .prefix(request.prefix.unwrap_or_default())
             .set_delimiter(request.delimiter.map(str::to_string))
-            .set_start_after(request.start_after.map(|after| self.source_key(after)))
+            .set_start_after(request.start_after.map(str::to_string))
             .set_continuation_token(request.continuation_token.map(str::to_string))
             .max_keys(request.max_keys)
             .send()
@@ -731,13 +809,13 @@ impl SourceClient {
             .contents
             .unwrap_or_default()
             .into_iter()
-            .filter_map(|object| self.source_object(object))
+            .filter_map(s3_source_object)
             .collect();
         let common_prefixes = output
             .common_prefixes
             .unwrap_or_default()
             .into_iter()
-            .filter_map(|prefix| Some(self.local_key(prefix.prefix.as_deref()?)?.to_string()))
+            .filter_map(|prefix| prefix.prefix)
             .collect();
 
         Ok(SourcePage {
@@ -748,48 +826,43 @@ impl SourceClient {
         })
     }
 
-    fn source_object(&self, object: SdkObject) -> Option<SourceObject> {
-        let key = self.local_key(object.key.as_deref()?)?.to_string();
-        let etag = normalize_etag(object.e_tag);
-        let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
-        Some(SourceObject {
-            key,
-            etag,
-            size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
-            last_modified: system_time(object.last_modified),
-            storage_class: object.storage_class.map(|class| class.as_str().to_string()),
-            is_multipart_etag,
-        })
-    }
-
-    pub async fn get_object_tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
+    async fn tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
         let output = self
             .client
             .get_object_tagging()
             .bucket(&self.bucket)
-            .key(self.source_key(key))
+            .key(key)
             .send()
             .await
             .map_err(classify_sdk_error)?;
         Ok(output.tag_set.into_iter().map(|tag| (tag.key, tag.value)).collect())
     }
 
-    /// Admin validation: HeadBucket plus a one-key listing under the prefix.
-    pub async fn probe(&self) -> Result<SourceProbe, SourceError> {
+    async fn probe(&self) -> Result<(), SourceError> {
         self.client
             .head_bucket()
             .bucket(&self.bucket)
             .send()
             .await
             .map_err(classify_sdk_error)?;
-        let page = self.list_objects_v2(None, None, 1).await?;
-        Ok(SourceProbe {
-            sample_object: page.objects.into_iter().next(),
-            has_more_objects: page.is_truncated,
-        })
+        Ok(())
     }
 }
 
+fn s3_source_object(object: SdkObject) -> Option<SourceObject> {
+    let key = object.key?;
+    let etag = normalize_etag(object.e_tag);
+    let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
+    Some(SourceObject {
+        key,
+        etag,
+        size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
+        last_modified: system_time(object.last_modified),
+        storage_class: object.storage_class.map(|class| class.as_str().to_string()),
+        is_multipart_etag,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1194,6 +1267,68 @@ mod tests {
         assert!(requests[1].uri.contains("continuation-token=token-1"), "{}", requests[1].uri);
     }
 
+    #[tokio::test]
+    async fn list_page_maps_delimiter_prefixes_and_start_after_but_not_cursors() {
+        let body = r#"<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+<IsTruncated>true</IsTruncated><NextContinuationToken>data/opaque</NextContinuationToken>
+<CommonPrefixes><Prefix>data/photos/</Prefix></CommonPrefixes>
+<CommonPrefixes><Prefix>outside/</Prefix></CommonPrefixes>
+</ListBucketResult>"#;
+        let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), body)]).await;
+        let first = client
+            .list_page(&SourceListRequest {
+                prefix: Some("photos/"),
+                delimiter: Some("/"),
+                start_after: Some("photos/a"),
+                max_keys: 2,
+                ..Default::default()
+            })
+            .await
+            .expect("delimiter listing should succeed");
+        assert_eq!(first.common_prefixes, vec!["photos/"]);
+        assert_eq!(first.next_continuation_token.as_deref(), Some("data/opaque"));
+        let second = client
+            .list_page(&SourceListRequest {
+                continuation_token: first.next_continuation_token.as_deref(),
+                max_keys: 2,
+                ..Default::default()
+            })
+            .await
+            .expect("opaque continuation should succeed");
+        assert_eq!(second.common_prefixes, first.common_prefixes);
+        let requests = recorded(&requests);
+        let query = |request: &RecordedRequest| {
+            Url::parse(&request.uri)
+                .expect("request URI")
+                .query_pairs()
+                .into_owned()
+                .collect::<HashMap<_, _>>()
+        };
+        let first_query = query(&requests[0]);
+        assert_eq!(first_query.get("prefix").map(String::as_str), Some("data/photos/"));
+        assert_eq!(first_query.get("start-after").map(String::as_str), Some("data/photos/a"));
+        assert_eq!(first_query.get("delimiter").map(String::as_str), Some("/"));
+        let second_query = query(&requests[1]);
+        assert_eq!(second_query.get("continuation-token").map(String::as_str), Some("data/opaque"));
+        assert!(!second_query.contains_key("start-after"));
+    }
+
+    #[tokio::test]
+    async fn list_page_rejects_ambiguous_cursor_before_sending() {
+        let (client, requests) = scripted_client(&spec(Some("data/")), vec![]).await;
+        let err = client
+            .list_page(&SourceListRequest {
+                start_after: Some("a"),
+                continuation_token: Some("opaque"),
+                max_keys: 1,
+                ..Default::default()
+            })
+            .await
+            .expect_err("ambiguous list position must fail");
+        assert!(matches!(err, SourceError::Other(_)));
+        assert!(recorded(&requests).is_empty(), "invalid request must never reach the source");
+    }
+
     #[tokio::test]
     async fn list_objects_v2_rejects_truncated_page_without_token() {
         let (client, _) = scripted_client(&spec(None), vec![ok(Vec::new(), LIST_TRUNCATED_WITHOUT_TOKEN)]).await;
@@ -1357,11 +1492,14 @@ mod tests {
 
     fn prefix_client(prefix: Option<String>) -> SourceClient {
         SourceClient {
-            client: S3Client::from_conf(
-                aws_sdk_s3::Config::builder()
-                    .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
-                    .build(),
-            ),
+            backend: Box::new(S3SourceBackend {
+                client: S3Client::from_conf(
+                    aws_sdk_s3::Config::builder()
+                        .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                        .build(),
+                ),
+                bucket: "bucket".to_string(),
+            }),
             endpoint: "https://source.example.com".to_string(),
             bucket: "bucket".to_string(),
             source_prefix: prefix.filter(|prefix| !prefix.is_empty()),

crates/ecstore/src/services/notification_sys.rs

@@ -62,27 +62,12 @@ const REMOTE_VERSION_STATE_PROOF_TTL: Duration = Duration::from_secs(30);
 const CROSS_POOL_FENCE_SUPPORTED_VERSION: u32 = 2;
 const TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION: u32 = 3;
 const DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION: u32 = 4;
-// Keep this synchronized with the version served by node_service. Including
-// the local member in the minimum prevents an older coordinator from
-// self-authorizing a policy implemented only by newer remote peers.
-const LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION: u32 = 4;
-/// Version 5 is reserved for a fleet whose every metadata writer preserves
-/// explicit transition version state and destination identity, and implements
-/// conditional per-generation `xl.meta` writes with strong readback. The node
-/// service must not advertise this version until the conditional writer from
-/// rustfs/backlog#684 is available.
-const LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION: u32 = 5;
 type CrossPoolFencePolicyResult = Result<BTreeMap<String, Uuid>>;
 
 fn cross_pool_fence_policy_results(
     peer_epochs: BTreeMap<String, Uuid>,
     minimum_version: u32,
-) -> (
-    CrossPoolFencePolicyResult,
-    CrossPoolFencePolicyResult,
-    CrossPoolFencePolicyResult,
-    CrossPoolFencePolicyResult,
-) {
+) -> (CrossPoolFencePolicyResult, CrossPoolFencePolicyResult, CrossPoolFencePolicyResult) {
     let journal_result = if minimum_version >= TIER_DELETE_JOURNAL_POLICY_SUPPORTED_VERSION {
         Ok(peer_epochs.clone())
     } else {
@@ -93,18 +78,7 @@ fn cross_pool_fence_policy_results(
     } else {
         Err(Error::other("decommission target fence policy capability version is unsupported"))
     };
-    let legacy_transition_state_reconcile_result =
-        if minimum_version >= LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION {
-            Ok(peer_epochs.clone())
-        } else {
-            Err(Error::other("legacy transition state reconcile policy capability version is unsupported"))
-        };
-    (
-        Ok(peer_epochs),
-        journal_result,
-        decommission_target_fence_result,
-        legacy_transition_state_reconcile_result,
-    )
+    (Ok(peer_epochs), journal_result, decommission_target_fence_result)
 }
 
 #[derive(Clone, Debug)]
@@ -278,21 +252,10 @@ pub(crate) struct TierDeleteJournalFleetProofToken {
     _permit: FleetCapabilityProofPermit,
 }
 
-/// Effect-window authority for one legacy transition-state reconciliation.
-///
-/// The token intentionally cannot be cloned. Its permit keeps the admitted
-/// fleet generation alive until the caller finishes the final strong
-/// readback, while revocation makes every later validation fail immediately.
-pub struct LegacyTransitionStateReconcileFleetProofToken {
-    token: FleetCapabilityProofToken,
-    _permit: FleetCapabilityProofPermit,
-}
-
 static REMOTE_VERSION_STATE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static CROSS_POOL_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static TIER_DELETE_JOURNAL_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static DECOMMISSION_TARGET_FENCE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
-static LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF: OnceLock<std::sync::RwLock<FleetCapabilityProofState>> = OnceLock::new();
 static REMOTE_VERSION_STATE_PROBE_TOPOLOGY: OnceLock<String> = OnceLock::new();
 
 fn cross_pool_fence_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
@@ -311,10 +274,6 @@ fn decommission_target_fence_fleet_proof_slot() -> &'static std::sync::RwLock<Fl
     DECOMMISSION_TARGET_FENCE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
 }
 
-fn legacy_transition_state_reconcile_fleet_proof_slot() -> &'static std::sync::RwLock<FleetCapabilityProofState> {
-    LEGACY_TRANSITION_STATE_RECONCILE_FLEET_PROOF.get_or_init(|| std::sync::RwLock::new(FleetCapabilityProofState::default()))
-}
-
 fn revoke_fleet_capability_proof_state(state: &mut FleetCapabilityProofState) {
     if let Some(proof) = state.proof.take() {
         proof.generation.revoke();
@@ -485,125 +444,6 @@ pub(crate) fn tier_delete_journal_topology_generation(proof: &TierDeleteJournalF
     stable_tier_delete_journal_topology_generation(&proof.token.topology_fingerprint)
 }
 
-/// Acquire one non-cloneable authority that must span the complete reconcile
-/// effect window, including its final strong readback.
-pub async fn acquire_legacy_transition_state_reconcile_fleet_proof() -> Option<LegacyTransitionStateReconcileFleetProofToken> {
-    let expected_topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get()?;
-    let proof = {
-        let state = legacy_transition_state_reconcile_fleet_proof_slot()
-            .read()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, expected_topology, Instant::now())?
-    };
-    let observed_peer_epochs = observe_legacy_transition_state_reconcile_fleet(expected_topology).await?;
-    let state = legacy_transition_state_reconcile_fleet_proof_slot()
-        .read()
-        .unwrap_or_else(std::sync::PoisonError::into_inner);
-    legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
-        &state,
-        &proof,
-        expected_topology,
-        &observed_peer_epochs,
-        Instant::now(),
-    )
-    .then_some(proof)
-}
-
-fn acquire_legacy_transition_state_reconcile_fleet_proof_from(
-    state: &FleetCapabilityProofState,
-    expected_topology: &str,
-    now: Instant,
-) -> Option<LegacyTransitionStateReconcileFleetProofToken> {
-    let token = acquire_fleet_capability_proof_from(state, expected_topology, now)?;
-    let permit = state.proof.as_ref()?.generation.try_acquire()?;
-    Some(LegacyTransitionStateReconcileFleetProofToken { token, _permit: permit })
-}
-
-async fn observe_legacy_transition_state_reconcile_fleet(expected_topology: &str) -> Option<BTreeMap<String, Uuid>> {
-    let notification_sys = get_global_notification_sys()?;
-    let (peer_epochs, minimum_version) = timeout(
-        REMOTE_VERSION_STATE_PROBE_TIMEOUT,
-        notification_sys.probe_cross_pool_fence_fleet(expected_topology),
-    )
-    .await
-    .ok()?
-    .ok()?;
-    let (_, _, _, reconcile_result) = cross_pool_fence_policy_results(peer_epochs, minimum_version);
-    reconcile_result.ok()
-}
-
-/// Revalidate the exact fleet generation captured by a reconcile token with a
-/// fresh synchronous observation. Callers must await this before each
-/// conditional metadata write and after the final strong readback.
-pub async fn legacy_transition_state_reconcile_fleet_proof_matches(
-    proof: &LegacyTransitionStateReconcileFleetProofToken,
-) -> bool {
-    let Some(expected_topology) = REMOTE_VERSION_STATE_PROBE_TOPOLOGY.get() else {
-        return false;
-    };
-    legacy_transition_state_reconcile_fleet_proof_matches_with_observer(
-        legacy_transition_state_reconcile_fleet_proof_slot(),
-        proof,
-        expected_topology,
-        || observe_legacy_transition_state_reconcile_fleet(expected_topology),
-    )
-    .await
-}
-
-async fn legacy_transition_state_reconcile_fleet_proof_matches_with_observer<F, Fut>(
-    slot: &std::sync::RwLock<FleetCapabilityProofState>,
-    proof: &LegacyTransitionStateReconcileFleetProofToken,
-    expected_topology: &str,
-    observe: F,
-) -> bool
-where
-    F: FnOnce() -> Fut,
-    Fut: Future<Output = Option<BTreeMap<String, Uuid>>>,
-{
-    {
-        let state = slot.read().unwrap_or_else(std::sync::PoisonError::into_inner);
-        if !legacy_transition_state_reconcile_fleet_proof_matches_at(&state, proof, expected_topology, Instant::now()) {
-            return false;
-        }
-    }
-    let Some(observed_peer_epochs) = observe().await else {
-        return false;
-    };
-    let state = slot.read().unwrap_or_else(std::sync::PoisonError::into_inner);
-    legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
-        &state,
-        proof,
-        expected_topology,
-        &observed_peer_epochs,
-        Instant::now(),
-    )
-}
-
-fn legacy_transition_state_reconcile_fleet_proof_matches_at(
-    state: &FleetCapabilityProofState,
-    proof: &LegacyTransitionStateReconcileFleetProofToken,
-    expected_topology: &str,
-    now: Instant,
-) -> bool {
-    proof._permit.generation.is_accepting()
-        && fleet_capability_proof_matches_at(state, &proof.token, expected_topology, now)
-        && state
-            .proof
-            .as_ref()
-            .is_some_and(|current| Arc::ptr_eq(&current.generation, &proof._permit.generation))
-}
-
-fn legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
-    state: &FleetCapabilityProofState,
-    proof: &LegacyTransitionStateReconcileFleetProofToken,
-    expected_topology: &str,
-    observed_peer_epochs: &BTreeMap<String, Uuid>,
-    now: Instant,
-) -> bool {
-    legacy_transition_state_reconcile_fleet_proof_matches_at(state, proof, expected_topology, now)
-        && proof.token.peer_epochs.as_ref() == observed_peer_epochs
-}
-
 #[cfg(all(test, feature = "test-util"))]
 pub(crate) fn tier_delete_journal_fleet_proof_has_inflight_for_test() -> bool {
     let state = tier_delete_journal_fleet_proof_slot()
@@ -926,7 +766,6 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 cross_pool_fence_fleet_proof_slot(),
                 tier_delete_journal_fleet_proof_slot(),
                 decommission_target_fence_fleet_proof_slot(),
-                legacy_transition_state_reconcile_fleet_proof_slot(),
             ] {
                 mark_fleet_capability_topology_conflict(slot);
             }
@@ -959,12 +798,11 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 .unwrap_or_else(|_| Err(Error::other("cross-pool fence fleet capability probe timed out"))),
                 None => Err(Error::other("cross-pool fence fleet capability notification system is unavailable")),
             };
-            let (fence_result, journal_result, decommission_target_fence_result, reconcile_result) = match fence_probe {
+            let (fence_result, journal_result, decommission_target_fence_result) = match fence_probe {
                 Ok((peer_epochs, minimum_version)) => cross_pool_fence_policy_results(peer_epochs, minimum_version),
                 Err(err) => {
                     let message = err.to_string();
                     (
-                        Err(Error::other(message.clone())),
                         Err(Error::other(message.clone())),
                         Err(Error::other(message.clone())),
                         Err(Error::other(message)),
@@ -980,7 +818,6 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                 revoke_fleet_capability_proof(cross_pool_fence_fleet_proof_slot());
                 revoke_fleet_capability_proof(tier_delete_journal_fleet_proof_slot());
                 revoke_fleet_capability_proof(decommission_target_fence_fleet_proof_slot());
-                revoke_fleet_capability_proof(legacy_transition_state_reconcile_fleet_proof_slot());
             } else if let Some(err) = publish_fleet_capability_probe_result(
                 remote_version_state_fleet_proof_slot(),
                 &topology_fingerprint,
@@ -1043,24 +880,6 @@ pub fn start_remote_version_state_fleet_probe(topology_fingerprint: String) {
                     "notification capability probe"
                 );
             }
-            if !topology_conflict
-                && let Some(err) = publish_fleet_capability_probe_result(
-                    legacy_transition_state_reconcile_fleet_proof_slot(),
-                    &topology_fingerprint,
-                    reconcile_result,
-                    Instant::now(),
-                )
-            {
-                debug!(
-                    event = EVENT_NOTIFICATION_CAPABILITY_PROBE,
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_NOTIFICATION,
-                    capability = "legacy_transition_state_reconcile_v1",
-                    state = "failed_closed",
-                    error = %err,
-                    "notification capability probe"
-                );
-            }
             sleep(REMOTE_VERSION_STATE_PROBE_INTERVAL).await;
         }
     });
@@ -1140,7 +959,7 @@ impl NotificationSys {
             client.probe_cross_pool_fence(topology_fingerprint.to_string()).await
         });
         let mut peer_epochs = BTreeMap::new();
-        let mut minimum_version = LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION;
+        let mut minimum_version = u32::MAX;
         for result in join_all(probes).await {
             let (peer, version, epoch) = result?;
             if version < CROSS_POOL_FENCE_SUPPORTED_VERSION {
@@ -1149,6 +968,11 @@ impl NotificationSys {
             minimum_version = minimum_version.min(version);
             insert_remote_version_state_peer(&mut peer_epochs, peer, epoch)?;
         }
+        // A single-node deployment has no remote member to lower the local
+        // policy version advertised by this binary.
+        if minimum_version == u32::MAX {
+            minimum_version = DECOMMISSION_TARGET_FENCE_POLICY_SUPPORTED_VERSION;
+        }
         Ok((peer_epochs, minimum_version))
     }
 }
@@ -3366,36 +3190,20 @@ mod tests {
     #[test]
     fn cross_pool_policy_versions_authorize_only_their_supported_protocols() {
         let peers = BTreeMap::from([("node-b:9000".to_string(), Uuid::new_v4())]);
-        let (generic_v2, journal_v2, decommission_v2, reconcile_v2) = cross_pool_fence_policy_results(peers.clone(), 2);
+        let (generic_v2, journal_v2, decommission_v2) = cross_pool_fence_policy_results(peers.clone(), 2);
         assert!(generic_v2.is_ok(), "v2 remains valid for existing cross-pool fencing");
         assert!(journal_v2.is_err(), "a mixed v2/v3 fleet must fail closed for journal-v6 deletion");
         assert!(decommission_v2.is_err(), "v2 cannot authorize the sticky per-target decommission fence");
-        assert!(reconcile_v2.is_err(), "v2 cannot authorize legacy transition-state reconciliation");
 
-        let (generic_v3, journal_v3, decommission_v3, reconcile_v3) = cross_pool_fence_policy_results(peers.clone(), 3);
+        let (generic_v3, journal_v3, decommission_v3) = cross_pool_fence_policy_results(peers.clone(), 3);
         assert!(generic_v3.is_ok());
         assert!(journal_v3.is_ok(), "an all-v3 fleet may authorize journal-v6 deletion");
         assert!(decommission_v3.is_err(), "v3 members do not understand the per-target decommission fence");
-        assert!(reconcile_v3.is_err());
 
-        let (generic_v4, journal_v4, decommission_v4, reconcile_v4) =
-            cross_pool_fence_policy_results(peers.clone(), LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION);
+        let (generic_v4, journal_v4, decommission_v4) = cross_pool_fence_policy_results(peers, 4);
         assert!(generic_v4.is_ok());
         assert!(journal_v4.is_ok());
         assert!(decommission_v4.is_ok(), "an all-v4 fleet may create sticky per-target reservations");
-        assert!(
-            reconcile_v4.is_err(),
-            "the current local policy lacks the conditional xl.meta writer required by reconcile"
-        );
-
-        let (generic_v5, journal_v5, decommission_v5, reconcile_v5) = cross_pool_fence_policy_results(peers, 5);
-        assert!(generic_v5.is_ok());
-        assert!(journal_v5.is_ok());
-        assert!(decommission_v5.is_ok());
-        assert!(
-            reconcile_v5.is_ok(),
-            "only an all-v5 fleet preserves destination identity and conditional reconcile writes"
-        );
     }
 
     #[test]
@@ -3650,234 +3458,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn legacy_transition_state_reconcile_admits_only_compatible_single_and_multi_node_fleets() {
-        let now = Instant::now();
-        for peers in [
-            BTreeMap::new(),
-            BTreeMap::from([("peer-a".to_string(), Uuid::new_v4()), ("peer-b".to_string(), Uuid::new_v4())]),
-        ] {
-            let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
-            let (_, _, _, result) =
-                cross_pool_fence_policy_results(peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
-            assert!(publish_fleet_capability_probe_result(&slot, "topology-a", result, now).is_none());
-
-            let admitted = {
-                let state = slot.read().expect("reconcile proof slot should not poison");
-                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
-                    .expect("an all-compatible fleet should admit reconciliation")
-            };
-            let state = slot.read().expect("reconcile proof slot should not poison");
-            assert!(legacy_transition_state_reconcile_fleet_proof_matches_at(
-                &state,
-                &admitted,
-                "topology-a",
-                now,
-            ));
-        }
-    }
-
-    #[test]
-    fn legacy_transition_state_reconcile_restart_drains_concurrent_effect_windows() {
-        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
-        let now = Instant::now();
-        let original_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
-        let (_, _, _, original_result) =
-            cross_pool_fence_policy_results(original_peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
-        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", original_result, now).is_none());
-
-        let (first, second) = {
-            let state = slot.read().expect("reconcile proof slot should not poison");
-            (
-                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
-                    .expect("the first reconcile writer should be admitted"),
-                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
-                    .expect("the second reconcile writer should be admitted"),
-            )
-        };
-
-        let restarted_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
-        let (_, _, _, restarted_result) =
-            cross_pool_fence_policy_results(restarted_peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
-        let blocked =
-            publish_fleet_capability_probe_result(&slot, "topology-a", restarted_result, now + Duration::from_millis(1))
-                .expect("a restarted member must revoke the old generation and wait for both writers");
-        assert!(blocked.to_string().contains("previous generation to drain"));
-        {
-            let state = slot.read().expect("reconcile proof slot should not poison");
-            assert!(state.proof.is_none());
-            assert!(state.draining_generation.is_some());
-            assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
-                &state,
-                &first,
-                "topology-a",
-                now + Duration::from_millis(1),
-            ));
-            assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
-                &state,
-                &second,
-                "topology-a",
-                now + Duration::from_millis(1),
-            ));
-        }
-
-        drop(first);
-        let (_, _, _, still_blocked_result) =
-            cross_pool_fence_policy_results(restarted_peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
-        assert!(
-            publish_fleet_capability_probe_result(&slot, "topology-a", still_blocked_result, now + Duration::from_millis(2),)
-                .is_some(),
-            "one remaining writer must keep the successor generation closed"
-        );
-
-        drop(second);
-        let (_, _, _, admitted_result) =
-            cross_pool_fence_policy_results(restarted_peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
-        assert!(
-            publish_fleet_capability_probe_result(&slot, "topology-a", admitted_result, now + Duration::from_millis(3),)
-                .is_none(),
-            "the restarted generation may publish only after every old writer drains"
-        );
-    }
-
-    #[test]
-    fn legacy_transition_state_reconcile_fresh_observation_closes_the_polling_window() {
-        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
-        let now = Instant::now();
-        let original_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
-        let (_, _, _, original_result) =
-            cross_pool_fence_policy_results(original_peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
-        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", original_result, now).is_none());
-        let admitted = {
-            let state = slot.read().expect("reconcile proof slot should not poison");
-            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
-                .expect("the original fleet should admit reconciliation")
-        };
-
-        let restarted_peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
-        let state = slot.read().expect("reconcile proof slot should not poison");
-        assert!(
-            legacy_transition_state_reconcile_fleet_proof_matches_at(&state, &admitted, "topology-a", now),
-            "the periodic cache has not observed the restart yet"
-        );
-        assert!(!legacy_transition_state_reconcile_fleet_proof_matches_observation_at(
-            &state,
-            &admitted,
-            "topology-a",
-            &restarted_peers,
-            now,
-        ));
-
-        let (_, _, _, downgraded) = cross_pool_fence_policy_results(original_peers, 4);
-        assert!(
-            downgraded.is_err(),
-            "a synchronous observation of a downgraded peer must fail before any cached proof can authorize a write"
-        );
-    }
-
-    #[tokio::test]
-    async fn legacy_transition_state_reconcile_invalid_token_skips_fleet_observation() {
-        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
-        let now = Instant::now();
-        let peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
-        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(peers), now).is_none());
-        let admitted = {
-            let state = slot.read().expect("reconcile proof slot should not poison");
-            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
-                .expect("the original fleet should admit reconciliation")
-        };
-        revoke_fleet_capability_proof(&slot);
-
-        assert!(
-            !legacy_transition_state_reconcile_fleet_proof_matches_with_observer(&slot, &admitted, "topology-a", || async {
-                panic!("an invalid local generation must not trigger a fleet observation");
-            },)
-            .await
-        );
-    }
-
-    #[test]
-    fn legacy_transition_state_reconcile_membership_and_topology_changes_revoke_authority() {
-        let now = Instant::now();
-        for replacement in [
-            BTreeMap::from([("peer-a".to_string(), Uuid::new_v4()), ("peer-b".to_string(), Uuid::new_v4())]),
-            BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]),
-        ] {
-            let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
-            let original = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
-            assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(original), now).is_none());
-            let admitted = {
-                let state = slot.read().expect("reconcile proof slot should not poison");
-                acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
-                    .expect("the original fleet should admit reconciliation")
-            };
-
-            assert!(
-                publish_fleet_capability_probe_result(&slot, "topology-a", Ok(replacement), now + Duration::from_millis(1),)
-                    .is_some(),
-                "membership or process-epoch replacement must wait for the admitted writer"
-            );
-            let state = slot.read().expect("reconcile proof slot should not poison");
-            assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
-                &state,
-                &admitted,
-                "topology-a",
-                now + Duration::from_millis(1),
-            ));
-        }
-
-        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
-        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", Ok(BTreeMap::new()), now).is_none());
-        let admitted = {
-            let state = slot.read().expect("reconcile proof slot should not poison");
-            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
-                .expect("the original topology should admit reconciliation")
-        };
-        mark_fleet_capability_topology_conflict(&slot);
-        let state = slot.read().expect("reconcile proof slot should not poison");
-        assert!(state.topology_conflict);
-        assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
-            &state,
-            &admitted,
-            "topology-a",
-            now,
-        ));
-    }
-
-    #[test]
-    fn legacy_transition_state_reconcile_capability_downgrade_fails_closed() {
-        let slot = std::sync::RwLock::new(FleetCapabilityProofState::default());
-        let now = Instant::now();
-        let peers = BTreeMap::from([("peer-a".to_string(), Uuid::new_v4())]);
-        let (_, _, _, compatible_result) =
-            cross_pool_fence_policy_results(peers.clone(), LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION);
-        assert!(publish_fleet_capability_probe_result(&slot, "topology-a", compatible_result, now).is_none());
-        let admitted = {
-            let state = slot.read().expect("reconcile proof slot should not poison");
-            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now)
-                .expect("v5 should admit reconciliation")
-        };
-
-        let (_, _, _, downgraded_result) =
-            cross_pool_fence_policy_results(peers, LEGACY_TRANSITION_STATE_RECONCILE_POLICY_SUPPORTED_VERSION - 1);
-        let err = publish_fleet_capability_probe_result(&slot, "topology-a", downgraded_result, now + Duration::from_millis(1))
-            .expect("a v4 member must revoke reconcile authority");
-        assert!(err.to_string().contains("reconcile policy capability version is unsupported"));
-        let state = slot.read().expect("reconcile proof slot should not poison");
-        assert!(state.proof.is_none());
-        assert!(!legacy_transition_state_reconcile_fleet_proof_matches_at(
-            &state,
-            &admitted,
-            "topology-a",
-            now + Duration::from_millis(1),
-        ));
-        assert!(
-            acquire_legacy_transition_state_reconcile_fleet_proof_from(&state, "topology-a", now + Duration::from_millis(1),)
-                .is_none(),
-            "a downgraded fleet must remain inspect-only"
-        );
-    }
-
     #[test]
     fn remote_version_state_fleet_proof_conflict_revokes_atomic_snapshot() {
         let now = Instant::now();
@@ -3959,57 +3539,6 @@ mod tests {
         assert!(err.to_string().contains("incomplete"));
     }
 
-    #[tokio::test]
-    async fn legacy_transition_state_reconcile_probe_rejects_missing_or_unreachable_members() {
-        let missing = NotificationSys {
-            peer_clients: Vec::new(),
-            all_peer_clients: vec![None],
-            peer_topology_hosts: vec!["peer-a".to_string()],
-            peer_admin_caches: Vec::new(),
-            tier_config_reload_workers: Default::default(),
-        };
-        let missing_err = missing
-            .probe_cross_pool_fence_fleet("topology-a")
-            .await
-            .expect_err("a missing member slot must prevent reconcile capability proof");
-        assert!(missing_err.to_string().contains("incomplete"));
-
-        let unreachable = NotificationSys {
-            peer_clients: vec![None],
-            all_peer_clients: vec![None, None],
-            peer_topology_hosts: vec!["peer-a".to_string()],
-            peer_admin_caches: vec![Mutex::new(PeerAdminCache::new())],
-            tier_config_reload_workers: Default::default(),
-        };
-        let unreachable_err = unreachable
-            .probe_cross_pool_fence_fleet("topology-a")
-            .await
-            .expect_err("an unreachable member must prevent reconcile capability proof");
-        assert!(unreachable_err.to_string().contains("unreachable"));
-    }
-
-    #[tokio::test]
-    async fn legacy_transition_state_reconcile_single_node_stays_closed_before_local_cas_support() {
-        let notification_sys = NotificationSys {
-            peer_clients: Vec::new(),
-            all_peer_clients: vec![None],
-            peer_topology_hosts: Vec::new(),
-            peer_admin_caches: Vec::new(),
-            tier_config_reload_workers: Default::default(),
-        };
-        let (peers, minimum_version) = notification_sys
-            .probe_cross_pool_fence_fleet("topology-a")
-            .await
-            .expect("a single-node capability probe should complete");
-        assert!(peers.is_empty());
-        assert_eq!(minimum_version, LOCAL_CROSS_POOL_FENCE_POLICY_SUPPORTED_VERSION);
-        let (_, _, _, reconcile_result) = cross_pool_fence_policy_results(peers, minimum_version);
-        assert!(
-            reconcile_result.is_err(),
-            "the current node must not self-authorize reconcile before the conditional writer lands"
-        );
-    }
-
     fn build_props(endpoint: &str) -> ServerProperties {
         ServerProperties {
             endpoint: endpoint.to_string(),

crates/scanner/src/scanner_io.rs

@@ -105,6 +105,12 @@ struct DirtyUsageSnapshot {
     covers_all_pending: bool,
 }
 
+#[derive(Clone, Debug, Default)]
+pub(crate) struct ScannerBucketScanScope {
+    selected_buckets: Option<Arc<HashSet<String>>>,
+    baseline_scan_plan_digest: Option<DataUsageScanPlanDigest>,
+}
+
 pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool {
     matches!(err, StorageError::Io(io) if io.to_string().starts_with(SCANNER_METADATA_CORRUPT_ERROR))
 }
@@ -146,6 +152,7 @@ fn object_lock_config_enabled(config: &ObjectLockConfiguration) -> bool {
 pub struct ScannerBucketScanPlan {
     buckets: Vec<BucketInfo>,
     all_buckets: Arc<Vec<BucketInfo>>,
+    scope: ScannerBucketScanScope,
     digest: DataUsageScanPlanDigest,
     leader_epoch: u64,
     tier_registry_generation: u64,
@@ -732,6 +739,8 @@ mod dirty_usage;
 mod guards;
 mod io_cache;
 mod io_cycle;
+#[cfg(test)]
+use io_cache::{ScannerSetCacheGeneration, prepare_scoped_set_scan};
 pub(crate) use io_cycle::nsscanner_with_storage_status;
 mod io_disk;
 #[cfg(test)]

crates/scanner/src/scanner_io/io_cache.rs

@@ -14,6 +14,93 @@
 /// ScannerIOCache implementation for SetDisks: bucket ordering, worker fan-out, merge, and publish.
 use super::*;
 
+#[derive(Clone, Copy)]
+pub(super) struct ScannerSetCacheGeneration {
+    pub(super) want_cycle: u64,
+    pub(super) leader_epoch: u64,
+    pub(super) tier_registry_generation: u64,
+    pub(super) source: DataUsageCacheSource,
+    pub(super) scan_plan_digest: DataUsageScanPlanDigest,
+}
+
+pub(super) struct PreparedScopedSetScan {
+    pub(super) buckets: Vec<BucketInfo>,
+    pub(super) cache: DataUsageCache,
+}
+
+pub(super) fn prepare_scoped_set_scan(
+    old_cache: &DataUsageCache,
+    set_buckets: &[BucketInfo],
+    all_buckets: &[BucketInfo],
+    scope: &ScannerBucketScanScope,
+    generation: ScannerSetCacheGeneration,
+) -> Option<PreparedScopedSetScan> {
+    let (Some(selected_buckets), Some(baseline_scan_plan_digest)) = (&scope.selected_buckets, scope.baseline_scan_plan_digest)
+    else {
+        return None;
+    };
+    if selected_buckets.is_empty()
+        || !old_cache.info.snapshot_complete
+        || old_cache.info.last_update.is_none()
+        || old_cache.info.name != DATA_USAGE_ROOT
+        || old_cache.info.next_cycle > generation.want_cycle
+        || old_cache.info.leader_epoch != generation.leader_epoch
+        || old_cache.info.tier_registry_generation != Some(generation.tier_registry_generation)
+        || old_cache.info.source != Some(generation.source)
+        || old_cache.info.scan_plan_digest != Some(baseline_scan_plan_digest)
+        || old_cache.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT
+        || old_cache.checked_flatten_complete_scope(DATA_USAGE_ROOT).is_none()
+    {
+        return None;
+    }
+
+    let mut cache = DataUsageCache {
+        info: DataUsageCacheInfo {
+            name: DATA_USAGE_ROOT.to_string(),
+            next_cycle: generation.want_cycle,
+            leader_epoch: generation.leader_epoch,
+            tier_registry_generation: Some(generation.tier_registry_generation),
+            source: Some(generation.source),
+            snapshot_complete: false,
+            scan_plan_digest: Some(generation.scan_plan_digest),
+            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+            lkg_snapshot_complete: true,
+            lkg_next_cycle: Some(old_cache.info.next_cycle),
+            lkg_last_update: old_cache.info.last_update,
+            lkg_leader_epoch: Some(old_cache.info.leader_epoch),
+            lkg_scan_plan_digest: old_cache.info.scan_plan_digest,
+            ..Default::default()
+        },
+        cache: HashMap::new(),
+    };
+    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+    let root_hash = crate::hash_path(DATA_USAGE_ROOT);
+    let mut current_bucket_names = HashSet::with_capacity(all_buckets.len());
+    for bucket in all_buckets {
+        if !current_bucket_names.insert(bucket.name.as_str()) {
+            return None;
+        }
+        if selected_buckets.contains(&bucket.name) {
+            cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+            continue;
+        }
+
+        let bucket_hash = crate::hash_path(&bucket.name);
+        old_cache.find(&bucket.name)?;
+        cache.copy_with_children(old_cache, &bucket_hash, &Some(root_hash.clone()));
+        cache.find(&bucket.name)?;
+    }
+
+    Some(PreparedScopedSetScan {
+        buckets: set_buckets
+            .iter()
+            .filter(|bucket| selected_buckets.contains(&bucket.name))
+            .cloned()
+            .collect(),
+        cache,
+    })
+}
+
 #[async_trait::async_trait]
 impl ScannerIOCache for SetDisks {
     #[tracing::instrument(skip(self, budget, scan_plan, updates))]
@@ -27,8 +114,9 @@ impl ScannerIOCache for SetDisks {
         scan_mode: HealScanMode,
     ) -> Result<()> {
         let ScannerBucketScanPlan {
-            buckets,
+            mut buckets,
             all_buckets,
+            scope,
             digest: scan_plan_digest,
             leader_epoch,
             tier_registry_generation,
@@ -63,26 +151,57 @@ impl ScannerIOCache for SetDisks {
                 "Scanner old data usage cache load failed; rebuilding from bucket caches"
             );
         }
+        let scoped_scan = prepare_scoped_set_scan(
+            &old_cache,
+            &buckets,
+            &all_buckets,
+            &scope,
+            ScannerSetCacheGeneration {
+                want_cycle,
+                leader_epoch,
+                tier_registry_generation,
+                source,
+                scan_plan_digest,
+            },
+        );
+        let mut scoped_cache = scoped_scan.map(|prepared| {
+            buckets = prepared.buckets;
+            prepared.cache
+        });
         if buckets.is_empty() {
             let now = SystemTime::now();
-            let mut cache = DataUsageCache {
-                info: DataUsageCacheInfo {
-                    name: DATA_USAGE_ROOT.to_string(),
-                    next_cycle: want_cycle,
-                    last_update: Some(now),
-                    leader_epoch,
-                    tier_registry_generation: Some(tier_registry_generation),
-                    source: Some(source),
-                    snapshot_complete: true,
-                    scan_plan_digest: Some(scan_plan_digest),
-                    cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-                    ..Default::default()
-                },
-                cache: HashMap::new(),
+            let mut cache = match scoped_cache.take() {
+                Some(cache) => cache,
+                None => {
+                    let mut cache = DataUsageCache {
+                        info: DataUsageCacheInfo {
+                            name: DATA_USAGE_ROOT.to_string(),
+                            next_cycle: want_cycle,
+                            leader_epoch,
+                            tier_registry_generation: Some(tier_registry_generation),
+                            source: Some(source),
+                            scan_plan_digest: Some(scan_plan_digest),
+                            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                            ..Default::default()
+                        },
+                        cache: HashMap::new(),
+                    };
+                    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+                    for bucket in all_buckets.iter() {
+                        cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+                    }
+                    cache
+                }
             };
-            cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
-            for bucket in all_buckets.iter() {
-                cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+            cache.info.last_update = Some(now);
+            cache.info.snapshot_complete = true;
+            cache.info.lkg_snapshot_complete = false;
+            cache.info.lkg_next_cycle = None;
+            cache.info.lkg_last_update = None;
+            cache.info.lkg_leader_epoch = None;
+            cache.info.lkg_scan_plan_digest = None;
+            if cache.find(DATA_USAGE_ROOT).is_none() {
+                cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
             }
             reset_disk_bucket_scan_gauges(&pool_label, &set_label);
             return persist_and_publish_cache_snapshot(
@@ -269,92 +388,102 @@ impl ScannerIOCache for SetDisks {
         record_disk_bucket_scans_active(0, &pool_label, &set_label);
         let _reset_disk_bucket_scan_gauges = DiskBucketScanGaugeReset::new(pool_label.clone(), set_label.clone());
 
-        // Fence a stale set aggregate before copying entries into per-bucket work caches.
-        if old_cache.info.next_cycle <= want_cycle
-            && old_cache.info.leader_epoch <= leader_epoch
-            && old_cache.info.tier_registry_generation != Some(tier_registry_generation)
-        {
-            old_cache.info.scan_plan_digest = None;
-        }
-        let old_lkg = old_cache.info.snapshot_complete.then_some({
-            (
-                old_cache.info.next_cycle,
-                old_cache.info.last_update,
-                old_cache.info.leader_epoch,
-                old_cache.info.scan_plan_digest,
-            )
-        });
-        let prepare_outcome = match old_cache.prepare_for_scan(
-            DATA_USAGE_ROOT,
-            want_cycle,
-            leader_epoch,
-            source,
-            scan_plan_digest,
-            require_cache_source,
-        ) {
-            DataUsageCachePrepareOutcome::RejectedNewerCycle => {
-                cache_cycle_floor.fetch_max(old_cache.info.next_cycle, Ordering::AcqRel);
-                warn!(
-                    target: "rustfs::scanner::io",
-                    event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_IO,
-                    pool = self.pool_index,
-                    set = self.set_index,
-                    cache_name = DATA_USAGE_CACHE_NAME,
-                    requested_cycle = want_cycle,
-                    cached_cycle = old_cache.info.next_cycle,
-                    state = "stale_cycle_rejected",
-                    "Scanner rejected a set cache cycle regression"
-                );
-                return Ok(());
+        let mut cache = if let Some(cache) = scoped_cache.take() {
+            cache
+        } else {
+            // Fence a stale set aggregate before copying entries into per-bucket work caches.
+            if old_cache.info.next_cycle <= want_cycle
+                && old_cache.info.leader_epoch <= leader_epoch
+                && old_cache.info.tier_registry_generation != Some(tier_registry_generation)
+            {
+                old_cache.info.scan_plan_digest = None;
             }
-            DataUsageCachePrepareOutcome::RejectedNewerLeader => {
-                warn!(
-                    target: "rustfs::scanner::io",
-                    event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                    component = LOG_COMPONENT_SCANNER,
-                    subsystem = LOG_SUBSYSTEM_IO,
-                    pool = self.pool_index,
-                    set = self.set_index,
-                    cache_name = DATA_USAGE_CACHE_NAME,
-                    requested_epoch = leader_epoch,
-                    cached_epoch = old_cache.info.leader_epoch,
-                    state = "stale_leader_rejected",
-                    "Scanner rejected work from an older leader epoch"
-                );
-                return Ok(());
-            }
-            outcome => outcome,
-        };
-        if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused)
-            && let Some((cycle, last_update, epoch, digest)) = old_lkg
-        {
-            old_cache.info.lkg_snapshot_complete = true;
-            old_cache.info.lkg_next_cycle = Some(cycle);
-            old_cache.info.lkg_last_update = last_update;
-            old_cache.info.lkg_leader_epoch = Some(epoch);
-            old_cache.info.lkg_scan_plan_digest = digest;
-        }
-
-        let mut cache = DataUsageCache {
-            info: DataUsageCacheInfo {
-                name: DATA_USAGE_ROOT.to_string(),
-                next_cycle: want_cycle,
+            let old_lkg = old_cache.info.snapshot_complete.then_some({
+                (
+                    old_cache.info.next_cycle,
+                    old_cache.info.last_update,
+                    old_cache.info.leader_epoch,
+                    old_cache.info.scan_plan_digest,
+                )
+            });
+            let prepare_outcome = match old_cache.prepare_for_scan(
+                DATA_USAGE_ROOT,
+                want_cycle,
                 leader_epoch,
-                tier_registry_generation: Some(tier_registry_generation),
-                source: Some(source),
-                snapshot_complete: false,
-                scan_plan_digest: Some(scan_plan_digest),
-                cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-                ..Default::default()
-            },
-            cache: HashMap::new(),
+                source,
+                scan_plan_digest,
+                require_cache_source,
+            ) {
+                DataUsageCachePrepareOutcome::RejectedNewerCycle => {
+                    cache_cycle_floor.fetch_max(old_cache.info.next_cycle, Ordering::AcqRel);
+                    warn!(
+                        target: "rustfs::scanner::io",
+                        event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_IO,
+                        pool = self.pool_index,
+                        set = self.set_index,
+                        cache_name = DATA_USAGE_CACHE_NAME,
+                        requested_cycle = want_cycle,
+                        cached_cycle = old_cache.info.next_cycle,
+                        state = "stale_cycle_rejected",
+                        "Scanner rejected a set cache cycle regression"
+                    );
+                    return Ok(());
+                }
+                DataUsageCachePrepareOutcome::RejectedNewerLeader => {
+                    warn!(
+                        target: "rustfs::scanner::io",
+                        event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+                        component = LOG_COMPONENT_SCANNER,
+                        subsystem = LOG_SUBSYSTEM_IO,
+                        pool = self.pool_index,
+                        set = self.set_index,
+                        cache_name = DATA_USAGE_CACHE_NAME,
+                        requested_epoch = leader_epoch,
+                        cached_epoch = old_cache.info.leader_epoch,
+                        state = "stale_leader_rejected",
+                        "Scanner rejected work from an older leader epoch"
+                    );
+                    return Ok(());
+                }
+                outcome => outcome,
+            };
+            if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused)
+                && let Some((cycle, last_update, epoch, digest)) = old_lkg
+            {
+                old_cache.info.lkg_snapshot_complete = true;
+                old_cache.info.lkg_next_cycle = Some(cycle);
+                old_cache.info.lkg_last_update = last_update;
+                old_cache.info.lkg_leader_epoch = Some(epoch);
+                old_cache.info.lkg_scan_plan_digest = digest;
+            }
+
+            let mut cache = DataUsageCache {
+                info: DataUsageCacheInfo {
+                    name: DATA_USAGE_ROOT.to_string(),
+                    next_cycle: want_cycle,
+                    leader_epoch,
+                    tier_registry_generation: Some(tier_registry_generation),
+                    source: Some(source),
+                    snapshot_complete: false,
+                    scan_plan_digest: Some(scan_plan_digest),
+                    cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                    lkg_snapshot_complete: old_cache.info.lkg_snapshot_complete,
+                    lkg_next_cycle: old_cache.info.lkg_next_cycle,
+                    lkg_last_update: old_cache.info.lkg_last_update,
+                    lkg_leader_epoch: old_cache.info.lkg_leader_epoch,
+                    lkg_scan_plan_digest: old_cache.info.lkg_scan_plan_digest,
+                    ..Default::default()
+                },
+                cache: HashMap::new(),
+            };
+            cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+            for bucket in all_buckets.iter() {
+                cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+            }
+            cache
         };
-        cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
-        for bucket in all_buckets.iter() {
-            cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
-        }
 
         let (bucket_tx, bucket_rx) = mpsc::channel::<BucketInfo>(buckets.len());
 
@@ -1257,11 +1386,6 @@ impl ScannerIOCache for SetDisks {
             incomplete_scope.info.snapshot_complete = false;
             incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest);
             incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
-            incomplete_scope.info.lkg_snapshot_complete = old_cache.info.lkg_snapshot_complete;
-            incomplete_scope.info.lkg_next_cycle = old_cache.info.lkg_next_cycle;
-            incomplete_scope.info.lkg_last_update = old_cache.info.lkg_last_update;
-            incomplete_scope.info.lkg_leader_epoch = old_cache.info.lkg_leader_epoch;
-            incomplete_scope.info.lkg_scan_plan_digest = old_cache.info.lkg_scan_plan_digest;
             if let Err(e) = updates.send(incomplete_scope).await {
                 error!(
                     target: "rustfs::scanner::io",

crates/scanner/src/scanner_io/io_cycle.rs

@@ -63,6 +63,41 @@ pub(crate) async fn nsscanner_with_storage_status<S>(
 where
     S: ScannerStorage,
 {
+    let request = ScannerCycleRequest {
+        ctx,
+        budget,
+        updates,
+        want_cycle,
+        leader_epoch,
+        scan_mode,
+        scan_scope: ScannerBucketScanScope::default(),
+    };
+    nsscanner_with_storage_status_scoped(store, request).await
+}
+
+pub(crate) struct ScannerCycleRequest {
+    pub(crate) ctx: CancellationToken,
+    pub(crate) budget: Arc<ScannerCycleBudget>,
+    pub(crate) updates: mpsc::Sender<DataUsageInfo>,
+    pub(crate) want_cycle: u64,
+    pub(crate) leader_epoch: u64,
+    pub(crate) scan_mode: HealScanMode,
+    pub(crate) scan_scope: ScannerBucketScanScope,
+}
+
+pub(crate) async fn nsscanner_with_storage_status_scoped<S>(store: &S, request: ScannerCycleRequest) -> Result<ScannerCycleResult>
+where
+    S: ScannerStorage,
+{
+    let ScannerCycleRequest {
+        ctx,
+        budget,
+        updates,
+        want_cycle,
+        leader_epoch,
+        scan_mode,
+        scan_scope,
+    } = request;
     let child_token = ctx.child_token();
     let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);
 
@@ -280,6 +315,7 @@ where
         let scan_plan = ScannerBucketScanPlan {
             buckets: set_buckets,
             all_buckets: Arc::clone(&all_buckets),
+            scope: scan_scope.clone(),
             digest: scan_plan_digest,
             leader_epoch,
             tier_registry_generation,

crates/scanner/src/scanner_io/tests.rs

@@ -765,6 +765,155 @@ fn bucket_usage_scan_order_prioritizes_dirty_buckets() {
     assert_eq!(names, vec!["dirty", "missing", "cached"]);
 }
 
+fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsageScanPlanDigest) -> DataUsageCache {
+    let mut cache = DataUsageCache {
+        info: DataUsageCacheInfo {
+            name: DATA_USAGE_ROOT.to_string(),
+            next_cycle: 7,
+            last_update: Some(SystemTime::now()),
+            leader_epoch: 11,
+            source: Some(DataUsageCacheSource::new(1, 2)),
+            snapshot_complete: true,
+            scan_plan_digest: Some(scan_plan_digest),
+            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+            tier_registry_generation: Some(13),
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+    for (bucket, size) in buckets {
+        cache.replace(
+            bucket,
+            DATA_USAGE_ROOT,
+            DataUsageEntry {
+                size: *size,
+                objects: 1,
+                ..Default::default()
+            },
+        );
+    }
+    cache
+}
+
+#[test]
+fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
+    let baseline_digest = DataUsageScanPlanDigest([1; 32]);
+    let current_digest = DataUsageScanPlanDigest([2; 32]);
+    let mut old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20), ("deleted", 30)], baseline_digest);
+    old_cache.replace(
+        "stable/prefix",
+        "stable",
+        DataUsageEntry {
+            size: 5,
+            objects: 1,
+            ..Default::default()
+        },
+    );
+    let all_buckets = vec![bucket_info("stable"), bucket_info("dirty")];
+    let selected_buckets = Arc::new(HashSet::from(["dirty".to_string(), "deleted".to_string()]));
+
+    let prepared = prepare_scoped_set_scan(
+        &old_cache,
+        &all_buckets,
+        &all_buckets,
+        &ScannerBucketScanScope {
+            selected_buckets: Some(selected_buckets),
+            baseline_scan_plan_digest: Some(baseline_digest),
+        },
+        ScannerSetCacheGeneration {
+            want_cycle: 8,
+            leader_epoch: 11,
+            tier_registry_generation: 13,
+            source: DataUsageCacheSource::new(1, 2),
+            scan_plan_digest: current_digest,
+        },
+    )
+    .expect("complete matching set cache should support a scoped scan");
+
+    assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(), ["dirty"]);
+    let stable = prepared
+        .cache
+        .checked_flatten("stable")
+        .expect("unselected bucket subtree should be retained");
+    assert_eq!((stable.size, stable.objects), (15, 2));
+    assert_eq!(prepared.cache.find("dirty").map(|entry| (entry.size, entry.objects)), Some((0, 0)));
+    assert!(prepared.cache.find("deleted").is_none());
+    assert_eq!(prepared.cache.info.scan_plan_digest, Some(current_digest));
+    assert_eq!(prepared.cache.info.next_cycle, 8);
+    assert!(!prepared.cache.info.snapshot_complete);
+    assert!(prepared.cache.info.lkg_snapshot_complete);
+    assert_eq!(prepared.cache.info.lkg_next_cycle, Some(7));
+    assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest));
+}
+
+#[test]
+fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() {
+    let baseline_digest = DataUsageScanPlanDigest([3; 32]);
+    let old_cache = complete_set_usage_cache(&[("stable", 10)], baseline_digest);
+    let all_buckets = vec![bucket_info("stable"), bucket_info("new")];
+
+    assert!(
+        prepare_scoped_set_scan(
+            &old_cache,
+            &all_buckets,
+            &all_buckets,
+            &ScannerBucketScanScope {
+                selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
+                baseline_scan_plan_digest: Some(baseline_digest),
+            },
+            ScannerSetCacheGeneration {
+                want_cycle: 8,
+                leader_epoch: 11,
+                tier_registry_generation: 13,
+                source: DataUsageCacheSource::new(1, 2),
+                scan_plan_digest: DataUsageScanPlanDigest([4; 32]),
+            },
+        )
+        .is_none()
+    );
+}
+
+#[test]
+fn scoped_set_scan_requires_an_exact_complete_baseline() {
+    let baseline_digest = DataUsageScanPlanDigest([5; 32]);
+    let all_buckets = vec![bucket_info("dirty")];
+    let scope = ScannerBucketScanScope {
+        selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
+        baseline_scan_plan_digest: Some(baseline_digest),
+    };
+    let generation = ScannerSetCacheGeneration {
+        want_cycle: 8,
+        leader_epoch: 11,
+        tier_registry_generation: 13,
+        source: DataUsageCacheSource::new(1, 2),
+        scan_plan_digest: DataUsageScanPlanDigest([6; 32]),
+    };
+
+    let mut incomplete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    incomplete.info.snapshot_complete = false;
+    assert!(prepare_scoped_set_scan(&incomplete, &all_buckets, &all_buckets, &scope, generation).is_none());
+
+    let mut not_durable = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    not_durable.info.last_update = None;
+    assert!(prepare_scoped_set_scan(&not_durable, &all_buckets, &all_buckets, &scope, generation).is_none());
+
+    let mut wrong_digest = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    wrong_digest.info.scan_plan_digest = Some(DataUsageScanPlanDigest([7; 32]));
+    assert!(prepare_scoped_set_scan(&wrong_digest, &all_buckets, &all_buckets, &scope, generation).is_none());
+
+    let empty_scope = ScannerBucketScanScope {
+        selected_buckets: Some(Arc::new(HashSet::new())),
+        baseline_scan_plan_digest: Some(baseline_digest),
+    };
+    let complete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &empty_scope, generation).is_none());
+
+    let mut future_cache = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
+    future_cache.info.next_cycle = generation.want_cycle.saturating_add(1);
+    assert!(prepare_scoped_set_scan(&future_cache, &all_buckets, &all_buckets, &scope, generation).is_none());
+}
+
 #[test]
 fn record_set_scan_failure_preserves_first_error() {
     let mut first = None;

rustfs/src/version.rs

@@ -14,8 +14,6 @@
 
 use const_str::concat;
 use shadow_rs::shadow;
-use std::path::Path;
-use std::process::Command;
 
 shadow!(build);
 
@@ -47,10 +45,6 @@ pub const DISPLAY_VERSION: &str = {
 
 type VersionParseResult = Result<(u32, u32, u32, Option<String>), Box<dyn std::error::Error>>;
 
-fn build_version_override() -> Option<&'static str> {
-    BUILD_VERSION_OVERRIDE.filter(|version| !version.is_empty())
-}
-
 fn version_ref(version: &str) -> String {
     if version.starts_with("refs/tags/") || version.starts_with('@') {
         version.to_string()
@@ -61,91 +55,7 @@ fn version_ref(version: &str) -> String {
 
 #[allow(clippy::const_is_empty)]
 pub fn get_version() -> String {
-    if let Some(version) = build_version_override() {
-        return version_ref(version);
-    }
-
-    // Get the latest tag
-    if let Ok(latest_tag) = get_latest_tag() {
-        // Check if current commit is newer than the latest tag
-        if is_head_newer_than_tag(&latest_tag) {
-            // If current commit is newer, increment the version number
-            if let Ok(new_version) = increment_version(&latest_tag) {
-                return format!("refs/tags/{new_version}");
-            }
-        }
-
-        // If current commit is the latest tag, or version increment failed, return current tag
-        return format!("refs/tags/{latest_tag}");
-    }
-
-    // If no tag exists, use original logic
-    if !build::TAG.is_empty() {
-        format!("refs/tags/{}", build::TAG)
-    } else if !build::SHORT_COMMIT.is_empty() {
-        format!("@{}", build::SHORT_COMMIT)
-    } else {
-        format!("refs/tags/{}", build::PKG_VERSION)
-    }
-}
-
-/// Get the latest git tag
-fn get_latest_tag() -> Result<String, Box<dyn std::error::Error>> {
-    let output = Command::new("git").args(["describe", "--tags", "--abbrev=0"]).output()?;
-
-    if output.status.success() {
-        let tag = String::from_utf8(output.stdout)?;
-        Ok(tag.trim().to_string())
-    } else {
-        Err("Failed to get latest tag".into())
-    }
-}
-
-/// Check if current HEAD is newer than specified tag
-fn is_head_newer_than_tag(tag: &str) -> bool {
-    is_head_newer_than_tag_in(Path::new("."), tag)
-}
-
-fn is_head_newer_than_tag_in(repo: &Path, tag: &str) -> bool {
-    let head = Command::new("git").current_dir(repo).args(["rev-parse", "HEAD"]).output();
-    let tag_commit = Command::new("git")
-        .current_dir(repo)
-        .args(["rev-list", "-n", "1", tag])
-        .output();
-
-    let (Ok(head), Ok(tag_commit)) = (head, tag_commit) else {
-        return false;
-    };
-
-    if !head.status.success() || !tag_commit.status.success() || head.stdout == tag_commit.stdout {
-        return false;
-    }
-
-    let output = Command::new("git")
-        .current_dir(repo)
-        .args(["merge-base", "--is-ancestor", tag, "HEAD"])
-        .output();
-
-    match output {
-        Ok(result) => result.status.success(),
-        Err(_) => false,
-    }
-}
-
-/// Increment version number (increase patch version)
-fn increment_version(version: &str) -> Result<String, Box<dyn std::error::Error>> {
-    // Parse version number, e.g. "1.0.0-alpha.19" -> (1, 0, 0, Some("alpha.19"))
-    let (major, minor, patch, pre_release) = parse_version(version)?;
-
-    // If there's a pre-release identifier, increment the pre-release version number
-    if let Some(pre) = pre_release
-        && let Some(new_pre) = increment_pre_release(&pre)
-    {
-        return Ok(format!("{major}.{minor}.{patch}-{new_pre}"));
-    }
-
-    // Otherwise increment patch version number
-    Ok(format!("{major}.{minor}.{}", patch + 1))
+    version_ref(DISPLAY_VERSION)
 }
 
 /// Parse version number
@@ -166,28 +76,6 @@ pub fn parse_version(version: &str) -> VersionParseResult {
     Ok((major, minor, patch, pre_release))
 }
 
-/// Increment pre-release version number
-fn increment_pre_release(pre_release: &str) -> Option<String> {
-    // Handle pre-release versions like "alpha.19"
-    let parts: Vec<&str> = pre_release.split('.').collect();
-    if parts.len() == 2
-        && let Ok(num) = parts[1].parse::<u32>()
-    {
-        return Some(format!("{}.{}", parts[0], num + 1));
-    }
-
-    // Handle pre-release versions like "alpha19"
-    if let Some(pos) = pre_release.rfind(|c: char| c.is_alphabetic()) {
-        let prefix = &pre_release[..=pos];
-        let suffix = &pre_release[pos + 1..];
-        if let Ok(num) = suffix.parse::<u32>() {
-            return Some(format!("{prefix}{}", num + 1));
-        }
-    }
-
-    None
-}
-
 /// Clean version string - removes common prefixes
 pub fn clean_version(version: &str) -> String {
     version
@@ -284,34 +172,6 @@ mod tests {
     use super::*;
     use tracing::debug;
 
-    fn run_git(repo: &Path, args: &[&str]) {
-        let status = Command::new("git").current_dir(repo).args(args).status().unwrap();
-        assert!(status.success(), "git command failed: git {}", args.join(" "));
-    }
-
-    #[test]
-    fn test_is_head_newer_than_tag_requires_strict_descendant() {
-        let repo = tempfile::tempdir().unwrap();
-        run_git(repo.path(), &["init", "--quiet"]);
-        run_git(repo.path(), &["config", "user.name", "RustFS Tests"]);
-        run_git(repo.path(), &["config", "user.email", "rustfs@example.com"]);
-        run_git(repo.path(), &["commit", "--allow-empty", "--quiet", "-m", "tagged commit"]);
-        run_git(repo.path(), &["tag", "--annotate", "1.2.3", "--message", "1.2.3"]);
-
-        assert!(!is_head_newer_than_tag_in(repo.path(), "1.2.3"));
-
-        run_git(repo.path(), &["commit", "--allow-empty", "--quiet", "-m", "newer commit"]);
-
-        assert!(is_head_newer_than_tag_in(repo.path(), "1.2.3"));
-    }
-
-    #[test]
-    fn build_version_override_is_used_for_current_version_when_set() {
-        if let Some(version) = build_version_override() {
-            assert_eq!(get_version(), version_ref(version));
-        }
-    }
-
     #[test]
     fn version_ref_keeps_existing_ref_prefixes() {
         assert_eq!(version_ref("1.2.3"), "refs/tags/1.2.3");
@@ -319,6 +179,11 @@ mod tests {
         assert_eq!(version_ref("@abc123"), "@abc123");
     }
 
+    #[test]
+    fn get_version_uses_build_metadata() {
+        assert_eq!(get_version(), version_ref(DISPLAY_VERSION));
+    }
+
     #[test]
     fn test_parse_version() {
         // Test standard version parsing
@@ -336,27 +201,6 @@ mod tests {
         assert_eq!(pre_release, Some("alpha.19".to_string()));
     }
 
-    #[test]
-    fn test_increment_pre_release() {
-        // Test alpha.19 -> alpha.20
-        assert_eq!(increment_pre_release("alpha.19"), Some("alpha.20".to_string()));
-
-        // Test beta.5 -> beta.6
-        assert_eq!(increment_pre_release("beta.5"), Some("beta.6".to_string()));
-
-        // Test unparsable case
-        assert_eq!(increment_pre_release("unknown"), None);
-    }
-
-    #[test]
-    fn test_increment_version() {
-        // Test pre-release version increment
-        assert_eq!(increment_version("1.0.0-alpha.19").unwrap(), "1.0.0-alpha.20");
-
-        // Test standard version increment
-        assert_eq!(increment_version("1.0.0").unwrap(), "1.0.1");
-    }
-
     #[test]
     fn test_version_format() {
         // Test if version format starts with refs/tags/