test(ecstore): narrow barrier re-export cfgs

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

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/set_disk/mod.rs

@@ -876,7 +876,7 @@ pub use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
 pub(crate) use ops::object::DeleteObjectCommitBarrier;
 #[cfg(any(test, feature = "test-util"))]
 pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier;
-#[cfg(test)]
+#[cfg(all(test, feature = "test-util"))]
 pub(crate) use ops::object::TransitionUploadedCommitBarrier as SetDiskTransitionUploadedCommitBarrier;
 pub(crate) use ops::object::body_cache_plaintext_len;
 #[cfg(all(test, feature = "test-util"))]

crates/ecstore/src/store/mod.rs

@@ -425,7 +425,7 @@ pub(crate) mod init_format;
 pub(crate) mod list_objects;
 mod multipart;
 mod object;
-#[cfg(any(test, feature = "test-util"))]
+#[cfg(feature = "test-util")]
 pub use object::DeleteAfterObjectLockSnapshotBarrier;
 pub(crate) use object::{
     DecommissionFixedReadAnchor, ObjectLockDiagGuard, RemoteTuplePublicationCommitGuard, RemoteTuplePublicationFence,

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/