fix(tier): probe legacy transition version state (#7138)

crates/ecstore/src/services/tier/test_util.rs

@@ -701,7 +701,7 @@ impl WarmBackend for MockWarmBackend {
         Ok(version)
     }
 
-    async fn get(&self, object: &str, _rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
+    async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
         self.precondition().await?;
         let barrier = self.inner.get_barrier.lock().await.take();
         if let Some(barrier) = barrier {
@@ -719,6 +719,9 @@ impl WarmBackend for MockWarmBackend {
         let Some(stored) = objects.get(object) else {
             return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "mock object not found"));
         };
+        if !rv.is_empty() && stored.remote_version_id != rv {
+            return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "NoSuchVersion"));
+        }
         let bytes = &stored.bytes;
 
         let start = opts.start_offset.max(0) as usize;

crates/ecstore/src/services/tier/tier.rs

@@ -2346,6 +2346,10 @@ impl WarmBackend for SharedWarmBackendProxy {
         self.0.probe_transition_candidate(object).await
     }
 
+    async fn probe_transition_version(&self, object: &str, remote_version_id: &str) -> io::Result<TransitionCandidateProbe> {
+        self.0.probe_transition_version(object, remote_version_id).await
+    }
+
     async fn in_use(&self) -> io::Result<bool> {
         self.0.in_use().await
     }
@@ -2458,6 +2462,15 @@ impl TierOperationLease {
         Ok(())
     }
 
+    pub(crate) async fn probe_transition_version(
+        &self,
+        object: &str,
+        remote_version_id: &str,
+    ) -> io::Result<TransitionCandidateProbe> {
+        self.validate_remote_version_id(remote_version_id)?;
+        self.inner.driver.probe_transition_version(object, remote_version_id).await
+    }
+
     pub(crate) fn is_current_generation(&self) -> bool {
         lock_unpoisoned(&self.runtime)
             .generations

crates/ecstore/src/services/tier/warm_backend.rs

@@ -40,6 +40,7 @@ use rustfs_s3_client::credentials::{Credentials, SignatureType, Static, Value};
 use rustfs_s3_client::transition_api::{BucketLookupType, Options, TransitionClient, TransitionCore};
 use rustfs_s3_client::{
     admin_handler_utils::AdminError,
+    api_error_response::to_error_response,
     api_put_object::{AdvancedPutOptions, PutObjectOptions},
     transition_api::{ReadCloser, ReaderImpl},
 };
@@ -48,11 +49,14 @@ use rustfs_utils::egress::validate_outbound_url;
 use rustfs_utils::http::headers::{
     CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, EXPIRES, HeaderExt as _,
 };
-use s3s::dto::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode, ReplicationStatus};
 use s3s::header::{
     X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_REPLICATION_STATUS,
     X_AMZ_STORAGE_CLASS,
 };
+use s3s::{
+    S3ErrorCode,
+    dto::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode, ReplicationStatus},
+};
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
@@ -141,6 +145,42 @@ pub trait WarmBackend {
     async fn probe_transition_candidate(&self, _object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
         Ok(TransitionCandidateProbe::Unsupported)
     }
+    async fn probe_transition_version(
+        &self,
+        object: &str,
+        remote_version_id: &str,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        if remote_version_id.is_empty() {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "an exact tier probe requires a remote version ID",
+            ));
+        }
+        self.validate_remote_version_id(remote_version_id)?;
+        match self
+            .get(
+                object,
+                remote_version_id,
+                WarmBackendGetOpts {
+                    start_offset: 0,
+                    length: 1,
+                },
+            )
+            .await
+        {
+            Ok(_) => Ok(TransitionCandidateProbe::VersionedPresent(remote_version_id.to_string())),
+            Err(err) if matches!(to_error_response(&err).code, S3ErrorCode::InvalidRange) => {
+                Ok(TransitionCandidateProbe::VersionedPresent(remote_version_id.to_string()))
+            }
+            Err(err)
+                if err.kind() == std::io::ErrorKind::NotFound
+                    || matches!(to_error_response(&err).code, S3ErrorCode::NoSuchKey | S3ErrorCode::NoSuchVersion) =>
+            {
+                Ok(TransitionCandidateProbe::Missing)
+            }
+            Err(err) => Err(err),
+        }
+    }
     async fn in_use(&self) -> Result<bool, std::io::Error>;
 }
 
@@ -437,6 +477,17 @@ impl WarmBackend for MeteredWarmBackend {
         Self::record(TierRequestOperation::Probe, result)
     }
 
+    async fn probe_transition_version(
+        &self,
+        object: &str,
+        remote_version_id: &str,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        Self::record(
+            TierRequestOperation::Probe,
+            self.inner.probe_transition_version(object, remote_version_id).await,
+        )
+    }
+
     async fn in_use(&self) -> Result<bool, std::io::Error> {
         Self::record(TierRequestOperation::InUse, self.inner.in_use().await)
     }

crates/ecstore/src/services/tier/warm_backend_s3.rs

@@ -529,6 +529,10 @@ mod tests {
                 "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 63\r\nConnection: close\r\n\r\n<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>",
                 "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 66\r\nConnection: close\r\n\r\n<Error><Code>NoSuchObject</Code><Message>missing</Message></Error>",
                 "HTTP/1.1 403 Forbidden\r\nContent-Type: application/xml\r\nContent-Length: 65\r\nConnection: close\r\n\r\n<Error><Code>AccessDenied</Code><Message>denied</Message></Error>",
+                "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 63\r\nConnection: close\r\n\r\n<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>",
+                "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Type: application/xml\r\nContent-Length: 72\r\nConnection: close\r\n\r\n<Error><Code>InvalidRange</Code><Message>empty version</Message></Error>",
+                "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 67\r\nConnection: close\r\n\r\n<Error><Code>NoSuchVersion</Code><Message>missing</Message></Error>",
+                "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 63\r\nConnection: close\r\n\r\n<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>",
             ];
             let mut requests = Vec::new();
             for response in responses {
@@ -622,15 +626,52 @@ mod tests {
             .await
             .expect_err("an authorization failure must not be mistaken for a missing key");
         assert_eq!(to_error_response(&err).code, S3ErrorCode::AccessDenied);
+        assert_eq!(
+            backend
+                .probe_transition_candidate("delete-marker-hidden")
+                .await
+                .expect("a current delete marker should hide the data version"),
+            TransitionCandidateProbe::Missing
+        );
+        assert_eq!(
+            backend
+                .probe_transition_version("delete-marker-hidden", "historical-version")
+                .await
+                .expect("the stored historical version should be probed exactly"),
+            TransitionCandidateProbe::VersionedPresent("historical-version".to_string())
+        );
+        assert_eq!(
+            backend
+                .probe_transition_version("delete-marker-hidden", "missing-version")
+                .await
+                .expect("a missing exact version should be classified"),
+            TransitionCandidateProbe::Missing
+        );
+        assert_eq!(
+            backend
+                .probe_transition_version("missing-object", "historical-version")
+                .await
+                .expect("a missing key for an exact version probe should be classified"),
+            TransitionCandidateProbe::Missing
+        );
 
         let requests = fixture.await.expect("candidate fixture should join");
-        for request in requests {
+        for request in &requests[..6] {
             let request = request.to_ascii_lowercase();
             assert!(request.starts_with("get /bucket/"), "candidate discovery must use object GET");
             assert!(request.contains("\r\nrange: bytes=0-0\r\n"));
             assert!(!request.contains("?versioning"));
             assert!(!request.contains("?versions"));
         }
+        for request in &requests[6..] {
+            let request = request.to_ascii_lowercase();
+            assert!(request.starts_with("get /bucket/"), "exact discovery must use object GET");
+            assert!(request.contains("\r\nrange: bytes=0-0\r\n"));
+        }
+        assert!(!requests[5].to_ascii_lowercase().contains("versionid="));
+        assert!(requests[6].to_ascii_lowercase().contains("?versionid=historical-version"));
+        assert!(requests[7].to_ascii_lowercase().contains("?versionid=missing-version"));
+        assert!(requests[8].to_ascii_lowercase().contains("?versionid=historical-version"));
     }
 
     fn list_versions(versions: &[(&str, &str)], delete_markers: &[(&str, &str)], is_truncated: bool) -> ListVersionsResult {