feat(tier): probe transition candidates from providers (#5112)

* feat(tier): add transition candidate probe contract

Add a fail-closed WarmBackend probe contract for provider-authoritative transition candidate state. Default providers report Unsupported, while the shared mock backend can now model missing, unversioned, and exact-version candidates for follow-up recovery tests.

This is a forward-compatible foundation for #1352/#1358 recovery work and does not change production cleanup behavior.

Co-Authored-By: heihutu <heihutu@gmail.com>

* feat(tier): probe transition candidates from providers

Implement provider-authoritative transition candidate probing for S3-family warm backends by querying ListObjectVersions with exact-key filtering and fail-closed classification for delete markers, multiple versions, truncation, and unknown versioning state.

This keeps non-S3 providers on the default Unsupported probe result and forwards MinIO, RustFS, and R2 through the S3 probe implementation.

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-22 14:13:32 +08:00
committed by GitHub
parent daca7294c7
commit 31dc78eab0
9 changed files with 484 additions and 107 deletions
+96 -1
View File
@@ -74,7 +74,9 @@ use crate::disk::format::FormatV3;
use crate::disk::{DiskAPI, DiskOption, FORMAT_CONFIG_FILE, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE, new_disk};
use crate::services::tier::tier::TierConfigMgr;
use crate::services::tier::tier_config::{TierConfig, TierMinIO, TierType};
use crate::services::tier::warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options};
use crate::services::tier::warm_backend::{
TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options,
};
use rustfs_filemeta::FileMeta;
use rustfs_utils::path::path_join_buf;
@@ -142,6 +144,7 @@ pub enum MockWarmOp {
Put { object: String },
Get { object: String },
Remove { object: String },
Probe { object: String },
ExternalRemove { object: String },
InUse,
}
@@ -444,6 +447,11 @@ impl MockWarmBackend {
self.inner.remove_versions.lock().await.clone()
}
/// Return the provider-authoritative view of a remote transition candidate.
pub async fn probe_transition_candidate_state(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
self.probe_transition_candidate(object).await
}
/// Number of `get` calls recorded — useful to assert restore reads hit the
/// local copy rather than the remote tier.
pub async fn get_count(&self) -> usize {
@@ -725,6 +733,23 @@ impl WarmBackend for MockWarmBackend {
self.remove(object, rv).await
}
async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
self.precondition().await?;
self.record(MockWarmOp::Probe {
object: object.to_string(),
})
.await;
let objects = self.inner.objects.lock().await;
let Some(stored) = objects.get(object) else {
return Ok(TransitionCandidateProbe::Missing);
};
if stored.remote_version_id.is_empty() {
Ok(TransitionCandidateProbe::UnversionedPresent)
} else {
Ok(TransitionCandidateProbe::VersionedPresent(stored.remote_version_id.clone()))
}
}
async fn in_use(&self) -> Result<bool, std::io::Error> {
self.precondition().await?;
self.record(MockWarmOp::InUse).await;
@@ -909,3 +934,73 @@ pub async fn wait_for_free_version_absence(disk_path: &Path, bucket: &str, objec
tokio::time::sleep(POLL_INTERVAL).await;
}
}
#[cfg(test)]
mod tests {
use super::*;
use bytes::Bytes;
#[tokio::test]
async fn mock_probe_distinguishes_missing_unversioned_and_versioned_candidates() {
let backend = MockWarmBackend::new();
assert_eq!(
backend
.probe_transition_candidate_state("missing")
.await
.expect("probe missing candidate"),
TransitionCandidateProbe::Missing
);
backend.set_put_remote_version(Some(String::new())).await;
backend
.put("unversioned", ReaderImpl::Body(Bytes::new()), 0)
.await
.expect("put unversioned candidate");
assert_eq!(
backend
.probe_transition_candidate_state("unversioned")
.await
.expect("probe unversioned candidate"),
TransitionCandidateProbe::UnversionedPresent
);
let remote_version = Uuid::new_v4().to_string();
backend.set_put_remote_version(Some(remote_version.clone())).await;
backend
.put("versioned", ReaderImpl::Body(Bytes::new()), 0)
.await
.expect("put versioned candidate");
assert_eq!(
backend
.probe_transition_candidate_state("versioned")
.await
.expect("probe versioned candidate"),
TransitionCandidateProbe::VersionedPresent(remote_version)
);
assert_eq!(
backend
.op_log()
.await
.into_iter()
.filter(|op| matches!(op, MockWarmOp::Probe { .. }))
.count(),
3
);
}
#[tokio::test]
async fn mock_probe_preserves_fault_fail_closed_behavior() {
let backend = MockWarmBackend::new();
backend.set_reject_credentials(true).await;
let err = backend
.probe_transition_candidate("remote-object")
.await
.expect_err("credential rejection must fail the authoritative probe");
assert_eq!(err.kind(), std::io::ErrorKind::PermissionDenied);
assert!(backend.op_log().await.is_empty());
}
}
+5 -1
View File
@@ -53,7 +53,7 @@ use crate::services::tier::{
tier_admin::TierCreds,
tier_config::{TierConfig, TierType, TierWasabi},
tier_handlers::{ERR_TIER_ALREADY_EXISTS, ERR_TIER_NAME_NOT_UPPERCASE, ERR_TIER_NOT_FOUND, ERR_TIER_RESERVED_NAME},
warm_backend::{WarmBackend, check_warm_backend, new_warm_backend},
warm_backend::{TransitionCandidateProbe, WarmBackend, check_warm_backend, new_warm_backend},
};
use crate::storage_api_contracts::{
bucket::BucketOperations,
@@ -1113,6 +1113,10 @@ impl WarmBackend for SharedWarmBackendProxy {
self.0.remove_exact(object, rv).await
}
async fn probe_transition_candidate(&self, object: &str) -> io::Result<TransitionCandidateProbe> {
self.0.probe_transition_candidate(object).await
}
async fn in_use(&self) -> io::Result<bool> {
self.0.in_use().await
}
@@ -64,6 +64,15 @@ pub struct WarmBackendGetOpts {
pub length: i64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum TransitionCandidateProbe {
Missing,
UnversionedPresent,
VersionedPresent(String),
Ambiguous,
Unsupported,
}
#[async_trait::async_trait]
pub trait WarmBackend {
async fn validate(&self) -> Result<(), std::io::Error> {
@@ -101,6 +110,9 @@ pub trait WarmBackend {
}
self.remove(object, rv).await
}
async fn probe_transition_candidate(&self, _object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
Ok(TransitionCandidateProbe::Unsupported)
}
async fn in_use(&self) -> Result<bool, std::io::Error>;
}
@@ -615,6 +627,22 @@ mod tests {
assert_eq!(removes.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn default_transition_candidate_probe_is_unsupported() {
let backend = RejectingValidationBackend {
validations: Arc::new(AtomicUsize::new(0)),
puts: Arc::new(AtomicUsize::new(0)),
removes: Arc::new(AtomicUsize::new(0)),
};
let probe = backend
.probe_transition_candidate("remote-object")
.await
.expect("default candidate probe should be a safe capability response");
assert_eq!(probe, TransitionCandidateProbe::Unsupported);
}
#[tokio::test]
async fn check_warm_backend_removes_exact_probe_when_versioning_drifts() {
let gets = Arc::new(AtomicUsize::new(0));
@@ -29,7 +29,7 @@ use crate::client::{
};
use crate::services::tier::{
tier_config::TierMinIO,
warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options},
warm_backend_s3::WarmBackendS3,
};
use tracing::warn;
@@ -126,6 +126,10 @@ impl WarmBackend for WarmBackendMinIO {
self.0.remove(object, rv).await
}
async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
self.0.probe_transition_candidate(object).await
}
async fn in_use(&self) -> Result<bool, std::io::Error> {
self.0.in_use().await
}
@@ -29,7 +29,7 @@ use crate::client::{
};
use crate::services::tier::{
tier_config::TierR2,
warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options},
warm_backend_s3::WarmBackendS3,
};
use tracing::warn;
@@ -126,6 +126,10 @@ impl WarmBackend for WarmBackendR2 {
self.0.remove(object, rv).await
}
async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
self.0.probe_transition_candidate(object).await
}
async fn in_use(&self) -> Result<bool, std::io::Error> {
self.0.in_use().await
}
@@ -29,7 +29,7 @@ use crate::client::{
};
use crate::services::tier::{
tier_config::TierRustFS,
warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options},
warm_backend_s3::WarmBackendS3,
};
@@ -123,6 +123,10 @@ impl WarmBackend for WarmBackendRustFS {
self.0.remove(object, rv).await
}
async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
self.0.probe_transition_candidate(object).await
}
async fn in_use(&self) -> Result<bool, std::io::Error> {
self.0.in_use().await
}
@@ -24,8 +24,10 @@ use url::Url;
use crate::client::{
api_get_options::GetObjectOptions,
api_list::ListObjectsOptions,
api_put_object::PutObjectOptions,
api_remove::{RemoveObjectOptions, RemoveObjectResult},
api_s3_datatypes::ListVersionsResult,
credentials::{Credentials, SignatureType, Static, Value},
transition_api::{BucketLookupType, Options, TransitionClient, TransitionCore},
transition_api::{ReadCloser, ReaderImpl},
@@ -34,11 +36,12 @@ use crate::error::ErrorResponse;
use crate::error::error_resp_to_object_err;
use crate::services::tier::{
tier_config::TierS3,
warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options},
};
use http::HeaderMap;
use rustfs_utils::egress::validate_outbound_url;
use rustfs_utils::path::SLASH_SEPARATOR;
use s3s::dto::BucketVersioningStatus;
pub struct WarmBackendS3 {
pub client: Arc<TransitionClient>,
@@ -48,6 +51,27 @@ pub struct WarmBackendS3 {
pub storage_class: String,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RemoteBucketVersioning {
Disabled,
Suspended,
Enabled,
}
fn remote_bucket_versioning_from_status(status: Option<&str>) -> Result<RemoteBucketVersioning, std::io::Error> {
Ok(match status {
Some(BucketVersioningStatus::ENABLED) => RemoteBucketVersioning::Enabled,
Some(BucketVersioningStatus::SUSPENDED) => RemoteBucketVersioning::Suspended,
Some(status) => {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("remote tier bucket returned unsupported versioning status {status}"),
));
}
None => RemoteBucketVersioning::Disabled,
})
}
impl WarmBackendS3 {
pub async fn new(conf: &TierS3, _tier: &str) -> Result<Self, std::io::Error> {
Self::new_with_bucket_lookup(conf, BucketLookupType::BucketLookupAuto, "s3").await
@@ -159,11 +183,57 @@ impl WarmBackendS3 {
let (_, headers, reader) = self.core.get_object(&self.bucket, &self.get_dest(object), &gopts).await?;
Ok((headers, reader))
}
async fn remote_bucket_versioning(&self) -> Result<RemoteBucketVersioning, std::io::Error> {
let config = self.client.get_bucket_versioning(&self.bucket).await?;
remote_bucket_versioning_from_status(config.status.as_ref().map(|status| status.as_str()))
}
async fn list_transition_candidate_versions(&self, object: &str) -> Result<ListVersionsResult, std::io::Error> {
let mut opts = ListObjectsOptions::default();
opts.set("prefix", &self.get_dest(object));
opts.set("max-keys", "2");
self.client.list_object_versions_query(&self.bucket, &opts, "", "", "").await
}
}
fn classify_transition_candidate_versions(
remote_object: &str,
bucket_versioning: RemoteBucketVersioning,
versions: &ListVersionsResult,
) -> TransitionCandidateProbe {
if versions.is_truncated {
return TransitionCandidateProbe::Ambiguous;
}
if versions.delete_markers.iter().any(|marker| marker.key == remote_object) {
return TransitionCandidateProbe::Ambiguous;
}
let mut exact_versions = versions.versions.iter().filter(|version| version.key == remote_object);
let Some(version) = exact_versions.next() else {
return TransitionCandidateProbe::Missing;
};
if exact_versions.next().is_some() {
return TransitionCandidateProbe::Ambiguous;
}
match bucket_versioning {
RemoteBucketVersioning::Disabled => TransitionCandidateProbe::UnversionedPresent,
RemoteBucketVersioning::Suspended if version.version_id == "null" => {
TransitionCandidateProbe::VersionedPresent(version.version_id.clone())
}
RemoteBucketVersioning::Suspended | RemoteBucketVersioning::Enabled if !version.version_id.is_empty() => {
TransitionCandidateProbe::VersionedPresent(version.version_id.clone())
}
RemoteBucketVersioning::Suspended | RemoteBucketVersioning::Enabled => TransitionCandidateProbe::Ambiguous,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::client::api_s3_datatypes::{ListVersionsResult, Version};
#[tokio::test]
async fn new_rejects_loopback_endpoint_before_network_setup() {
@@ -181,6 +251,116 @@ mod tests {
Err(err) => assert!(err.to_string().contains("not allowed")),
}
}
fn list_versions(versions: &[(&str, &str)], delete_markers: &[(&str, &str)], is_truncated: bool) -> ListVersionsResult {
ListVersionsResult {
versions: versions
.iter()
.map(|(key, version_id)| Version {
key: (*key).to_string(),
version_id: (*version_id).to_string(),
..Default::default()
})
.collect(),
delete_markers: delete_markers
.iter()
.map(|(key, version_id)| Version {
key: (*key).to_string(),
version_id: (*version_id).to_string(),
..Default::default()
})
.collect(),
is_truncated,
..Default::default()
}
}
#[test]
fn transition_candidate_probe_classifier_is_fail_closed() {
assert_eq!(
classify_transition_candidate_versions(
"archive/object",
RemoteBucketVersioning::Disabled,
&list_versions(&[], &[], false),
),
TransitionCandidateProbe::Missing
);
assert_eq!(
classify_transition_candidate_versions(
"archive/object",
RemoteBucketVersioning::Disabled,
&list_versions(&[("archive/object", "")], &[], false),
),
TransitionCandidateProbe::UnversionedPresent
);
assert_eq!(
classify_transition_candidate_versions(
"archive/object",
RemoteBucketVersioning::Enabled,
&list_versions(&[("archive/object", "version-a")], &[], false),
),
TransitionCandidateProbe::VersionedPresent("version-a".to_string())
);
assert_eq!(
classify_transition_candidate_versions(
"archive/object",
RemoteBucketVersioning::Suspended,
&list_versions(&[("archive/object", "null")], &[], false),
),
TransitionCandidateProbe::VersionedPresent("null".to_string())
);
assert_eq!(
classify_transition_candidate_versions(
"archive/object",
RemoteBucketVersioning::Enabled,
&list_versions(&[("archive/object", "")], &[], false),
),
TransitionCandidateProbe::Ambiguous
);
assert_eq!(
classify_transition_candidate_versions(
"archive/object",
RemoteBucketVersioning::Enabled,
&list_versions(&[("archive/object", "version-a"), ("archive/object", "version-b")], &[], false),
),
TransitionCandidateProbe::Ambiguous
);
assert_eq!(
classify_transition_candidate_versions(
"archive/object",
RemoteBucketVersioning::Enabled,
&list_versions(&[("archive/object", "version-a")], &[("archive/object", "marker-a")], false),
),
TransitionCandidateProbe::Ambiguous
);
assert_eq!(
classify_transition_candidate_versions(
"archive/object",
RemoteBucketVersioning::Enabled,
&list_versions(&[("archive/object", "version-a")], &[], true),
),
TransitionCandidateProbe::Ambiguous
);
}
#[test]
fn remote_bucket_versioning_status_parser_fails_closed() {
assert_eq!(
remote_bucket_versioning_from_status(None).expect("absent status means disabled"),
RemoteBucketVersioning::Disabled
);
assert_eq!(
remote_bucket_versioning_from_status(Some(BucketVersioningStatus::ENABLED)).expect("enabled status should parse"),
RemoteBucketVersioning::Enabled
);
assert_eq!(
remote_bucket_versioning_from_status(Some(BucketVersioningStatus::SUSPENDED)).expect("suspended status should parse"),
RemoteBucketVersioning::Suspended
);
let err = remote_bucket_versioning_from_status(Some("UnexpectedStatus"))
.expect_err("unknown versioning status must fail closed");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
}
}
#[async_trait::async_trait]
@@ -215,6 +395,16 @@ impl WarmBackend for WarmBackendS3 {
self.remove_with_result(object, rv).await.map(|_| ())
}
async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
let bucket_versioning = self.remote_bucket_versioning().await?;
let versions = self.list_transition_candidate_versions(object).await?;
Ok(classify_transition_candidate_versions(
&self.get_dest(object),
bucket_versioning,
&versions,
))
}
async fn in_use(&self) -> Result<bool, std::io::Error> {
let result = self
.core