diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs
index 0b902719d..31a34e37d 100644
--- a/crates/ecstore/src/api/mod.rs
+++ b/crates/ecstore/src/api/mod.rs
@@ -426,7 +426,7 @@ pub mod tier {
pub mod tier_config {
pub use crate::services::tier::tier_config::{
ServicePrincipalAuth, TierAliyun, TierAzure, TierConfig, TierGCS, TierHuaweicloud, TierMinIO, TierR2, TierRustFS,
- TierS3, TierTencent, TierType,
+ TierS3, TierTencent, TierType, TierWasabi,
};
}
diff --git a/crates/ecstore/src/client/api_error_response.rs b/crates/ecstore/src/client/api_error_response.rs
index 8cd576e90..f586113e3 100644
--- a/crates/ecstore/src/client/api_error_response.rs
+++ b/crates/ecstore/src/client/api_error_response.rs
@@ -103,7 +103,7 @@ pub fn http_resp_to_error_response(
object_name: &str,
) -> ErrorResponse {
let err_body = String::from_utf8_lossy(&b).to_string();
- if h.is_empty() || resp_status.is_client_error() || resp_status.is_server_error() {
+ if h.is_empty() || !(resp_status.is_client_error() || resp_status.is_server_error()) {
return ErrorResponse {
status_code: resp_status,
code: S3ErrorCode::ResponseInterrupted,
@@ -329,4 +329,21 @@ mod tests {
assert_eq!(value["RequestId"], Value::String("req-xml-123".to_string()));
assert!(value.get("request_id").is_none(), "external error contract must not expose request_id");
}
+
+ #[test]
+ fn parses_s3_error_code_from_client_error_response() {
+ let mut headers = HeaderMap::new();
+ headers.insert("x-amz-request-id", "request-id".parse().expect("request ID header should parse"));
+
+ let response = http_resp_to_error_response(
+ StatusCode::NOT_FOUND,
+ &headers,
+ b"NoSuchVersionremote detail".to_vec(),
+ "bucket",
+ "object",
+ );
+
+ assert_eq!(response.code, S3ErrorCode::NoSuchVersion);
+ assert_eq!(response.status_code, StatusCode::NOT_FOUND);
+ }
}
diff --git a/crates/ecstore/src/client/api_stat.rs b/crates/ecstore/src/client/api_stat.rs
index c8557b4ce..e68b21d5e 100644
--- a/crates/ecstore/src/client/api_stat.rs
+++ b/crates/ecstore/src/client/api_stat.rs
@@ -18,7 +18,7 @@
#![allow(unused_must_use)]
#![allow(clippy::all)]
-use http::{HeaderMap, HeaderValue};
+use http::{HeaderMap, HeaderValue, StatusCode};
use http_body_util::BodyExt;
use hyper::body::Body;
use hyper::body::Bytes;
@@ -31,13 +31,62 @@ use uuid::Uuid;
use crate::client::{
api_error_response::{ErrorResponse, err_invalid_argument, http_resp_to_error_response},
api_get_options::GetObjectOptions,
- transition_api::{ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, to_object_info_for_provider},
+ transition_api::{
+ ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, collect_response_body, to_object_info_for_provider,
+ },
};
use s3s::{
- dto::VersioningConfiguration,
+ dto::{BucketVersioningStatus, MFADelete, VersioningConfiguration},
header::{X_AMZ_DELETE_MARKER, X_AMZ_VERSION_ID},
};
+const S3_XML_NAMESPACE: &str = "http://s3.amazonaws.com/doc/2006-03-01/";
+
+#[derive(serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct StrictVersioningConfiguration {
+ #[serde(rename = "@xmlns")]
+ namespace: Option,
+ #[serde(rename = "MfaDelete")]
+ mfa_delete: Option,
+ #[serde(rename = "Status")]
+ status: Option,
+}
+
+#[derive(serde::Deserialize)]
+enum StrictVersioningResponse {
+ VersioningConfiguration(StrictVersioningConfiguration),
+}
+
+fn parse_bucket_versioning_response(
+ status: StatusCode,
+ headers: &HeaderMap,
+ body: Vec,
+ bucket_name: &str,
+) -> Result {
+ if status != StatusCode::OK {
+ return Err(std::io::Error::other(http_resp_to_error_response(status, headers, body, bucket_name, "")));
+ }
+
+ let StrictVersioningResponse::VersioningConfiguration(parsed) =
+ quick_xml::de::from_reader(body.as_slice()).map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
+ if parsed
+ .namespace
+ .as_deref()
+ .is_some_and(|namespace| namespace != S3_XML_NAMESPACE)
+ {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::InvalidData,
+ "unexpected GetBucketVersioning XML namespace",
+ ));
+ }
+ Ok(VersioningConfiguration {
+ mfa_delete: parsed.mfa_delete.map(MFADelete::from),
+ status: parsed.status.map(BucketVersioningStatus::from),
+ ..Default::default()
+ })
+}
+
impl TransitionClient {
pub async fn bucket_exists(&self, bucket_name: &str) -> Result {
let resp = self
@@ -123,19 +172,8 @@ impl TransitionClient {
let resp_status = resp.status();
let h = resp.headers().clone();
- let mut body_vec = Vec::new();
- let mut body = resp.into_body();
- while let Some(frame) = body.frame().await {
- let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
- if let Some(data) = frame.data_ref() {
- body_vec.extend_from_slice(data);
- }
- }
- let resperr = http_resp_to_error_response(resp_status, &h, body_vec, bucket_name, "");
-
- warn!("get bucket versioning, resperr: {:?}", resperr);
-
- Ok(VersioningConfiguration::default())
+ let body_vec = collect_response_body(resp.into_body(), rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE).await?;
+ parse_bucket_versioning_response(resp_status, &h, body_vec, bucket_name)
}
Err(err) => Err(std::io::Error::other(err)),
@@ -235,3 +273,72 @@ impl TransitionClient {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::parse_bucket_versioning_response;
+ use http::{HeaderMap, StatusCode};
+ use s3s::dto::BucketVersioningStatus;
+
+ #[test]
+ fn parses_bucket_versioning_statuses_mfa_delete_and_unversioned_state() {
+ for (xml, expected) in [
+ (
+ br#"Enabled"#
+ .as_slice(),
+ Some(BucketVersioningStatus::ENABLED),
+ ),
+ (
+ br#"Suspended"#
+ .as_slice(),
+ Some(BucketVersioningStatus::SUSPENDED),
+ ),
+ (
+ br#""#.as_slice(),
+ None,
+ ),
+ ] {
+ let config = parse_bucket_versioning_response(StatusCode::OK, &HeaderMap::new(), xml.to_vec(), "tier-bucket")
+ .expect("valid GetBucketVersioning response should parse");
+ assert_eq!(config.status.as_ref().map(|status| status.as_str()), expected);
+ }
+
+ let config = parse_bucket_versioning_response(
+ StatusCode::OK,
+ &HeaderMap::new(),
+ br#"Enabled"#.to_vec(),
+ "tier-bucket",
+ )
+ .expect("valid MfaDelete should parse");
+ assert_eq!(config.mfa_delete.as_ref().map(|status| status.as_str()), Some("Enabled"));
+ }
+
+ #[test]
+ fn rejects_non_ok_and_malformed_bucket_versioning_responses() {
+ let valid = br#"Enabled"#.to_vec();
+ for status in [StatusCode::NO_CONTENT, StatusCode::PARTIAL_CONTENT] {
+ let status_err = parse_bucket_versioning_response(status, &HeaderMap::new(), valid.clone(), "tier-bucket")
+ .expect_err("GetBucketVersioning must return exactly HTTP 200");
+ assert_eq!(status_err.kind(), std::io::ErrorKind::Other);
+ }
+
+ let parse_err = parse_bucket_versioning_response(
+ StatusCode::OK,
+ &HeaderMap::new(),
+ b"".to_vec(),
+ "tier-bucket",
+ )
+ .expect_err("malformed GetBucketVersioning XML must fail closed");
+ assert_eq!(parse_err.kind(), std::io::ErrorKind::InvalidData);
+
+ for xml in [
+ b"Enabled".as_slice(),
+ b"Enabled".as_slice(),
+ b"".as_slice(),
+ ] {
+ let strict_err = parse_bucket_versioning_response(StatusCode::OK, &HeaderMap::new(), xml.to_vec(), "tier-bucket")
+ .expect_err("unknown GetBucketVersioning XML must fail closed");
+ assert_eq!(strict_err.kind(), std::io::ErrorKind::InvalidData);
+ }
+ }
+}
diff --git a/crates/ecstore/src/client/provider_versions.rs b/crates/ecstore/src/client/provider_versions.rs
index cad80ed4e..ddfff5910 100644
--- a/crates/ecstore/src/client/provider_versions.rs
+++ b/crates/ecstore/src/client/provider_versions.rs
@@ -60,6 +60,7 @@ impl ProviderVersionCapabilities {
|| tier_type.eq_ignore_ascii_case("rustfs")
|| tier_type.eq_ignore_ascii_case("minio")
|| tier_type.eq_ignore_ascii_case("r2")
+ || tier_type.eq_ignore_ascii_case("wasabi")
{
Self {
raw_version_header: Some(X_AMZ_VERSION_ID),
@@ -156,6 +157,8 @@ mod tests {
("MinIO", "x-amz-version-id"),
("r2", "x-amz-version-id"),
("R2", "x-amz-version-id"),
+ ("wasabi", "x-amz-version-id"),
+ ("Wasabi", "x-amz-version-id"),
("aliyun", "x-oss-version-id"),
("Aliyun", "x-oss-version-id"),
("tencent", "x-cos-version-id"),
diff --git a/crates/ecstore/src/client/transition_api.rs b/crates/ecstore/src/client/transition_api.rs
index 75fbcb206..008b9769b 100644
--- a/crates/ecstore/src/client/transition_api.rs
+++ b/crates/ecstore/src/client/transition_api.rs
@@ -42,7 +42,7 @@ use http::{
request::{Builder, Request},
};
use http_body::Body;
-use http_body_util::BodyExt;
+use http_body_util::{BodyExt, LengthLimitError, Limited};
use hyper::body::Bytes;
use hyper::body::Incoming;
use hyper_rustls::{ConfigBuilderExt, HttpsConnector};
@@ -55,9 +55,7 @@ use rustfs_rio::HashReader;
use rustfs_utils::HashAlgorithm;
use rustfs_utils::{
net::get_endpoint_url,
- retry::{
- DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, MAX_RETRY, RetryTimer, is_http_status_retryable, is_s3code_retryable,
- },
+ retry::{DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, MAX_RETRY, RetryTimer},
};
use rustls_pki_types::PrivateKeyDer;
use rustls_pki_types::pem::PemObject;
@@ -82,9 +80,25 @@ use url::{Url, form_urlencoded};
use uuid::Uuid;
const C_USER_AGENT: &str = "RustFS (linux; x86)";
+pub(crate) const MAX_S3_ERROR_RESPONSE_SIZE: usize = 64 * 1024;
const SUCCESS_STATUS: [StatusCode; 3] = [StatusCode::OK, StatusCode::NO_CONTENT, StatusCode::PARTIAL_CONTENT];
+pub(crate) async fn collect_response_body(body: B, limit: usize) -> Result, std::io::Error>
+where
+ B: Body,
+ B::Error: Into>,
+{
+ let body = Limited::new(body, limit).collect().await.map_err(|err| {
+ if err.is::() {
+ std::io::Error::new(std::io::ErrorKind::InvalidData, "remote tier response body exceeds limit")
+ } else {
+ std::io::Error::other(err)
+ }
+ })?;
+ Ok(body.to_bytes().to_vec())
+}
+
const C_UNKNOWN: i32 = -1;
const C_OFFLINE: i32 = 0;
const C_ONLINE: i32 = 1;
@@ -397,20 +411,17 @@ impl TransitionClient {
pub async fn doit(&self, req: Request) -> Result, std::io::Error> {
let req_method;
let req_uri;
- let req_headers;
let resp;
let http_client = self.http_client.clone();
{
req_method = req.method().clone();
req_uri = req.uri().clone();
- req_headers = req.headers().clone();
debug!("endpoint_url: {}", self.endpoint_url.as_str().to_string());
resp = http_client.request(req);
}
let resp = resp.await;
debug!("http_client url: {} {}", req_method, req_uri);
- debug!("http_client headers: {:?}", req_headers);
if let Err(err) = resp {
error!("http_client call error: {:?}", err);
return Err(std::io::Error::other(err));
@@ -420,28 +431,23 @@ impl TransitionClient {
Ok(r) => r,
Err(_) => return Err(std::io::Error::other("Unexpected error in response")),
};
- debug!("http_resp: {:?}", resp);
+ debug!(status = %resp.status(), "remote tier response received");
//let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
//debug!("http_resp_body: {}", String::from_utf8(b).unwrap());
//if self.is_trace_enabled && !(self.trace_errors_only && resp.status() == StatusCode::OK) {
if !resp.status().is_success() {
- //self.dump_http(&cloned_req, &resp)?;
- let mut body_vec = Vec::new();
- let mut body = resp.into_body();
- while let Some(frame) = body.frame().await {
- let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
- if let Some(data) = frame.data_ref() {
- body_vec.extend_from_slice(data);
- }
- }
- let body_str = String::from_utf8_lossy(&body_vec);
- warn!("err_body: {}", body_str);
- Err(std::io::Error::other(format!("http_client call error: {}", body_str)))
- } else {
- Ok(resp)
+ let status = resp.status();
+ let request_id = resp
+ .headers()
+ .get("x-amz-request-id")
+ .and_then(|value| value.to_str().ok())
+ .unwrap_or_default()
+ .to_string();
+ warn!(status = %status, request_id, "remote tier request rejected");
}
+ Ok(resp)
}
pub async fn execute_method(
@@ -483,17 +489,22 @@ impl TransitionClient {
let resp_status = resp.status();
let h = resp.headers().clone();
- let mut body_vec = Vec::new();
- let mut body = resp.into_body();
- while let Some(frame) = body.frame().await {
- let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
- if let Some(data) = frame.data_ref() {
- body_vec.extend_from_slice(data);
- }
- }
- let mut err_response =
- http_resp_to_error_response(resp_status, &h, body_vec.clone(), &metadata.bucket_name, &metadata.object_name);
- err_response.message = format!("remote tier error: {}", err_response.message);
+ let body_vec = collect_response_body(resp.into_body(), MAX_S3_ERROR_RESPONSE_SIZE).await?;
+ let parsed_error =
+ http_resp_to_error_response(resp_status, &h, body_vec, &metadata.bucket_name, &metadata.object_name);
+ let routing_region = parsed_error.region;
+ let code = match parsed_error.code {
+ S3ErrorCode::Custom(_) => S3ErrorCode::ResponseInterrupted,
+ code => code,
+ };
+ let err_response = ErrorResponse {
+ status_code: resp_status,
+ message: format!("remote tier request failed with status {resp_status}: {}", code.as_str()),
+ code,
+ bucket_name: metadata.bucket_name.clone(),
+ key: metadata.object_name.clone(),
+ ..Default::default()
+ };
if self.region == "" {
return match err_response.code {
@@ -502,20 +513,20 @@ impl TransitionClient {
Err(std::io::Error::other(err_response))
}
S3ErrorCode::AccessDenied => {
- if err_response.region == "" {
+ if routing_region.is_empty() {
return Err(std::io::Error::other(err_response));
}
if metadata.bucket_name != "" {
if let Ok(mut bucket_loc_cache) = self.bucket_loc_cache.lock() {
if let Some(location) = bucket_loc_cache.get(&metadata.bucket_name) {
- if location != err_response.region {
- bucket_loc_cache.set(&metadata.bucket_name, &err_response.region);
+ if location != routing_region {
+ bucket_loc_cache.set(&metadata.bucket_name, &routing_region);
//continue;
}
}
}
- } else if err_response.region != metadata.bucket_location {
- metadata.bucket_location = err_response.region.clone();
+ } else if routing_region != metadata.bucket_location {
+ metadata.bucket_location = routing_region;
//continue;
}
Err(std::io::Error::other(err_response))
@@ -526,18 +537,10 @@ impl TransitionClient {
};
}
- if is_s3code_retryable(err_response.code.as_str()) {
- continue;
- }
-
- if is_http_status_retryable(&resp_status) {
- continue;
- }
-
- break;
+ return Err(std::io::Error::other(err_response));
}
- Err(std::io::Error::other("resp err"))
+ Err(std::io::Error::other("remote tier request did not produce a response"))
}
async fn new_request(
@@ -1408,13 +1411,43 @@ pub struct CreateBucketConfiguration {
#[cfg(test)]
mod tests {
use super::{
- SignatureType, build_tls_config, signer_error_to_io_error, to_object_info_for_provider, validate_header_values,
- with_rustls_init_guard,
+ MAX_S3_CLIENT_RESPONSE_SIZE, MAX_S3_ERROR_RESPONSE_SIZE, SignatureType, build_tls_config, collect_response_body,
+ signer_error_to_io_error, to_object_info_for_provider, validate_header_values, with_rustls_init_guard,
};
use crate::client::provider_versions::ProviderVersionCapabilities;
use http::{HeaderMap, HeaderValue};
+ use http_body_util::Full;
+ use hyper::body::Bytes;
use uuid::Uuid;
+ #[tokio::test]
+ async fn response_body_limit_rejects_the_first_byte_past_the_boundary() {
+ let exact = collect_response_body(
+ Full::new(Bytes::from(vec![0_u8; MAX_S3_CLIENT_RESPONSE_SIZE])),
+ MAX_S3_CLIENT_RESPONSE_SIZE,
+ )
+ .await
+ .expect("a response exactly at the limit should be accepted");
+ assert_eq!(exact.len(), MAX_S3_CLIENT_RESPONSE_SIZE);
+
+ let err = collect_response_body(
+ Full::new(Bytes::from(vec![0_u8; MAX_S3_CLIENT_RESPONSE_SIZE + 1])),
+ MAX_S3_CLIENT_RESPONSE_SIZE,
+ )
+ .await
+ .expect_err("oversized remote response must fail closed");
+
+ assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+
+ let err = collect_response_body(
+ Full::new(Bytes::from(vec![0_u8; MAX_S3_ERROR_RESPONSE_SIZE + 1])),
+ MAX_S3_ERROR_RESPONSE_SIZE,
+ )
+ .await
+ .expect_err("oversized S3 error response must use the smaller error limit");
+ assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+ }
+
#[test]
fn rustls_guard_converts_panics_to_io_errors() {
let err = with_rustls_init_guard(|| -> Result<(), std::io::Error> { panic!("missing provider") })
diff --git a/crates/ecstore/src/services/tier/mod.rs b/crates/ecstore/src/services/tier/mod.rs
index 15f975f99..3b501eb15 100644
--- a/crates/ecstore/src/services/tier/mod.rs
+++ b/crates/ecstore/src/services/tier/mod.rs
@@ -30,3 +30,4 @@ pub mod warm_backend_rustfs;
pub mod warm_backend_s3;
pub mod warm_backend_s3sdk;
pub mod warm_backend_tencent;
+pub mod warm_backend_wasabi;
diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs
index 134e5b1d1..26bb97e85 100644
--- a/crates/ecstore/src/services/tier/tier.rs
+++ b/crates/ecstore/src/services/tier/tier.rs
@@ -51,7 +51,7 @@ use crate::client::admin_handler_utils::AdminError;
use crate::error::{Error, Result, StorageError};
use crate::services::tier::{
tier_admin::TierCreds,
- tier_config::{TierConfig, TierType},
+ 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},
};
@@ -146,6 +146,8 @@ const EXTERNAL_TIER_TYPE_S3: i32 = 1;
const EXTERNAL_TIER_TYPE_AZURE: i32 = 2;
const EXTERNAL_TIER_TYPE_GCS: i32 = 3;
const EXTERNAL_TIER_TYPE_MINIO: i32 = 4;
+const EXTERNAL_TIER_VERSION_WASABI_V1: &str = "rustfs-wasabi-v1";
+const EXTERNAL_TIER_WASABI_ENDPOINT_SENTINEL: &str = "rustfs-wasabi-v1:";
const TIER_BACKEND_IDENTITY_VERSION: u8 = 2;
const _TIER_CFG_REFRESH_AT_HDR: &str = "X-RustFS-TierCfg-RefreshedAt";
@@ -484,6 +486,14 @@ fn replaced_tier_destinations(
for (tier_name, current_config) in ¤t.tiers {
let changed = match replacement.tiers.get(tier_name) {
Some(replacement_config) => {
+ if matches!(current_config.tier_type, TierType::Wasabi)
+ != matches!(replacement_config.tier_type, TierType::Wasabi)
+ {
+ let mut admin_err = ERR_TIER_INVALID_CONFIG.clone();
+ admin_err.message =
+ "Changing an existing remote tier to or from Wasabi is not supported; add a new tier instead".to_string();
+ return Err(admin_err);
+ }
let current_identity = tier_backend_identity(current_config).map_err(|err| {
let mut admin_err = ERR_TIER_INVALID_CONFIG.clone();
admin_err.message = err.to_string();
@@ -505,7 +515,43 @@ fn replaced_tier_destinations(
Ok(replaced)
}
+fn encode_tier_backend_identity(
+ tier_type: &str,
+ endpoint: &str,
+ bucket: &str,
+ prefix: &str,
+ region: &str,
+ routing_account: &str,
+) -> io::Result {
+ let encoded = rmp_serde::to_vec(&(
+ TIER_BACKEND_IDENTITY_VERSION,
+ tier_type,
+ endpoint,
+ bucket,
+ prefix,
+ region,
+ routing_account,
+ ))
+ .map_err(|err| io::Error::other(format!("serialize tier backend identity failed: {err}")))?;
+ Ok(Sha256::digest(encoded).into())
+}
+
fn tier_backend_identity(config: &TierConfig) -> io::Result {
+ if matches!(config.tier_type, TierType::Wasabi) {
+ let wasabi = config
+ .wasabi
+ .as_ref()
+ .ok_or_else(|| io::Error::other("tier backend identity payload is missing"))?;
+ return encode_tier_backend_identity(
+ "s3",
+ &wasabi.canonical_endpoint()?,
+ &wasabi.bucket,
+ normalized_tier_prefix(&config.tier_type, &wasabi.prefix),
+ &wasabi.region,
+ "",
+ );
+ }
+
let (tier_type, endpoint, bucket, prefix, region, routing_account) = match config.tier_type {
TierType::S3 => config.s3.as_ref().map(|value| {
(
@@ -517,6 +563,7 @@ fn tier_backend_identity(config: &TierConfig) -> io::Result {
"",
)
}),
+ TierType::Wasabi => None,
TierType::RustFS => config.rustfs.as_ref().map(|value| {
(
"rustfs",
@@ -601,17 +648,7 @@ fn tier_backend_identity(config: &TierConfig) -> io::Result {
}
.ok_or_else(|| io::Error::other("tier backend identity payload is missing"))?;
let prefix = normalized_tier_prefix(&config.tier_type, prefix);
- let encoded = rmp_serde::to_vec(&(
- TIER_BACKEND_IDENTITY_VERSION,
- tier_type,
- endpoint,
- bucket,
- prefix,
- region,
- routing_account,
- ))
- .map_err(|err| io::Error::other(format!("serialize tier backend identity failed: {err}")))?;
- Ok(Sha256::digest(encoded).into())
+ encode_tier_backend_identity(tier_type, endpoint, bucket, prefix, region, routing_account)
}
pub(crate) fn tier_destination_id_from_metadata(metadata: &HashMap) -> io::Result> {
@@ -642,7 +679,7 @@ pub(crate) fn tier_destination_id_from_metadata(metadata: &HashMap(tier_type: &TierType, prefix: &'a str) -> &'a str {
match tier_type {
- TierType::S3 => prefix.trim_matches('/'),
+ TierType::S3 | TierType::Wasabi => prefix.trim_matches('/'),
TierType::RustFS
| TierType::MinIO
| TierType::Aliyun
@@ -949,6 +986,7 @@ fn tier_config_path(file: &str) -> String {
fn tier_hint_for_type(tier_type: TierType) -> Option<&'static str> {
match tier_type {
TierType::RustFS => Some("rustfs"),
+ TierType::Wasabi => Some("wasabi"),
TierType::Aliyun => Some("aliyun"),
TierType::Tencent => Some("tencent"),
TierType::Huaweicloud => Some("huaweicloud"),
@@ -1051,6 +1089,26 @@ fn to_external_tier_config(name: &str, tier: &TierConfig) -> io::Result {
+ let backend = tier
+ .wasabi
+ .as_ref()
+ .ok_or_else(|| io::Error::other("tier config missing Wasabi backend payload"))?;
+ out.version = EXTERNAL_TIER_VERSION_WASABI_V1.to_string();
+ out.tier_type = EXTERNAL_TIER_TYPE_S3;
+ out.tier_type_hint = tier_hint_for_type(tier.tier_type.clone()).map(ToString::to_string);
+ out.s3 = Some(external_tier_s3_from_compatible_payload(
+ {
+ backend.canonical_endpoint()?;
+ EXTERNAL_TIER_WASABI_ENDPOINT_SENTINEL.to_string()
+ },
+ backend.access_key.clone(),
+ backend.secret_key.clone(),
+ backend.bucket.clone(),
+ backend.prefix.clone(),
+ backend.region.clone(),
+ ));
+ }
TierType::Azure => {
let az = tier
.azure
@@ -1217,17 +1275,36 @@ fn from_external_tier_config(name: String, ext: ExternalTierConfig) -> io::Resul
..Default::default()
};
- let hinted = tier_type_from_hint(ext.tier_type_hint.as_deref());
- let tier_type = if let Some(h) = hinted {
- h
+ let wasabi_hint = ext.tier_type_hint.as_deref() == Some("wasabi");
+ let wasabi_version = ext.version == EXTERNAL_TIER_VERSION_WASABI_V1;
+ if wasabi_hint && !wasabi_version {
+ return Err(io::Error::other(format!(
+ "tier config '{}' has inconsistent Wasabi type discriminators",
+ cfg.name
+ )));
+ }
+ if (wasabi_hint || wasabi_version) && ext.tier_type != EXTERNAL_TIER_TYPE_S3 {
+ return Err(io::Error::other(format!(
+ "tier config '{}' has inconsistent Wasabi type discriminators",
+ cfg.name
+ )));
+ }
+ if wasabi_version && ext.tier_type_hint.as_deref().is_some_and(|hint| hint != "wasabi") {
+ return Err(io::Error::other(format!(
+ "tier config '{}' has inconsistent Wasabi type discriminators",
+ cfg.name
+ )));
+ }
+ let tier_type = if wasabi_version {
+ TierType::Wasabi
} else {
- match ext.tier_type {
+ tier_type_from_hint(ext.tier_type_hint.as_deref()).unwrap_or_else(|| match ext.tier_type {
EXTERNAL_TIER_TYPE_S3 => TierType::S3,
EXTERNAL_TIER_TYPE_AZURE => TierType::Azure,
EXTERNAL_TIER_TYPE_GCS => TierType::GCS,
EXTERNAL_TIER_TYPE_MINIO => TierType::MinIO,
_ => TierType::Unsupported,
- }
+ })
};
cfg.tier_type = tier_type.clone();
@@ -1254,6 +1331,41 @@ fn from_external_tier_config(name: String, ext: ExternalTierConfig) -> io::Resul
aws_role_duration_seconds: s3.aws_role_duration_seconds,
});
}
+ TierType::Wasabi => {
+ let s3 = ext
+ .s3
+ .as_ref()
+ .ok_or_else(|| io::Error::other(format!("tier config '{}' missing Wasabi S3 payload", cfg.name)))?;
+ if !s3.storage_class.is_empty()
+ || s3.aws_role
+ || !s3.aws_role_web_identity_token_file.is_empty()
+ || !s3.aws_role_arn.is_empty()
+ || !s3.aws_role_session_name.is_empty()
+ || s3.aws_role_duration_seconds != 0
+ {
+ return Err(io::Error::other(format!(
+ "tier config '{}' has unsupported Wasabi S3 credential fields",
+ cfg.name
+ )));
+ }
+ if s3.endpoint != EXTERNAL_TIER_WASABI_ENDPOINT_SENTINEL {
+ return Err(io::Error::other(format!(
+ "tier config '{}' has an invalid Wasabi compatibility endpoint",
+ cfg.name
+ )));
+ }
+ let mut wasabi = TierWasabi {
+ name: cfg.name.clone(),
+ endpoint: String::new(),
+ access_key: s3.access_key.clone(),
+ secret_key: s3.secret_key.clone(),
+ bucket: s3.bucket.clone(),
+ prefix: s3.prefix.clone(),
+ region: s3.region.clone(),
+ };
+ wasabi.normalize_endpoint()?;
+ cfg.wasabi = Some(wasabi);
+ }
TierType::Azure => {
let az = ext
.azure
@@ -1432,7 +1544,19 @@ impl TierConfigMgr {
}
pub fn unmarshal(data: &[u8]) -> std::result::Result {
- let cfg: TierConfigMgr = serde_json::from_slice(data)?;
+ let mut cfg: TierConfigMgr = serde_json::from_slice(data)?;
+ for (name, tier) in &mut cfg.tiers {
+ if matches!(&tier.tier_type, TierType::Wasabi) {
+ let wasabi = tier
+ .wasabi
+ .as_mut()
+ .ok_or_else(|| io::Error::other(format!("tier config '{name}' missing Wasabi backend payload")))?;
+ if wasabi.endpoint.is_empty() {
+ return Err(io::Error::other(format!("tier config '{name}' missing Wasabi endpoint")));
+ }
+ wasabi.normalize_endpoint()?;
+ }
+ }
Ok(cfg)
}
@@ -1459,7 +1583,7 @@ impl TierConfigMgr {
(TierType::Unsupported, false)
}
- pub async fn add(&mut self, tier_config: TierConfig, force: bool) -> std::result::Result<(), AdminError> {
+ pub async fn add(&mut self, mut tier_config: TierConfig, force: bool) -> std::result::Result<(), AdminError> {
self.ensure_generation_is_idle(&tier_config.name)?;
let tier_name = tier_config.name.clone();
if tier_name != tier_name.to_uppercase() {
@@ -1481,6 +1605,16 @@ impl TierConfigMgr {
return Err(ERR_TIER_ALREADY_EXISTS.clone());
}
+ if matches!(&tier_config.tier_type, TierType::Wasabi)
+ && let Some(wasabi) = tier_config.wasabi.as_mut()
+ {
+ wasabi.normalize_endpoint().map_err(|source| {
+ let mut err = ERR_TIER_INVALID_CONFIG.clone();
+ err.message = source.to_string();
+ err
+ })?;
+ }
+
let d = new_warm_backend(&tier_config, true).await?;
if !force {
@@ -1615,6 +1749,15 @@ impl TierConfigMgr {
}
}
}
+ TierType::Wasabi => {
+ if let Some(wasabi) = tier_config.wasabi.as_mut() {
+ if creds.access_key.is_empty() || creds.secret_key.is_empty() {
+ return Err(ERR_TIER_MISSING_CREDENTIALS.clone());
+ }
+ wasabi.access_key = creds.access_key;
+ wasabi.secret_key = creds.secret_key;
+ }
+ }
TierType::RustFS => {
if let Some(rustfs) = tier_config.rustfs.as_mut() {
if creds.access_key == "" || creds.secret_key == "" {
@@ -2918,6 +3061,24 @@ mod tests {
}
}
+ fn build_wasabi_tier(name: &str) -> TierConfig {
+ TierConfig {
+ version: "v1".to_string(),
+ tier_type: TierType::Wasabi,
+ name: name.to_string(),
+ wasabi: Some(TierWasabi {
+ name: name.to_string(),
+ endpoint: "https://s3.ap-northeast-1.wasabisys.com".to_string(),
+ access_key: "ak".to_string(),
+ secret_key: "sk".to_string(),
+ bucket: "wasabi-bucket".to_string(),
+ prefix: "archive".to_string(),
+ region: "ap-northeast-1".to_string(),
+ }),
+ ..Default::default()
+ }
+ }
+
#[test]
fn test_tiering_external_blob_roundtrip_for_standard_type() {
let mut cfg = TierConfigMgr {
@@ -2974,6 +3135,246 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn wasabi_external_blob_survives_old_node_rewrite_and_fences_old_drivers() {
+ const WASABI_V2_GOLDEN: &[u8] = &[
+ 1, 0, 2, 0, 145, 129, 171, 67, 79, 76, 68, 45, 87, 65, 83, 65, 66, 73, 152, 176, 114, 117, 115, 116, 102, 115, 45,
+ 119, 97, 115, 97, 98, 105, 45, 118, 49, 1, 171, 67, 79, 76, 68, 45, 87, 65, 83, 65, 66, 73, 156, 177, 114, 117, 115,
+ 116, 102, 115, 45, 119, 97, 115, 97, 98, 105, 45, 118, 49, 58, 162, 97, 107, 162, 115, 107, 173, 119, 97, 115, 97,
+ 98, 105, 45, 98, 117, 99, 107, 101, 116, 167, 97, 114, 99, 104, 105, 118, 101, 174, 97, 112, 45, 110, 111, 114, 116,
+ 104, 101, 97, 115, 116, 45, 49, 160, 194, 160, 160, 160, 0, 192, 192, 192, 166, 119, 97, 115, 97, 98, 105,
+ ];
+
+ let mut cfg = empty_mgr();
+ let mut tier = build_wasabi_tier("COLD-WASABI");
+ tier.wasabi.as_mut().expect("Wasabi payload should exist").endpoint.clear();
+ cfg.tiers.insert("COLD-WASABI".to_string(), tier);
+
+ let bytes = encode_external_tiering_config_blob(&cfg).expect("Wasabi tier should encode");
+ assert_eq!(bytes.as_ref(), WASABI_V2_GOLDEN, "Wasabi v2 bytes must remain stable");
+ assert_eq!(&bytes[0..2], &TIER_CONFIG_FORMAT.to_le_bytes());
+ assert_eq!(&bytes[2..4], &TIER_CONFIG_VERSION.to_le_bytes());
+ let external: ExternalTierConfigMgr = rmp_serde::from_slice(&bytes[4..]).expect("external Wasabi payload should decode");
+ let stored = external.tiers.get("COLD-WASABI").expect("stored Wasabi tier should exist");
+ assert_eq!(stored.version, EXTERNAL_TIER_VERSION_WASABI_V1);
+ assert_eq!(stored.tier_type, EXTERNAL_TIER_TYPE_S3);
+ assert_eq!(stored.tier_type_hint.as_deref(), Some("wasabi"));
+ assert!(stored.compatible_backend.is_none());
+ let s3 = stored.s3.as_ref().expect("Wasabi should use the S3 payload");
+ assert_eq!(s3.endpoint, EXTERNAL_TIER_WASABI_ENDPOINT_SENTINEL);
+ assert_eq!(s3.bucket, "wasabi-bucket");
+ assert_eq!(s3.prefix, "archive");
+ assert_eq!(s3.region, "ap-northeast-1");
+ assert!(s3.storage_class.is_empty());
+
+ let decoded = decode_external_tiering_config_blob(&bytes).expect("Wasabi tier should decode");
+ let wasabi = decoded.tiers["COLD-WASABI"]
+ .wasabi
+ .as_ref()
+ .expect("Wasabi payload should survive roundtrip");
+ assert_eq!(wasabi.endpoint, "https://s3.ap-northeast-1.wasabisys.com");
+ assert_eq!(wasabi.secret_key, "sk");
+ assert_eq!(decoded.tiers["COLD-WASABI"].version, EXTERNAL_TIER_VERSION_WASABI_V1);
+
+ let encode_fixture = |external: &ExternalTierConfigMgr| {
+ let payload = rmp_serde::to_vec(external).expect("Wasabi fixture should encode");
+ let mut fixture = Vec::with_capacity(4 + payload.len());
+ fixture.extend_from_slice(&TIER_CONFIG_FORMAT.to_le_bytes());
+ fixture.extend_from_slice(&TIER_CONFIG_VERSION.to_le_bytes());
+ fixture.extend_from_slice(&payload);
+ fixture
+ };
+
+ // Current older nodes preserve the per-tier Version and S3 payload when
+ // rewriting the binary document, but drop an unrecognized XTierType.
+ let mut old_node_rewrite = external.clone();
+ old_node_rewrite
+ .tiers
+ .get_mut("COLD-WASABI")
+ .expect("stored Wasabi tier should exist")
+ .tier_type_hint = None;
+ let rewritten = encode_fixture(&old_node_rewrite);
+ let decoded = decode_external_tiering_config_blob(&rewritten)
+ .expect("the durable per-tier marker must restore Wasabi after an old-node rewrite");
+ assert!(matches!(decoded.tiers["COLD-WASABI"].tier_type, TierType::Wasabi));
+
+ let old_s3_payload = old_node_rewrite.tiers["COLD-WASABI"]
+ .s3
+ .as_ref()
+ .expect("old nodes should retain the physical S3 payload");
+ let old_s3 = crate::services::tier::tier_config::TierS3 {
+ name: "COLD-WASABI".to_string(),
+ endpoint: old_s3_payload.endpoint.clone(),
+ access_key: old_s3_payload.access_key.clone(),
+ secret_key: old_s3_payload.secret_key.clone(),
+ bucket: old_s3_payload.bucket.clone(),
+ prefix: old_s3_payload.prefix.clone(),
+ region: old_s3_payload.region.clone(),
+ ..Default::default()
+ };
+ let err = match crate::services::tier::warm_backend_s3::WarmBackendS3::new(&old_s3, "COLD-WASABI").await {
+ Ok(_) => panic!("an older generic S3 driver must not operate a Wasabi compatibility envelope"),
+ Err(err) => err,
+ };
+ assert!(err.to_string().contains("missing a host"), "{err}");
+
+ let mut generic_s3 = old_node_rewrite.clone();
+ generic_s3
+ .tiers
+ .get_mut("COLD-WASABI")
+ .expect("stored Wasabi tier should exist")
+ .version = "v1".to_string();
+ let decoded = decode_external_tiering_config_blob(&encode_fixture(&generic_s3))
+ .expect("a generic S3 tier without an explicit marker must stay generic");
+ assert!(matches!(decoded.tiers["COLD-WASABI"].tier_type, TierType::S3));
+
+ let mut hint_only = external.clone();
+ hint_only
+ .tiers
+ .get_mut("COLD-WASABI")
+ .expect("stored Wasabi tier should exist")
+ .version = "v1".to_string();
+ let err = expect_decode_err(&encode_fixture(&hint_only));
+ assert!(err.to_string().contains("inconsistent Wasabi type discriminators"), "{err}");
+
+ let mut wrong_type = old_node_rewrite.clone();
+ wrong_type
+ .tiers
+ .get_mut("COLD-WASABI")
+ .expect("stored Wasabi tier should exist")
+ .tier_type = EXTERNAL_TIER_TYPE_MINIO;
+ let err = expect_decode_err(&encode_fixture(&wrong_type));
+ assert!(err.to_string().contains("inconsistent Wasabi type discriminators"), "{err}");
+
+ let mut wrong_hint = external.clone();
+ wrong_hint
+ .tiers
+ .get_mut("COLD-WASABI")
+ .expect("stored Wasabi tier should exist")
+ .tier_type_hint = Some("rustfs".to_string());
+ let err = expect_decode_err(&encode_fixture(&wrong_hint));
+ assert!(err.to_string().contains("inconsistent Wasabi type discriminators"), "{err}");
+
+ let poison_fields: [(&str, fn(&mut ExternalTierS3)); 6] = [
+ ("storage_class", |s3| s3.storage_class = "GLACIER".to_string()),
+ ("aws_role", |s3| s3.aws_role = true),
+ ("web_identity_token", |s3| s3.aws_role_web_identity_token_file = "/tmp/token".to_string()),
+ ("role_arn", |s3| s3.aws_role_arn = "arn:aws:iam::1:role/test".to_string()),
+ ("role_session", |s3| s3.aws_role_session_name = "session".to_string()),
+ ("role_duration", |s3| s3.aws_role_duration_seconds = 900),
+ ];
+ for (field, poison) in poison_fields {
+ let mut unsupported_credentials = external.clone();
+ let s3 = unsupported_credentials
+ .tiers
+ .get_mut("COLD-WASABI")
+ .expect("stored Wasabi tier should exist")
+ .s3
+ .as_mut()
+ .expect("Wasabi S3 payload should exist");
+ poison(s3);
+ let err = expect_decode_err(&encode_fixture(&unsupported_credentials));
+ assert!(
+ err.to_string().contains("unsupported Wasabi S3 credential fields"),
+ "field {field}: {err}"
+ );
+ }
+ }
+
+ #[test]
+ fn legacy_extended_type_hint_keeps_precedence_over_numeric_type() {
+ let mut external = ExternalTierConfigMgr::default();
+ external.tiers.insert(
+ "COLD-LEGACY".to_string(),
+ ExternalTierConfig {
+ version: "v1".to_string(),
+ tier_type: EXTERNAL_TIER_TYPE_MINIO,
+ name: "COLD-LEGACY".to_string(),
+ tier_type_hint: Some("rustfs".to_string()),
+ s3: Some(ExternalTierS3 {
+ endpoint: "https://rustfs.example.invalid".to_string(),
+ access_key: "ak".to_string(),
+ secret_key: "sk".to_string(),
+ bucket: "archive".to_string(),
+ prefix: "objects".to_string(),
+ region: "us-east-1".to_string(),
+ ..Default::default()
+ }),
+ compatible_backend: Some(ExternalTierCompatible {
+ endpoint: "https://minio.example.invalid".to_string(),
+ bucket: "different".to_string(),
+ ..Default::default()
+ }),
+ ..Default::default()
+ },
+ );
+ let payload = rmp_serde::to_vec(&external).expect("legacy fixture should encode");
+ let mut blob = Vec::with_capacity(4 + payload.len());
+ blob.extend_from_slice(&TIER_CONFIG_FORMAT.to_le_bytes());
+ blob.extend_from_slice(&TIER_CONFIG_VERSION.to_le_bytes());
+ blob.extend_from_slice(&payload);
+
+ let decoded = decode_external_tiering_config_blob(&blob).expect("legacy extended hint should decode as before");
+ let rustfs = decoded.tiers["COLD-LEGACY"]
+ .rustfs
+ .as_ref()
+ .expect("recognized legacy hint must override the numeric type");
+ assert_eq!(rustfs.endpoint, "https://rustfs.example.invalid");
+ assert_eq!(rustfs.bucket, "archive");
+ }
+
+ #[test]
+ fn wasabi_external_blob_rejects_noncanonical_endpoint_and_unsafe_region() {
+ let mut cfg = empty_mgr();
+ cfg.tiers.insert("COLD-WASABI".to_string(), build_wasabi_tier("COLD-WASABI"));
+ let bytes = encode_external_tiering_config_blob(&cfg).expect("Wasabi tier should encode");
+ let base: ExternalTierConfigMgr = rmp_serde::from_slice(&bytes[4..]).expect("external Wasabi payload should decode");
+
+ for (endpoint, region, expected) in [
+ ("https://s3.example.invalid", "ap-northeast-1", "invalid Wasabi compatibility endpoint"),
+ (EXTERNAL_TIER_WASABI_ENDPOINT_SENTINEL, "ap.northeast-1", "invalid Wasabi region"),
+ ] {
+ let mut external = base.clone();
+ let s3 = external
+ .tiers
+ .get_mut("COLD-WASABI")
+ .and_then(|tier| tier.s3.as_mut())
+ .expect("stored Wasabi S3 payload should exist");
+ s3.endpoint = endpoint.to_string();
+ s3.region = region.to_string();
+ let payload = rmp_serde::to_vec(&external).expect("corrupt fixture should encode");
+ let mut corrupt = Vec::with_capacity(4 + payload.len());
+ corrupt.extend_from_slice(&TIER_CONFIG_FORMAT.to_le_bytes());
+ corrupt.extend_from_slice(&TIER_CONFIG_VERSION.to_le_bytes());
+ corrupt.extend_from_slice(&payload);
+
+ let err = expect_decode_err(&corrupt);
+ assert!(err.to_string().contains(expected), "{err}");
+ }
+
+ let mut external = base;
+ let stored = external
+ .tiers
+ .get_mut("COLD-WASABI")
+ .expect("stored Wasabi tier should exist");
+ let s3 = stored.s3.take().expect("stored Wasabi S3 payload should exist");
+ stored.compatible_backend = Some(external_tier_alias_from_compatible_payload(
+ s3.endpoint,
+ s3.access_key,
+ s3.secret_key,
+ s3.bucket,
+ s3.prefix,
+ s3.region,
+ ));
+ let payload = rmp_serde::to_vec(&external).expect("wrong-payload fixture should encode");
+ let mut corrupt = Vec::with_capacity(4 + payload.len());
+ corrupt.extend_from_slice(&TIER_CONFIG_FORMAT.to_le_bytes());
+ corrupt.extend_from_slice(&TIER_CONFIG_VERSION.to_le_bytes());
+ corrupt.extend_from_slice(&payload);
+ let err = expect_decode_err(&corrupt);
+ assert!(err.to_string().contains("missing Wasabi S3 payload"), "{err}");
+ }
+
#[test]
fn test_decode_tiering_config_blob_accepts_legacy_json() {
let mut cfg = TierConfigMgr {
@@ -2995,6 +3396,42 @@ mod tests {
);
}
+ #[test]
+ fn wasabi_legacy_json_rejects_missing_payload_and_invalid_endpoints() {
+ let mut cfg = empty_mgr();
+ let mut tier = build_wasabi_tier("COLD-WASABI");
+ tier.wasabi.as_mut().expect("Wasabi payload should exist").endpoint = "https://s3.example.invalid".to_string();
+ cfg.tiers.insert("COLD-WASABI".to_string(), tier);
+
+ let data = serde_json::to_vec(&cfg).expect("legacy JSON fixture should encode");
+ let err = match TierConfigMgr::unmarshal(&data) {
+ Ok(_) => panic!("legacy Wasabi endpoint mismatch must fail closed"),
+ Err(err) => err,
+ };
+ assert!(err.to_string().contains("https://s3.ap-northeast-1.wasabisys.com"));
+
+ cfg.tiers
+ .get_mut("COLD-WASABI")
+ .and_then(|tier| tier.wasabi.as_mut())
+ .expect("Wasabi payload should exist")
+ .endpoint
+ .clear();
+ let data = serde_json::to_vec(&cfg).expect("empty-endpoint fixture should encode");
+ let err = match TierConfigMgr::unmarshal(&data) {
+ Ok(_) => panic!("legacy Wasabi endpoint must not be empty"),
+ Err(err) => err,
+ };
+ assert!(err.to_string().contains("missing Wasabi endpoint"), "{err}");
+
+ cfg.tiers.get_mut("COLD-WASABI").expect("Wasabi tier should exist").wasabi = None;
+ let data = serde_json::to_vec(&cfg).expect("missing-payload fixture should encode");
+ let err = match TierConfigMgr::unmarshal(&data) {
+ Ok(_) => panic!("legacy Wasabi payload must exist"),
+ Err(err) => err,
+ };
+ assert!(err.to_string().contains("missing Wasabi backend payload"), "{err}");
+ }
+
#[test]
fn test_tier_config_not_initialized_error_formats_operation_context() {
let err = tier_config_not_initialized_error("save tiering config");
@@ -3192,6 +3629,21 @@ mod tests {
assert!(err.message.contains("S3 tier configuration not found"), "{}", err.message);
}
+ #[tokio::test]
+ async fn test_add_rejects_custom_wasabi_endpoint_before_backend_setup() {
+ let mut mgr = empty_mgr();
+ let mut tier = build_wasabi_tier("COLD-WASABI");
+ tier.wasabi.as_mut().expect("Wasabi payload should exist").endpoint = "https://s3.example.invalid".to_string();
+
+ let err = mgr
+ .add(tier, true)
+ .await
+ .expect_err("custom Wasabi endpoint must fail before backend setup");
+ assert_eq!(err.code, ERR_TIER_INVALID_CONFIG.code);
+ assert_eq!(err.message, "Wasabi endpoint must be https://s3.ap-northeast-1.wasabisys.com");
+ assert!(mgr.tiers.is_empty());
+ }
+
#[tokio::test]
async fn test_add_rejects_reserved_names() {
// Supersedes the former `test_add_does_not_reserve_standard_name_regression_anchor`
@@ -3245,6 +3697,18 @@ mod tests {
assert_eq!(err.code, ERR_TIER_MISSING_CREDENTIALS.code);
}
+ #[tokio::test]
+ async fn test_edit_rejects_missing_credentials_for_wasabi() {
+ let mut mgr = empty_mgr();
+ mgr.tiers.insert("COLD-WASABI".to_string(), build_wasabi_tier("COLD-WASABI"));
+
+ let err = mgr
+ .edit("COLD-WASABI", TierCreds::default())
+ .await
+ .expect_err("empty Wasabi credentials must be rejected before backend setup");
+ assert_eq!(err.code, ERR_TIER_MISSING_CREDENTIALS.code);
+ }
+
#[tokio::test]
async fn test_edit_rejects_missing_credentials_for_minio() {
let mut mgr = empty_mgr();
@@ -4725,6 +5189,50 @@ mod tests {
);
}
+ #[test]
+ fn wasabi_backend_identity_uses_its_canonical_physical_destination() {
+ let mut wasabi = build_wasabi_tier("COLD-WASABI");
+ let wasabi_config = wasabi.wasabi.as_mut().expect("Wasabi payload should exist");
+ wasabi_config.endpoint = "https://s3.nl-1.wasabisys.com".to_string();
+ wasabi_config.prefix = "/archive//".to_string();
+ wasabi_config.region = "eu-central-1".to_string();
+
+ let mut physical_s3 = build_s3_tier("COLD-WASABI");
+ let s3 = physical_s3.s3.as_mut().expect("S3 payload should exist");
+ s3.endpoint = "https://s3.eu-central-1.wasabisys.com".to_string();
+ s3.bucket = wasabi_config.bucket.clone();
+ s3.prefix = "archive".to_string();
+ s3.region = wasabi_config.region.clone();
+
+ assert_eq!(
+ tier_backend_identity(&wasabi).expect("Wasabi identity should encode"),
+ tier_backend_identity(&physical_s3).expect("physical S3 identity should encode")
+ );
+ }
+
+ #[test]
+ fn in_place_wasabi_type_changes_are_rejected() {
+ let mut wasabi = build_wasabi_tier("COLD-WASABI");
+ wasabi.wasabi.as_mut().expect("Wasabi payload should exist").region = "us-east-1".to_string();
+ let mut s3 = build_s3_tier("COLD-WASABI");
+ let s3_config = s3.s3.as_mut().expect("S3 payload should exist");
+ s3_config.endpoint = "https://s3.wasabisys.com".to_string();
+ s3_config.bucket = "wasabi-bucket".to_string();
+ s3_config.prefix = "archive".to_string();
+ s3_config.region = "us-east-1".to_string();
+
+ for (current_tier, replacement_tier) in [(s3.clone_with_credentials(), wasabi.clone_with_credentials()), (wasabi, s3)] {
+ let mut current = empty_mgr();
+ current.tiers.insert("COLD-WASABI".to_string(), current_tier);
+ let mut replacement = empty_mgr();
+ replacement.tiers.insert("COLD-WASABI".to_string(), replacement_tier);
+
+ let err = replaced_tier_destinations(¤t, &replacement)
+ .expect_err("in-place type changes must not bypass mixed-version Wasabi fencing");
+ assert!(err.message.contains("add a new tier instead"), "{}", err.message);
+ }
+ }
+
#[test]
fn backend_identity_v2_has_stable_fixtures() {
assert_eq!(
@@ -4746,6 +5254,7 @@ mod tests {
#[test]
fn destination_identity_prefix_normalization_matches_driver_matrix() {
assert_eq!(normalized_tier_prefix(&TierType::S3, "/foo//"), "foo");
+ assert_eq!(normalized_tier_prefix(&TierType::Wasabi, "/foo//"), "foo");
for tier_type in [
TierType::RustFS,
TierType::MinIO,
diff --git a/crates/ecstore/src/services/tier/tier_config.rs b/crates/ecstore/src/services/tier/tier_config.rs
index 5d5717cb8..a5be98866 100644
--- a/crates/ecstore/src/services/tier/tier_config.rs
+++ b/crates/ecstore/src/services/tier/tier_config.rs
@@ -13,12 +13,22 @@
// limitations under the License.
use serde::{Deserialize, Serialize};
-use std::fmt::Display;
+use std::{fmt::Display, io};
use tracing::info;
const C_TIER_CONFIG_VER: &str = "v1";
const ERR_TIER_NAME_EMPTY: &str = "remote tier name empty";
+const WASABI_US_EAST_ENDPOINT: &str = "https://s3.wasabisys.com";
+const WASABI_ALTERNATIVE_ENDPOINTS: &[(&str, &str)] = &[
+ ("us-east-1", "https://s3.us-east-1.wasabisys.com"),
+ ("eu-central-1", "https://s3.nl-1.wasabisys.com"),
+ ("eu-central-2", "https://s3.de-1.wasabisys.com"),
+ ("eu-west-1", "https://s3.uk-1.wasabisys.com"),
+ ("eu-west-2", "https://s3.fr-1.wasabisys.com"),
+ ("eu-west-3", "https://s3.uk-2.wasabisys.com"),
+ ("eu-south-1", "https://s3.it-1.wasabisys.com"),
+];
#[derive(Serialize, Deserialize, Default, Debug, Clone)]
pub enum TierType {
@@ -26,6 +36,8 @@ pub enum TierType {
Unsupported,
#[serde(rename = "s3")]
S3,
+ #[serde(rename = "wasabi")]
+ Wasabi,
#[serde(rename = "rustfs")]
RustFS,
#[serde(rename = "minio")]
@@ -50,6 +62,9 @@ impl Display for TierType {
TierType::S3 => {
write!(f, "S3")
}
+ TierType::Wasabi => {
+ write!(f, "Wasabi")
+ }
TierType::RustFS => {
write!(f, "RustFS")
}
@@ -85,6 +100,7 @@ impl TierType {
pub fn new(sc_type: &str) -> Self {
match sc_type {
"S3" => TierType::S3,
+ "Wasabi" => TierType::Wasabi,
"RustFS" => TierType::RustFS,
"MinIO" => TierType::MinIO,
"Aliyun" => TierType::Aliyun,
@@ -100,6 +116,7 @@ impl TierType {
pub fn as_lowercase(&self) -> String {
match self {
TierType::S3 => "s3".to_string(),
+ TierType::Wasabi => "wasabi".to_string(),
TierType::RustFS => "rustfs".to_string(),
TierType::MinIO => "minio".to_string(),
TierType::Aliyun => "aliyun".to_string(),
@@ -124,6 +141,8 @@ pub struct TierConfig {
pub name: String,
#[serde(rename = "s3", skip_serializing_if = "Option::is_none")]
pub s3: Option,
+ #[serde(rename = "wasabi", skip_serializing_if = "Option::is_none")]
+ pub wasabi: Option,
#[serde(rename = "aliyun", skip_serializing_if = "Option::is_none")]
pub aliyun: Option,
#[serde(rename = "tencent", skip_serializing_if = "Option::is_none")]
@@ -145,6 +164,7 @@ pub struct TierConfig {
impl Clone for TierConfig {
fn clone(&self) -> TierConfig {
let mut s3 = None;
+ let mut wasabi = None;
let mut r = None;
let mut compatible_backend = None;
let mut aliyun = None;
@@ -161,6 +181,13 @@ impl Clone for TierConfig {
s3 = Some(s3_clone);
}
}
+ TierType::Wasabi => {
+ if let Some(wasabi_) = self.wasabi.as_ref() {
+ let mut wasabi_clone = wasabi_.clone();
+ wasabi_clone.secret_key = "REDACTED".to_string();
+ wasabi = Some(wasabi_clone);
+ }
+ }
TierType::RustFS => {
if let Some(r_) = self.rustfs.as_ref() {
let mut r_clone = r_.clone();
@@ -224,6 +251,7 @@ impl Clone for TierConfig {
tier_type: self.tier_type.clone(),
name: self.name.clone(),
s3,
+ wasabi,
rustfs: r,
minio: compatible_backend,
aliyun,
@@ -244,6 +272,7 @@ impl TierConfig {
tier_type: self.tier_type.clone(),
name: self.name.clone(),
s3: self.s3.clone(),
+ wasabi: self.wasabi.clone(),
aliyun: self.aliyun.clone(),
tencent: self.tencent.clone(),
huaweicloud: self.huaweicloud.clone(),
@@ -258,6 +287,7 @@ impl TierConfig {
fn endpoint(&self) -> String {
match self.tier_type {
TierType::S3 => self.s3.as_ref().map(|s| s.endpoint.clone()).unwrap_or_default(),
+ TierType::Wasabi => self.wasabi.as_ref().map(|w| w.endpoint.clone()).unwrap_or_default(),
TierType::RustFS => self.rustfs.as_ref().map(|r| r.endpoint.clone()).unwrap_or_default(),
TierType::MinIO => self.minio.as_ref().map(|m| m.endpoint.clone()).unwrap_or_default(),
TierType::Aliyun => self.aliyun.as_ref().map(|a| a.endpoint.clone()).unwrap_or_default(),
@@ -276,6 +306,7 @@ impl TierConfig {
fn bucket(&self) -> String {
match self.tier_type {
TierType::S3 => self.s3.as_ref().map(|s| s.bucket.clone()).unwrap_or_default(),
+ TierType::Wasabi => self.wasabi.as_ref().map(|w| w.bucket.clone()).unwrap_or_default(),
TierType::RustFS => self.rustfs.as_ref().map(|r| r.bucket.clone()).unwrap_or_default(),
TierType::MinIO => self.minio.as_ref().map(|m| m.bucket.clone()).unwrap_or_default(),
TierType::Aliyun => self.aliyun.as_ref().map(|a| a.bucket.clone()).unwrap_or_default(),
@@ -294,6 +325,7 @@ impl TierConfig {
fn prefix(&self) -> String {
match self.tier_type {
TierType::S3 => self.s3.as_ref().map(|s| s.prefix.clone()).unwrap_or_default(),
+ TierType::Wasabi => self.wasabi.as_ref().map(|w| w.prefix.clone()).unwrap_or_default(),
TierType::RustFS => self.rustfs.as_ref().map(|r| r.prefix.clone()).unwrap_or_default(),
TierType::MinIO => self.minio.as_ref().map(|m| m.prefix.clone()).unwrap_or_default(),
TierType::Aliyun => self.aliyun.as_ref().map(|a| a.prefix.clone()).unwrap_or_default(),
@@ -312,6 +344,7 @@ impl TierConfig {
fn region(&self) -> String {
match self.tier_type {
TierType::S3 => self.s3.as_ref().map(|s| s.region.clone()).unwrap_or_default(),
+ TierType::Wasabi => self.wasabi.as_ref().map(|w| w.region.clone()).unwrap_or_default(),
TierType::RustFS => self.rustfs.as_ref().map(|r| r.region.clone()).unwrap_or_default(),
TierType::MinIO => self.minio.as_ref().map(|m| m.region.clone()).unwrap_or_default(),
TierType::Aliyun => self.aliyun.as_ref().map(|a| a.region.clone()).unwrap_or_default(),
@@ -356,6 +389,73 @@ pub struct TierS3 {
pub aws_role_duration_seconds: i32,
}
+#[derive(Serialize, Deserialize, Default, Clone)]
+#[serde(deny_unknown_fields)]
+pub struct TierWasabi {
+ pub name: String,
+ #[serde(default)]
+ pub endpoint: String,
+ #[serde(rename = "accessKey")]
+ pub access_key: String,
+ #[serde(rename = "secretKey")]
+ pub secret_key: String,
+ pub bucket: String,
+ #[serde(default)]
+ pub prefix: String,
+ pub region: String,
+}
+
+impl std::fmt::Debug for TierWasabi {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("TierWasabi")
+ .field("name", &self.name)
+ .field("endpoint", &self.endpoint)
+ .field("access_key", &self.access_key)
+ .field("secret_key", &"REDACTED")
+ .field("bucket", &self.bucket)
+ .field("prefix", &self.prefix)
+ .field("region", &self.region)
+ .finish()
+ }
+}
+
+impl TierWasabi {
+ pub(crate) fn alternative_endpoint(&self) -> Option<&'static str> {
+ WASABI_ALTERNATIVE_ENDPOINTS
+ .iter()
+ .find_map(|(region, endpoint)| (*region == self.region).then_some(*endpoint))
+ }
+
+ pub(crate) fn canonical_endpoint(&self) -> io::Result {
+ let region = self.region.as_bytes();
+ let is_alphanumeric = |byte: &u8| byte.is_ascii_lowercase() || byte.is_ascii_digit();
+ if !(1..=63).contains(®ion.len())
+ || !region.first().is_some_and(is_alphanumeric)
+ || !region.last().is_some_and(is_alphanumeric)
+ || !region
+ .iter()
+ .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || *byte == b'-')
+ {
+ return Err(io::Error::other("invalid Wasabi region"));
+ }
+
+ let endpoint = if self.region == "us-east-1" {
+ WASABI_US_EAST_ENDPOINT.to_string()
+ } else {
+ format!("https://s3.{}.wasabisys.com", self.region)
+ };
+ if !self.endpoint.is_empty() && self.endpoint != endpoint && self.alternative_endpoint() != Some(self.endpoint.as_str()) {
+ return Err(io::Error::other(format!("Wasabi endpoint must be {endpoint}")));
+ }
+ Ok(endpoint)
+ }
+
+ pub(crate) fn normalize_endpoint(&mut self) -> io::Result<()> {
+ self.endpoint = self.canonical_endpoint()?;
+ Ok(())
+ }
+}
+
impl TierS3 {
#[allow(dead_code)]
fn create(
@@ -615,3 +715,200 @@ pub struct TierR2 {
pub prefix: String,
pub region: String,
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn wasabi_config() -> TierWasabi {
+ TierWasabi {
+ name: "COLD-WASABI".to_string(),
+ access_key: "access".to_string(),
+ secret_key: "secret".to_string(),
+ bucket: "archive".to_string(),
+ prefix: "objects".to_string(),
+ region: "ap-northeast-1".to_string(),
+ ..Default::default()
+ }
+ }
+
+ #[test]
+ fn wasabi_type_and_endpoint_are_canonical() {
+ let mut wasabi = wasabi_config();
+ let endpoint = wasabi.canonical_endpoint().expect("valid region should produce an endpoint");
+ assert_eq!(endpoint, "https://s3.ap-northeast-1.wasabisys.com");
+
+ wasabi
+ .normalize_endpoint()
+ .expect("valid region should normalize the endpoint");
+ assert_eq!(wasabi.endpoint, endpoint);
+ assert_eq!(
+ wasabi.canonical_endpoint().expect("canonical endpoint should be accepted"),
+ "https://s3.ap-northeast-1.wasabisys.com"
+ );
+ assert_eq!(TierType::new("Wasabi").to_string(), "Wasabi");
+ assert_eq!(TierType::Wasabi.as_lowercase(), "wasabi");
+
+ for (region, alternative, canonical) in [
+ ("us-east-1", "https://s3.us-east-1.wasabisys.com", "https://s3.wasabisys.com"),
+ ("eu-central-1", "https://s3.nl-1.wasabisys.com", "https://s3.eu-central-1.wasabisys.com"),
+ ("eu-central-2", "https://s3.de-1.wasabisys.com", "https://s3.eu-central-2.wasabisys.com"),
+ ("eu-west-1", "https://s3.uk-1.wasabisys.com", "https://s3.eu-west-1.wasabisys.com"),
+ ("eu-west-2", "https://s3.fr-1.wasabisys.com", "https://s3.eu-west-2.wasabisys.com"),
+ ("eu-west-3", "https://s3.uk-2.wasabisys.com", "https://s3.eu-west-3.wasabisys.com"),
+ ("eu-south-1", "https://s3.it-1.wasabisys.com", "https://s3.eu-south-1.wasabisys.com"),
+ ] {
+ let mut config = wasabi_config();
+ config.region = region.to_string();
+ config.endpoint = alternative.to_string();
+ config
+ .normalize_endpoint()
+ .expect("a documented Wasabi alternative endpoint should normalize");
+ assert_eq!(config.endpoint, canonical, "region {region}");
+ }
+ }
+
+ #[test]
+ fn wasabi_endpoint_rejects_unsafe_region_and_custom_endpoint() {
+ for region in [
+ "",
+ "US-EAST-1",
+ "-us-east-1",
+ "us-east-1-",
+ "us.east-1",
+ "us/east-1",
+ "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+ ] {
+ let mut wasabi = wasabi_config();
+ wasabi.region = region.to_string();
+ let err = wasabi.canonical_endpoint().expect_err("unsafe region must be rejected");
+ assert_eq!(err.to_string(), "invalid Wasabi region", "region {region}");
+ }
+
+ let mut wasabi = wasabi_config();
+ wasabi.endpoint = "https://s3.example.invalid".to_string();
+ let err = wasabi.canonical_endpoint().expect_err("custom endpoint must be rejected");
+ assert!(err.to_string().contains("https://s3.ap-northeast-1.wasabisys.com"));
+ }
+
+ #[test]
+ fn wasabi_config_rejects_unknown_fields_and_redacts_secret() {
+ let omitted_optional = serde_json::from_value::(serde_json::json!({
+ "name": "COLD-WASABI",
+ "accessKey": "access",
+ "secretKey": "secret",
+ "bucket": "archive",
+ "region": "us-east-1"
+ }))
+ .expect("endpoint and prefix should remain optional in Admin JSON");
+ assert!(omitted_optional.endpoint.is_empty());
+ assert!(omitted_optional.prefix.is_empty());
+ assert_eq!(
+ omitted_optional
+ .canonical_endpoint()
+ .expect("an omitted endpoint should be derived from the region"),
+ WASABI_US_EAST_ENDPOINT
+ );
+
+ let err = serde_json::from_value::(serde_json::json!({
+ "name": "COLD-WASABI",
+ "accessKey": "access",
+ "secretKey": "secret",
+ "bucket": "archive",
+ "region": "us-east-1",
+ "unexpected": true
+ }))
+ .expect_err("unknown Wasabi fields must be rejected");
+ assert!(err.to_string().contains("unknown field `unexpected`"), "{err}");
+
+ let err = serde_json::from_value::(serde_json::json!({
+ "name": "COLD-WASABI",
+ "accessKey": "access",
+ "bucket": "archive",
+ "region": "us-east-1"
+ }))
+ .expect_err("missing Wasabi credentials must be rejected during decoding");
+ assert!(err.to_string().contains("missing field `secretKey`"), "{err}");
+
+ let config = TierConfig {
+ tier_type: TierType::Wasabi,
+ wasabi: Some(wasabi_config()),
+ ..Default::default()
+ };
+ let redacted = config.clone();
+ assert_eq!(
+ redacted
+ .wasabi
+ .as_ref()
+ .expect("redacted Wasabi payload should remain")
+ .secret_key,
+ "REDACTED"
+ );
+ assert_eq!(
+ config
+ .clone_with_credentials()
+ .wasabi
+ .as_ref()
+ .expect("credential-bearing Wasabi payload should remain")
+ .secret_key,
+ "secret"
+ );
+
+ let mut debug_config = wasabi_config();
+ debug_config.secret_key = "wasabi-debug-secret-value".to_string();
+ let debug = format!("{debug_config:?}");
+ assert!(debug.contains("REDACTED"));
+ assert!(!debug.contains("wasabi-debug-secret-value"));
+ }
+
+ #[test]
+ fn wasabi_admin_json_shape_roundtrips_and_redacts() {
+ let mut wasabi = wasabi_config();
+ wasabi
+ .normalize_endpoint()
+ .expect("valid Wasabi configuration should normalize");
+ let config = TierConfig {
+ tier_type: TierType::Wasabi,
+ wasabi: Some(wasabi),
+ ..Default::default()
+ };
+ let expected = serde_json::json!({
+ "type": "wasabi",
+ "wasabi": {
+ "name": "COLD-WASABI",
+ "endpoint": "https://s3.ap-northeast-1.wasabisys.com",
+ "accessKey": "access",
+ "secretKey": "secret",
+ "bucket": "archive",
+ "prefix": "objects",
+ "region": "ap-northeast-1"
+ }
+ });
+
+ let encoded = serde_json::to_value(config.clone_with_credentials()).expect("Wasabi Admin JSON should encode");
+ assert_eq!(encoded, expected);
+ let decoded: TierConfig = serde_json::from_value(encoded).expect("Wasabi Admin JSON should decode");
+ assert!(matches!(decoded.tier_type, TierType::Wasabi));
+ let redacted = config.clone();
+ assert_eq!(
+ config
+ .wasabi
+ .as_ref()
+ .expect("source Wasabi payload should remain")
+ .secret_key,
+ "secret"
+ );
+ assert_eq!(
+ redacted
+ .wasabi
+ .as_ref()
+ .expect("redacted Wasabi payload should remain")
+ .secret_key,
+ "REDACTED"
+ );
+ assert_eq!(
+ serde_json::to_value(redacted).expect("redacted Wasabi JSON should encode")["wasabi"]["secretKey"],
+ "REDACTED"
+ );
+ }
+}
diff --git a/crates/ecstore/src/services/tier/warm_backend.rs b/crates/ecstore/src/services/tier/warm_backend.rs
index 9bf70acbd..62947cf24 100644
--- a/crates/ecstore/src/services/tier/warm_backend.rs
+++ b/crates/ecstore/src/services/tier/warm_backend.rs
@@ -37,6 +37,7 @@ use crate::services::tier::{
warm_backend_rustfs::WarmBackendRustFS,
warm_backend_s3::WarmBackendS3,
warm_backend_tencent::WarmBackendTencent,
+ warm_backend_wasabi::WarmBackendWasabi,
};
use bytes::Bytes;
use http::StatusCode;
@@ -235,6 +236,27 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result {
+ if let Some(wasabi_config) = tier.wasabi.as_ref() {
+ match WarmBackendWasabi::new(wasabi_config, &tier.name).await {
+ Ok(backend) => d = Some(Box::new(backend)),
+ Err(err) => {
+ warn!("{}", err);
+ return Err(AdminError {
+ code: "XRustFSAdminTierInvalidConfig".to_string(),
+ message: format!("Unable to setup remote tier, check tier configuration: {err}"),
+ status_code: StatusCode::BAD_REQUEST,
+ });
+ }
+ }
+ } else {
+ return Err(AdminError {
+ code: "XRustFSAdminTierInvalidConfig".to_string(),
+ message: "Wasabi tier configuration not found".to_string(),
+ status_code: StatusCode::BAD_REQUEST,
+ });
+ }
+ }
TierType::RustFS => {
if let Some(rustfs_config) = tier.rustfs.as_ref() {
let dd = WarmBackendRustFS::new(rustfs_config, &tier.name).await;
@@ -400,16 +422,22 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result panic!("probing a Wasabi backend must validate its credentials"),
+ Err(err) => err,
+ };
+ assert_eq!(err.code, ERR_TIER_INVALID_CONFIG.code);
+ }
+
#[test]
fn build_transition_put_options_preserves_content_headers() {
let mut metadata = HashMap::new();
diff --git a/crates/ecstore/src/services/tier/warm_backend_s3.rs b/crates/ecstore/src/services/tier/warm_backend_s3.rs
index 87c0d5a93..9b464ea8e 100644
--- a/crates/ecstore/src/services/tier/warm_backend_s3.rs
+++ b/crates/ecstore/src/services/tier/warm_backend_s3.rs
@@ -25,9 +25,9 @@ use url::Url;
use crate::client::{
api_get_options::GetObjectOptions,
api_put_object::PutObjectOptions,
- api_remove::RemoveObjectOptions,
+ api_remove::{RemoveObjectOptions, RemoveObjectResult},
credentials::{Credentials, SignatureType, Static, Value},
- transition_api::{Options, TransitionClient, TransitionCore},
+ transition_api::{BucketLookupType, Options, TransitionClient, TransitionCore},
transition_api::{ReadCloser, ReaderImpl},
};
use crate::error::ErrorResponse;
@@ -36,6 +36,7 @@ use crate::services::tier::{
tier_config::TierS3,
warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
};
+use http::HeaderMap;
use rustfs_utils::egress::validate_outbound_url;
use rustfs_utils::path::SLASH_SEPARATOR;
@@ -48,7 +49,15 @@ pub struct WarmBackendS3 {
}
impl WarmBackendS3 {
- pub async fn new(conf: &TierS3, tier: &str) -> Result {
+ pub async fn new(conf: &TierS3, _tier: &str) -> Result {
+ Self::new_with_bucket_lookup(conf, BucketLookupType::BucketLookupAuto, "s3").await
+ }
+
+ pub(crate) async fn new_with_bucket_lookup(
+ conf: &TierS3,
+ bucket_lookup: BucketLookupType,
+ tier_type: &str,
+ ) -> Result {
let u = match Url::parse(&conf.endpoint) {
Ok(u) => u,
Err(err) => {
@@ -94,12 +103,13 @@ impl WarmBackendS3 {
creds,
secure: u.scheme() == "https",
region: conf.region.clone(),
+ bucket_lookup,
..Default::default()
};
let host = u
.host()
.ok_or_else(|| std::io::Error::other("Invalid endpoint URL: missing host"))?;
- let client = TransitionClient::new(&host.to_string(), opts, "s3").await?;
+ let client = TransitionClient::new(&host.to_string(), opts, tier_type).await?;
let client = Arc::new(client);
let core = TransitionCore(Arc::clone(&client));
@@ -119,6 +129,36 @@ impl WarmBackendS3 {
}
return dest_obj;
}
+
+ pub(crate) async fn remove_with_result(&self, object: &str, rv: &str) -> Result {
+ let mut opts = RemoveObjectOptions::default();
+ if !rv.is_empty() {
+ opts.version_id = rv.to_string();
+ }
+ self.client
+ .remove_object_inner(&self.bucket, &self.get_dest(object), opts)
+ .await
+ }
+
+ pub(crate) async fn get_with_headers(
+ &self,
+ object: &str,
+ rv: &str,
+ opts: WarmBackendGetOpts,
+ ) -> Result<(HeaderMap, ReadCloser), std::io::Error> {
+ let mut gopts = GetObjectOptions::default();
+
+ if !rv.is_empty() {
+ gopts.version_id = rv.to_string();
+ }
+ if opts.start_offset >= 0 && opts.length > 0 {
+ gopts
+ .set_range(opts.start_offset, opts.start_offset + opts.length - 1)
+ .map_err(std::io::Error::other)?;
+ }
+ let (_, headers, reader) = self.core.get_object(&self.bucket, &self.get_dest(object), &gopts).await?;
+ Ok((headers, reader))
+ }
}
#[cfg(test)]
@@ -168,32 +208,11 @@ impl WarmBackend for WarmBackendS3 {
}
async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result {
- let mut gopts = GetObjectOptions::default();
-
- if rv != "" {
- gopts.version_id = rv.to_string();
- }
- if opts.start_offset >= 0 && opts.length > 0 {
- if let Err(err) = gopts.set_range(opts.start_offset, opts.start_offset + opts.length - 1) {
- return Err(std::io::Error::other(err));
- }
- }
- let c = TransitionCore(Arc::clone(&self.client));
- let (_, _, r) = c.get_object(&self.bucket, &self.get_dest(object), &gopts).await?;
-
- Ok(r)
+ self.get_with_headers(object, rv, opts).await.map(|(_, reader)| reader)
}
async fn remove(&self, object: &str, rv: &str) -> Result<(), std::io::Error> {
- let mut ropts = RemoveObjectOptions::default();
- if rv != "" {
- ropts.version_id = rv.to_string();
- }
- let client = self.client.clone();
- match client.remove_object(&self.bucket, &self.get_dest(object), ropts).await {
- None => Ok(()),
- Some(err) => Err(std::io::Error::other(err)),
- }
+ self.remove_with_result(object, rv).await.map(|_| ())
}
async fn in_use(&self) -> Result {
diff --git a/crates/ecstore/src/services/tier/warm_backend_wasabi.rs b/crates/ecstore/src/services/tier/warm_backend_wasabi.rs
new file mode 100644
index 000000000..cb367bcdc
--- /dev/null
+++ b/crates/ecstore/src/services/tier/warm_backend_wasabi.rs
@@ -0,0 +1,1004 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::{
+ collections::HashMap,
+ io,
+ sync::atomic::{AtomicBool, Ordering},
+};
+
+use s3s::header::{X_AMZ_DELETE_MARKER, X_AMZ_VERSION_ID};
+use uuid::Uuid;
+
+use crate::{
+ client::transition_api::{BucketLookupType, ReadCloser, ReaderImpl},
+ services::tier::{
+ tier_config::{TierS3, TierWasabi},
+ warm_backend::{WarmBackend, WarmBackendGetOpts},
+ warm_backend_s3::WarmBackendS3,
+ },
+};
+
+const WASABI_VERSIONING_DRIFT_ERROR: &str = "Wasabi tier bucket versioning changed after configuration";
+
+pub struct WarmBackendWasabi {
+ s3: WarmBackendS3,
+ versioning_checked: AtomicBool,
+ versioning_drifted: AtomicBool,
+}
+
+impl WarmBackendWasabi {
+ pub async fn new(conf: &TierWasabi, _tier: &str) -> io::Result {
+ let endpoint = conf.canonical_endpoint()?;
+ let s3 = TierS3 {
+ name: conf.name.clone(),
+ endpoint,
+ access_key: conf.access_key.clone(),
+ secret_key: conf.secret_key.clone(),
+ bucket: conf.bucket.clone(),
+ prefix: conf.prefix.clone(),
+ region: conf.region.clone(),
+ ..Default::default()
+ };
+ let backend = WarmBackendS3::new_with_bucket_lookup(&s3, BucketLookupType::BucketLookupPath, "wasabi").await?;
+ Ok(Self {
+ s3: backend,
+ versioning_checked: AtomicBool::new(false),
+ versioning_drifted: AtomicBool::new(false),
+ })
+ }
+
+ fn drift_error() -> io::Error {
+ io::Error::other(WASABI_VERSIONING_DRIFT_ERROR)
+ }
+
+ fn normalized_remote_version(remote_version: &str) -> &str {
+ if remote_version.is_empty() || Uuid::parse_str(remote_version).is_ok_and(|version_id| version_id.is_nil()) {
+ ""
+ } else {
+ remote_version
+ }
+ }
+
+ async fn check_remote_bucket_unversioned(&self) -> io::Result<()> {
+ if self.versioning_drifted.load(Ordering::Acquire) {
+ return Err(Self::drift_error());
+ }
+ let config = self.s3.client.get_bucket_versioning(&self.s3.bucket).await?;
+ if config.status.is_some() || config.mfa_delete.is_some() {
+ self.versioning_drifted.store(true, Ordering::Release);
+ return Err(io::Error::other("Wasabi tier bucket must be unversioned"));
+ }
+ self.versioning_checked.store(true, Ordering::Release);
+ if self.versioning_drifted.load(Ordering::Acquire) {
+ return Err(Self::drift_error());
+ }
+ Ok(())
+ }
+
+ async fn ensure_unversioned(&self) -> io::Result<()> {
+ if self.versioning_drifted.load(Ordering::Acquire) {
+ return Err(Self::drift_error());
+ }
+ if self.versioning_checked.load(Ordering::Acquire) {
+ return Ok(());
+ }
+ self.check_remote_bucket_unversioned().await
+ }
+}
+
+#[async_trait::async_trait]
+impl WarmBackend for WarmBackendWasabi {
+ async fn validate(&self) -> io::Result<()> {
+ self.check_remote_bucket_unversioned().await
+ }
+
+ fn validate_remote_version_id(&self, remote_version_id: &str) -> io::Result<()> {
+ if remote_version_id.is_empty() {
+ return Ok(());
+ }
+ self.versioning_drifted.store(true, Ordering::Release);
+ Err(io::Error::other(
+ "Wasabi returned an object version ID for a tier that requires an unversioned bucket",
+ ))
+ }
+
+ async fn put(&self, object: &str, r: ReaderImpl, length: i64) -> io::Result {
+ self.ensure_unversioned().await?;
+ self.s3.put(object, r, length).await
+ }
+
+ async fn put_with_meta(&self, object: &str, r: ReaderImpl, length: i64, meta: HashMap) -> io::Result {
+ self.ensure_unversioned().await?;
+ self.s3.put_with_meta(object, r, length, meta).await
+ }
+
+ async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> io::Result {
+ let rv = Self::normalized_remote_version(rv);
+ if rv.is_empty() {
+ self.ensure_unversioned().await?;
+ }
+ let (headers, reader) = self.s3.get_with_headers(object, rv, opts).await?;
+ if rv.is_empty() && (headers.contains_key(X_AMZ_VERSION_ID) || headers.contains_key(X_AMZ_DELETE_MARKER)) {
+ self.versioning_drifted.store(true, Ordering::Release);
+ return Err(io::Error::other(
+ "Wasabi bucket versioning changed while reading an unversioned tier object",
+ ));
+ }
+ Ok(reader)
+ }
+
+ async fn remove(&self, object: &str, rv: &str) -> io::Result<()> {
+ let rv = Self::normalized_remote_version(rv);
+ if !rv.is_empty() {
+ return self.s3.remove(object, rv).await;
+ }
+ self.ensure_unversioned().await?;
+
+ let result = self.s3.remove_with_result(object, rv).await?;
+ if result.delete_marker || !result.delete_marker_version_id.is_empty() {
+ self.versioning_drifted.store(true, Ordering::Release);
+ if !result.delete_marker_version_id.is_empty()
+ && let Err(rollback_error) = self.s3.remove(object, &result.delete_marker_version_id).await
+ {
+ return Err(io::Error::other(format!(
+ "{WASABI_VERSIONING_DRIFT_ERROR}; failed to remove the unexpected delete marker: {rollback_error}"
+ )));
+ }
+ return Err(io::Error::other(WASABI_VERSIONING_DRIFT_ERROR));
+ }
+ Ok(())
+ }
+
+ async fn remove_exact(&self, object: &str, rv: &str) -> io::Result<()> {
+ if rv.is_empty() {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ "an exact Wasabi tier delete requires a remote version ID",
+ ));
+ }
+ self.s3.remove(object, rv).await
+ }
+
+ async fn in_use(&self) -> io::Result {
+ self.check_remote_bucket_unversioned().await?;
+ let in_use = self.s3.in_use().await?;
+ if self.versioning_drifted.load(Ordering::Acquire) {
+ return Err(Self::drift_error());
+ }
+ Ok(in_use)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::client::{
+ credentials::{Credentials, SignatureType, Static, Value},
+ transition_api::{Options, TransitionCore},
+ };
+ use std::sync::{Arc, atomic::AtomicUsize};
+ use tokio::{
+ io::{AsyncReadExt, AsyncWriteExt},
+ net::TcpListener,
+ };
+
+ #[derive(Clone, Copy)]
+ struct TestResponse {
+ status: &'static str,
+ headers: &'static str,
+ body: &'static str,
+ }
+
+ #[derive(Debug)]
+ struct CapturedRequest {
+ head: String,
+ body: Vec,
+ }
+
+ fn test_config() -> TierWasabi {
+ TierWasabi {
+ name: "WASABI".to_string(),
+ endpoint: String::new(),
+ access_key: "access-key".to_string(),
+ secret_key: "secret-key".to_string(),
+ bucket: "tier-bucket".to_string(),
+ prefix: "/archive/".to_string(),
+ region: "ap-northeast-1".to_string(),
+ }
+ }
+
+ async fn backend_for_endpoint(endpoint: &str, max_retries: i64) -> WarmBackendWasabi {
+ let client = Arc::new(
+ crate::client::transition_api::TransitionClient::new(
+ endpoint,
+ Options {
+ creds: Credentials::new(Static(Value {
+ access_key_id: "access-key".to_string(),
+ secret_access_key: "secret-key".to_string(),
+ signer_type: SignatureType::SignatureV4,
+ ..Default::default()
+ })),
+ region: "us-east-1".to_string(),
+ bucket_lookup: BucketLookupType::BucketLookupPath,
+ max_retries,
+ ..Default::default()
+ },
+ "wasabi",
+ )
+ .await
+ .expect("test transition client should initialize"),
+ );
+ WarmBackendWasabi {
+ s3: WarmBackendS3 {
+ core: TransitionCore(client.clone()),
+ client,
+ bucket: "tier-bucket".to_string(),
+ prefix: "archive".to_string(),
+ storage_class: String::new(),
+ },
+ versioning_checked: AtomicBool::new(true),
+ versioning_drifted: AtomicBool::new(false),
+ }
+ }
+
+ async fn backend_with_responses(
+ responses: Vec,
+ ) -> Option<(WarmBackendWasabi, tokio::task::JoinHandle>)> {
+ let listener = match TcpListener::bind("127.0.0.1:0").await {
+ Ok(listener) => listener,
+ Err(err) if err.kind() == io::ErrorKind::PermissionDenied => return None,
+ Err(err) => panic!("test listener should bind: {err}"),
+ };
+ let endpoint = listener
+ .local_addr()
+ .expect("test listener address should resolve")
+ .to_string();
+ let request_task = tokio::spawn(async move {
+ let mut requests = Vec::with_capacity(responses.len());
+ for response in responses {
+ let (mut stream, _) = listener.accept().await.expect("S3 request should connect");
+ let mut request = Vec::new();
+ let mut buffer = [0_u8; 1024];
+ loop {
+ let read = stream.read(&mut buffer).await.expect("S3 request should be readable");
+ assert_ne!(read, 0, "connection closed before request headers were received");
+ request.extend_from_slice(&buffer[..read]);
+ if request.windows(4).any(|window| window == b"\r\n\r\n") {
+ break;
+ }
+ }
+ let header_end = request
+ .windows(4)
+ .position(|window| window == b"\r\n\r\n")
+ .expect("request should contain a complete header")
+ + 4;
+ let head = String::from_utf8(request[..header_end].to_vec()).expect("S3 request header should be UTF-8");
+ let content_length = head
+ .lines()
+ .filter_map(|line| line.split_once(':'))
+ .find_map(|(name, value)| {
+ name.eq_ignore_ascii_case("content-length")
+ .then(|| value.trim().parse::().expect("Content-Length should be numeric"))
+ })
+ .unwrap_or_default();
+ let request_end = header_end
+ .checked_add(content_length)
+ .expect("request length should not overflow");
+ while request.len() < request_end {
+ let remaining = request_end - request.len();
+ let read_len = remaining.min(buffer.len());
+ let read = stream
+ .read(&mut buffer[..read_len])
+ .await
+ .expect("S3 request body should be readable");
+ assert_ne!(read, 0, "connection closed before the complete request body was received");
+ request.extend_from_slice(&buffer[..read]);
+ }
+ assert_eq!(request.len(), request_end, "fixture must capture exactly one request body");
+ let wire_response = format!(
+ "HTTP/1.1 {}\r\nContent-Length: {}\r\n{}Connection: close\r\n\r\n{}",
+ response.status,
+ response.body.len(),
+ response.headers,
+ response.body
+ );
+ stream
+ .write_all(wire_response.as_bytes())
+ .await
+ .expect("S3 response should be writable");
+ requests.push(CapturedRequest {
+ head,
+ body: request[header_end..].to_vec(),
+ });
+ }
+ requests
+ });
+ let backend = backend_for_endpoint(&endpoint, 1).await;
+ Some((backend, request_task))
+ }
+
+ fn request_target(request: &CapturedRequest) -> &str {
+ request
+ .head
+ .lines()
+ .next()
+ .and_then(|line| line.split_whitespace().nth(1))
+ .expect("request should contain a target")
+ }
+
+ fn request_header<'a>(request: &'a CapturedRequest, name: &str) -> Option<&'a str> {
+ request
+ .head
+ .lines()
+ .filter_map(|line| line.split_once(':'))
+ .find_map(|(header, value)| header.eq_ignore_ascii_case(name).then(|| value.trim()))
+ }
+
+ #[tokio::test]
+ async fn new_uses_canonical_path_style_sigv4_client() {
+ let backend = WarmBackendWasabi::new(&test_config(), "WASABI")
+ .await
+ .expect("Wasabi backend should initialize without network I/O");
+ let mut credentials = backend
+ .s3
+ .client
+ .creds_provider
+ .lock()
+ .expect("test credentials lock should not be poisoned");
+ let credentials = credentials.get().expect("static Wasabi credentials should resolve");
+
+ assert_eq!(backend.s3.client.lookup, BucketLookupType::BucketLookupPath);
+ assert_eq!(backend.s3.client.endpoint_url.scheme(), "https");
+ assert_eq!(backend.s3.client.endpoint_url.host_str(), Some("s3.ap-northeast-1.wasabisys.com"));
+ assert_eq!(backend.s3.client.tier_type, "wasabi");
+ assert_eq!(credentials.signer_type, SignatureType::SignatureV4);
+ assert_eq!(backend.s3.prefix, "archive");
+ assert!(!backend.versioning_checked.load(Ordering::Acquire));
+ assert!(!backend.versioning_drifted.load(Ordering::Acquire));
+ }
+
+ #[tokio::test]
+ async fn validate_reads_remote_bucket_versioning_and_fails_closed() {
+ for (body, should_succeed) in [
+ (r#""#, true),
+ (
+ r#"Enabled"#,
+ false,
+ ),
+ (
+ r#"Suspended"#,
+ false,
+ ),
+ (
+ r#"FutureStatus"#,
+ false,
+ ),
+ (
+ r#"Enabled"#,
+ false,
+ ),
+ ] {
+ let Some((backend, request_task)) = backend_with_responses(vec![TestResponse {
+ status: "200 OK",
+ headers: "",
+ body,
+ }])
+ .await
+ else {
+ return;
+ };
+
+ let result = backend.validate().await;
+ let requests = request_task.await.expect("versioning request task should finish");
+
+ assert_eq!(result.is_ok(), should_succeed, "unexpected validation result for {body}");
+ assert_eq!(requests.len(), 1);
+ let target = request_target(&requests[0]);
+ assert!(target.starts_with("/tier-bucket"), "{target}");
+ assert!(target.contains("versioning"), "{target}");
+ }
+ }
+
+ #[tokio::test]
+ async fn empty_remote_version_get_and_delete_omit_version_id() {
+ for remote_version in [String::new(), Uuid::nil().to_string()] {
+ let Some((backend, request_task)) = backend_with_responses(vec![
+ TestResponse {
+ status: "200 OK",
+ headers: "ETag: \"etag\"\r\n",
+ body: "remote body",
+ },
+ TestResponse {
+ status: "204 No Content",
+ headers: "",
+ body: "",
+ },
+ ])
+ .await
+ else {
+ return;
+ };
+
+ let mut reader = backend
+ .get("remote-object", &remote_version, WarmBackendGetOpts::default())
+ .await
+ .expect("unversioned Wasabi object should be readable");
+ let mut body = String::new();
+ reader
+ .read_to_string(&mut body)
+ .await
+ .expect("remote object body should be readable");
+ backend
+ .remove("remote-object", &remote_version)
+ .await
+ .expect("unversioned Wasabi object should be removable");
+ let requests = request_task.await.expect("GET and DELETE request task should finish");
+
+ assert_eq!(body, "remote body");
+ assert_eq!(requests.len(), 2);
+ assert_eq!(request_target(&requests[0]), "/tier-bucket/archive/remote-object");
+ assert_eq!(request_target(&requests[1]), "/tier-bucket/archive/remote-object");
+ assert!(requests[0].head.starts_with("GET "));
+ assert!(requests[1].head.starts_with("DELETE "));
+ }
+ }
+
+ #[tokio::test]
+ async fn exact_nil_put_response_delete_preserves_version_id() {
+ let Some((backend, request_task)) = backend_with_responses(vec![TestResponse {
+ status: "204 No Content",
+ headers: "",
+ body: "",
+ }])
+ .await
+ else {
+ return;
+ };
+ let remote_version = Uuid::nil().to_string();
+
+ backend
+ .remove_exact("remote-object", &remote_version)
+ .await
+ .expect("an exact rejected PUT response should be removable");
+ let requests = request_task.await.expect("DELETE request task should finish");
+
+ assert_eq!(requests.len(), 1);
+ assert!(requests[0].head.starts_with("DELETE "));
+ assert_eq!(
+ request_target(&requests[0]),
+ format!("/tier-bucket/archive/remote-object?versionId={remote_version}")
+ );
+ }
+
+ #[tokio::test]
+ async fn empty_remote_version_get_rejects_versioned_response() {
+ // An empty `x-amz-version-id` is rejected one layer down by the shared remote-version
+ // validator, so it fails closed with that validator's message instead of the drift message.
+ for (response_headers, expected_error) in [
+ ("ETag: \"etag\"\r\nx-amz-version-id: unexpected-version\r\n", "bucket versioning changed"),
+ ("ETag: \"etag\"\r\nx-amz-version-id:\r\n", "empty object version id header"),
+ ("ETag: \"etag\"\r\nx-amz-delete-marker: true\r\n", "bucket versioning changed"),
+ ] {
+ let Some((backend, request_task)) = backend_with_responses(vec![TestResponse {
+ status: "200 OK",
+ headers: response_headers,
+ body: "newer body",
+ }])
+ .await
+ else {
+ return;
+ };
+
+ let err = backend
+ .get("remote-object", "", WarmBackendGetOpts::default())
+ .await
+ .expect_err("a versioned GET response must not return a possibly newer object");
+ let requests = request_task.await.expect("GET request task should finish");
+
+ assert!(err.to_string().contains(expected_error), "{err}");
+ assert_eq!(requests.len(), 1);
+ assert!(requests[0].head.starts_with("GET "));
+ assert_eq!(request_target(&requests[0]), "/tier-bucket/archive/remote-object");
+ }
+ }
+
+ #[tokio::test]
+ async fn nonempty_remote_version_get_and_delete_send_exact_version_id() {
+ let Some((backend, request_task)) = backend_with_responses(vec![
+ TestResponse {
+ status: "200 OK",
+ headers: "ETag: \"etag\"\r\n",
+ body: "legacy version body",
+ },
+ TestResponse {
+ status: "204 No Content",
+ headers: "",
+ body: "",
+ },
+ ])
+ .await
+ else {
+ return;
+ };
+
+ let mut reader = backend
+ .get("remote-object", "remote-version-1", WarmBackendGetOpts::default())
+ .await
+ .expect("legacy exact-version GET should succeed");
+ let mut body = String::new();
+ reader
+ .read_to_string(&mut body)
+ .await
+ .expect("legacy exact-version body should be readable");
+ backend
+ .remove("remote-object", "remote-version-1")
+ .await
+ .expect("versioned Wasabi object should use exact-version deletion");
+ let requests = request_task.await.expect("GET and DELETE request task should finish");
+
+ assert_eq!(body, "legacy version body");
+ assert_eq!(requests.len(), 2);
+ assert_eq!(
+ request_target(&requests[0]),
+ "/tier-bucket/archive/remote-object?versionId=remote-version-1"
+ );
+ assert_eq!(request_target(&requests[1]), request_target(&requests[0]));
+ assert!(requests[0].head.starts_with("GET "));
+ assert!(requests[1].head.starts_with("DELETE "));
+ }
+
+ #[tokio::test]
+ async fn exact_delete_preserves_remote_no_such_version_code() {
+ let Some((backend, request_task)) = backend_with_responses(vec![TestResponse {
+ status: "404 Not Found",
+ headers: "x-amz-request-id: request-id\r\n",
+ body: "NoSuchVersionThe specified version does not exist.",
+ }])
+ .await
+ else {
+ return;
+ };
+
+ let err = backend
+ .remove("remote-object", "missing-version")
+ .await
+ .expect_err("a missing exact version should retain its S3 error code");
+ let requests = request_task.await.expect("DELETE request task should finish");
+
+ assert!(err.to_string().contains("NoSuchVersion"), "{err}");
+ assert_eq!(requests.len(), 1);
+ assert_eq!(
+ request_target(&requests[0]),
+ "/tier-bucket/archive/remote-object?versionId=missing-version"
+ );
+ }
+
+ #[tokio::test]
+ async fn non_success_response_is_not_retried() {
+ let listener = match TcpListener::bind("127.0.0.1:0").await {
+ Ok(listener) => listener,
+ Err(err) if err.kind() == io::ErrorKind::PermissionDenied => return,
+ Err(err) => panic!("test listener should bind: {err}"),
+ };
+ let endpoint = listener
+ .local_addr()
+ .expect("test listener address should resolve")
+ .to_string();
+ let request_count = Arc::new(AtomicUsize::new(0));
+ let server_count = request_count.clone();
+ let server = tokio::spawn(async move {
+ loop {
+ let Ok((mut stream, _)) = listener.accept().await else {
+ return;
+ };
+ server_count.fetch_add(1, Ordering::AcqRel);
+ let mut request = Vec::new();
+ let mut buffer = [0_u8; 1024];
+ while !request.windows(4).any(|window| window == b"\r\n\r\n") {
+ let read = stream.read(&mut buffer).await.expect("request should be readable");
+ if read == 0 {
+ return;
+ }
+ request.extend_from_slice(&buffer[..read]);
+ }
+ stream
+ .write_all(
+ b"HTTP/1.1 500 Internal Server Error\r\nContent-Length: 41\r\nConnection: close\r\n\r\nInternalError",
+ )
+ .await
+ .expect("error response should be writable");
+ }
+ });
+ let backend = backend_for_endpoint(&endpoint, 4).await;
+
+ backend
+ .remove("remote-object", "exact-version")
+ .await
+ .expect_err("the first non-success response must be returned");
+ tokio::task::yield_now().await;
+ assert_eq!(request_count.load(Ordering::Acquire), 1);
+ server.abort();
+ let _ = server.await;
+ }
+
+ #[tokio::test]
+ async fn production_error_response_enforces_the_small_body_limit() {
+ let listener = match TcpListener::bind("127.0.0.1:0").await {
+ Ok(listener) => listener,
+ Err(err) if err.kind() == io::ErrorKind::PermissionDenied => return,
+ Err(err) => panic!("test listener should bind: {err}"),
+ };
+ let endpoint = listener
+ .local_addr()
+ .expect("test listener address should resolve")
+ .to_string();
+ let server = tokio::spawn(async move {
+ let (mut stream, _) = listener.accept().await.expect("S3 request should connect");
+ let mut request = Vec::new();
+ let mut buffer = [0_u8; 1024];
+ while !request.windows(4).any(|window| window == b"\r\n\r\n") {
+ let read = stream.read(&mut buffer).await.expect("request should be readable");
+ assert_ne!(read, 0, "connection closed before request headers were received");
+ request.extend_from_slice(&buffer[..read]);
+ }
+ let body = vec![b'x'; crate::client::transition_api::MAX_S3_ERROR_RESPONSE_SIZE + 1];
+ let head = format!(
+ "HTTP/1.1 500 Internal Server Error\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+ body.len()
+ );
+ stream
+ .write_all(head.as_bytes())
+ .await
+ .expect("response headers should be writable");
+ stream.write_all(&body).await.expect("response body should be writable");
+ });
+ let backend = backend_for_endpoint(&endpoint, 1).await;
+
+ let err = backend
+ .remove("remote-object", "exact-version")
+ .await
+ .expect_err("an oversized error response must fail closed");
+ server.await.expect("error response task should finish");
+
+ assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+ }
+
+ #[tokio::test]
+ async fn remote_error_fields_are_sanitized() {
+ let Some((backend, request_task)) = backend_with_responses(vec![TestResponse {
+ status: "404 Not Found",
+ headers: "x-amz-request-id: request-id\r\nx-minio-error-code: secret-header-value\r\n",
+ body: "NoSuchVersionsecret-body-messagesecret-requestsecret-host",
+ }])
+ .await
+ else {
+ return;
+ };
+
+ let err = backend
+ .remove("remote-object", "missing-version")
+ .await
+ .expect_err("the malicious remote error should remain an error");
+ let requests = request_task.await.expect("DELETE request task should finish");
+ let rendered = format!("{err} {err:?}");
+
+ assert!(rendered.contains("ResponseInterrupted"), "{rendered}");
+ for secret in ["secret-header-value", "secret-body-message", "secret-request", "secret-host"] {
+ assert!(!rendered.contains(secret), "remote error field leaked: {rendered}");
+ }
+ assert_eq!(requests.len(), 1);
+ }
+
+ #[tokio::test]
+ async fn empty_version_delete_checks_the_real_s3_response_for_markers() {
+ for (response_headers, marker_version) in [
+ ("x-amz-delete-marker: true\r\n", None),
+ ("x-amz-version-id: marker-version\r\n", Some("marker-version")),
+ ] {
+ let mut responses = vec![TestResponse {
+ status: "204 No Content",
+ headers: response_headers,
+ body: "",
+ }];
+ if marker_version.is_some() {
+ responses.push(TestResponse {
+ status: "204 No Content",
+ headers: "",
+ body: "",
+ });
+ }
+ responses.push(TestResponse {
+ status: "204 No Content",
+ headers: "",
+ body: "",
+ });
+ let Some((backend, request_task)) = backend_with_responses(responses).await else {
+ return;
+ };
+ let err = backend
+ .remove("remote-object", "")
+ .await
+ .expect_err("a delete-marker response must retain the local cleanup record");
+ backend
+ .remove("remote-object", "")
+ .await
+ .expect_err("a detected versioning drift must stop subsequent versionless deletes");
+ backend
+ .put("new-object", ReaderImpl::Body(hyper::body::Bytes::new()), 0)
+ .await
+ .expect_err("a detected versioning drift must stop subsequent uploads");
+ backend
+ .in_use()
+ .await
+ .expect_err("a detected versioning drift must make usage checks fail closed");
+ backend
+ .remove("remote-object", "legacy-version")
+ .await
+ .expect("exact cleanup must remain available after versioning drift");
+ let requests = request_task.await.expect("delete request task should finish");
+
+ assert!(err.to_string().contains("bucket versioning changed"), "{err}");
+ assert_eq!(requests.len(), 2 + usize::from(marker_version.is_some()));
+ assert!(requests[0].head.starts_with("DELETE "));
+ assert_eq!(request_target(&requests[0]), "/tier-bucket/archive/remote-object");
+ if let Some(marker_version) = marker_version {
+ assert_eq!(
+ request_target(&requests[1]),
+ format!("/tier-bucket/archive/remote-object?versionId={marker_version}")
+ );
+ }
+ assert_eq!(
+ request_target(requests.last().expect("exact cleanup request should exist")),
+ "/tier-bucket/archive/remote-object?versionId=legacy-version"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn in_use_delegates_to_the_path_style_s3_listing() {
+ let Some((backend, request_task)) = backend_with_responses(vec![
+ TestResponse {
+ status: "200 OK",
+ headers: "",
+ body: r#""#,
+ },
+ TestResponse {
+ status: "200 OK",
+ headers: "",
+ body: r#""#,
+ },
+ ])
+ .await
+ else {
+ return;
+ };
+
+ assert!(!backend.in_use().await.expect("an empty Wasabi prefix should not be in use"));
+ let requests = request_task.await.expect("LIST request task should finish");
+
+ assert_eq!(requests.len(), 2);
+ assert!(request_target(&requests[0]).contains("versioning"));
+ assert!(requests[1].head.starts_with("GET "));
+ let target = request_target(&requests[1]);
+ // Path-style bucket-level listing addresses the bucket root, so the target keeps the
+ // trailing slash before the query string.
+ assert!(target.starts_with("/tier-bucket/?"), "{target}");
+ assert!(target.contains("list-type=2"), "{target}");
+ assert!(target.contains("prefix=archive"), "{target}");
+ assert!(target.contains("max-keys=1"), "{target}");
+ }
+
+ #[tokio::test]
+ async fn remote_version_validation_rejects_every_nonempty_put_version() {
+ let backend = WarmBackendWasabi::new(&test_config(), "WASABI")
+ .await
+ .expect("Wasabi backend should initialize without network I/O");
+
+ assert!(backend.validate_remote_version_id("").is_ok());
+ assert!(backend.validate_remote_version_id(&Uuid::nil().to_string()).is_err());
+ assert!(backend.validate_remote_version_id("opaque-version").is_err());
+ assert!(
+ backend
+ .validate_remote_version_id("550e8400-e29b-41d4-a716-446655440000")
+ .is_err()
+ );
+ }
+
+ #[tokio::test]
+ async fn rejected_put_version_latches_nonexact_operations() {
+ let Some((backend, request_task)) = backend_with_responses(vec![TestResponse {
+ status: "204 No Content",
+ headers: "",
+ body: "",
+ }])
+ .await
+ else {
+ return;
+ };
+
+ backend
+ .validate_remote_version_id("unexpected-version")
+ .expect_err("a real version ID must detect versioning drift");
+ backend
+ .put("new-object", ReaderImpl::Body(hyper::body::Bytes::new()), 0)
+ .await
+ .expect_err("drift must stop subsequent uploads");
+ backend
+ .get("remote-object", "", WarmBackendGetOpts::default())
+ .await
+ .expect_err("drift must stop versionless reads");
+ backend
+ .remove("remote-object", "")
+ .await
+ .expect_err("drift must stop versionless deletes");
+ backend.in_use().await.expect_err("drift must make usage checks fail closed");
+ backend
+ .remove("remote-object", "unexpected-version")
+ .await
+ .expect("exact cleanup must remain available after rejecting a PUT version");
+ let requests = request_task.await.expect("exact cleanup request should finish");
+
+ assert_eq!(requests.len(), 1);
+ assert_eq!(
+ request_target(&requests[0]),
+ "/tier-bucket/archive/remote-object?versionId=unexpected-version"
+ );
+ }
+
+ #[tokio::test]
+ async fn a_fresh_backend_rechecks_versioning_before_an_empty_delete() {
+ let Some((backend, request_task)) = backend_with_responses(vec![TestResponse {
+ status: "200 OK",
+ headers: "",
+ body: r#"Enabled"#,
+ }])
+ .await
+ else {
+ return;
+ };
+ backend.versioning_checked.store(false, Ordering::Release);
+
+ let err = backend
+ .remove("remote-object", "")
+ .await
+ .expect_err("a restarted backend must reject versionless deletion after versioning drift");
+ assert!(err.to_string().contains("must be unversioned"), "{err}");
+ let requests = request_task.await.expect("versioning request task should finish");
+
+ assert_eq!(requests.len(), 1);
+ assert!(requests[0].head.starts_with("GET "));
+ assert!(request_target(&requests[0]).contains("versioning"));
+ }
+
+ #[tokio::test]
+ async fn a_fresh_backend_rechecks_versioning_before_put_with_meta() {
+ let Some((backend, request_task)) = backend_with_responses(vec![TestResponse {
+ status: "200 OK",
+ headers: "",
+ body: r#"Enabled"#,
+ }])
+ .await
+ else {
+ return;
+ };
+ backend.versioning_checked.store(false, Ordering::Release);
+
+ let err = backend
+ .put_with_meta("remote-object", ReaderImpl::Body(hyper::body::Bytes::new()), 0, HashMap::new())
+ .await
+ .expect_err("lifecycle PUT must reject a remotely versioned Wasabi bucket before upload");
+ assert!(err.to_string().contains("must be unversioned"), "{err}");
+ let requests = request_task.await.expect("versioning request task should finish");
+
+ assert_eq!(requests.len(), 1);
+ assert!(requests[0].head.starts_with("GET "));
+ assert!(request_target(&requests[0]).contains("versioning"));
+ }
+
+ #[tokio::test]
+ async fn delete_marker_rollback_failure_latches_versioning_drift() {
+ let Some((backend, request_task)) = backend_with_responses(vec![
+ TestResponse {
+ status: "204 No Content",
+ headers: "x-amz-delete-marker: true\r\nx-amz-version-id: marker-version\r\n",
+ body: "",
+ },
+ TestResponse {
+ status: "500 Internal Server Error",
+ headers: "",
+ body: "InternalError",
+ },
+ ])
+ .await
+ else {
+ return;
+ };
+
+ let err = backend
+ .remove("remote-object", "")
+ .await
+ .expect_err("marker rollback failure must retain the cleanup record");
+ backend
+ .remove("remote-object", "")
+ .await
+ .expect_err("the drift latch must stop another versionless delete");
+ let requests = request_task.await.expect("delete requests should finish");
+
+ assert!(err.to_string().contains("failed to remove the unexpected delete marker"), "{err}");
+ assert_eq!(requests.len(), 2);
+ assert_eq!(request_target(&requests[0]), "/tier-bucket/archive/remote-object");
+ assert_eq!(
+ request_target(&requests[1]),
+ "/tier-bucket/archive/remote-object?versionId=marker-version"
+ );
+ }
+
+ #[tokio::test]
+ async fn put_paths_send_complete_bodies_metadata_and_preserve_version_header() {
+ let Some((backend, request_task)) = backend_with_responses(vec![
+ TestResponse {
+ status: "200 OK",
+ headers: "ETag: \"etag-1\"\r\nx-amz-version-id: opaque.wasabi_01\r\n",
+ body: "",
+ },
+ TestResponse {
+ status: "200 OK",
+ headers: "ETag: \"etag-2\"\r\n",
+ body: "",
+ },
+ ])
+ .await
+ else {
+ return;
+ };
+ let with_meta = vec![0x5a; 4096];
+ let plain = vec![0x3c; 3072];
+ let mut metadata = HashMap::new();
+ metadata.insert("source-etag".to_string(), "etag-value".to_string());
+
+ let version = backend
+ .put_with_meta(
+ "with-meta",
+ ReaderImpl::Body(hyper::body::Bytes::from(with_meta.clone())),
+ i64::try_from(with_meta.len()).expect("test body length should fit i64"),
+ metadata,
+ )
+ .await
+ .expect("Wasabi PUT with metadata should succeed");
+ let unversioned = backend
+ .put(
+ "plain",
+ ReaderImpl::Body(hyper::body::Bytes::from(plain.clone())),
+ i64::try_from(plain.len()).expect("test body length should fit i64"),
+ )
+ .await
+ .expect("unversioned Wasabi PUT should succeed");
+ let requests = request_task.await.expect("PUT request task should finish");
+
+ assert_eq!(version, "opaque.wasabi_01");
+ assert!(unversioned.is_empty());
+ assert_eq!(requests.len(), 2);
+ assert!(requests[0].head.starts_with("PUT "));
+ assert_eq!(request_target(&requests[0]), "/tier-bucket/archive/with-meta");
+ assert_eq!(requests[0].body, with_meta);
+ assert_eq!(request_header(&requests[0], "content-length"), Some("4096"));
+ assert_eq!(request_header(&requests[0], "x-amz-meta-source-etag"), Some("etag-value"));
+ assert!(requests[1].head.starts_with("PUT "));
+ assert_eq!(request_target(&requests[1]), "/tier-bucket/archive/plain");
+ assert_eq!(requests[1].body, plain);
+ assert_eq!(request_header(&requests[1], "content-length"), Some("3072"));
+ assert_eq!(request_header(&requests[1], "x-amz-meta-source-etag"), None);
+ }
+}
diff --git a/rustfs/src/admin/handlers/tier.rs b/rustfs/src/admin/handlers/tier.rs
index 396d5bdf6..641fee008 100644
--- a/rustfs/src/admin/handlers/tier.rs
+++ b/rustfs/src/admin/handlers/tier.rs
@@ -79,6 +79,14 @@ pub struct AddTierQuery {
pub struct AddTier {}
+fn wasabi_payload_name(config: &TierConfig) -> S3Result {
+ config
+ .wasabi
+ .as_ref()
+ .map(|wasabi| wasabi.name.clone())
+ .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Wasabi configuration"))
+}
+
fn spawn_transition_tier_config_propagation(action: &'static str) {
if let Some(notification_sys) = current_notification_system() {
spawn_traced(async move {
@@ -271,6 +279,9 @@ impl Operation for AddTier {
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing S3 configuration"))?
.name;
}
+ TierType::Wasabi => {
+ args.name = wasabi_payload_name(&args)?;
+ }
TierType::RustFS => {
args.name = args
.rustfs
@@ -944,6 +955,20 @@ mod tests {
use http::Uri;
use matchit::Router;
+ #[test]
+ fn wasabi_payload_name_requires_nested_configuration() {
+ let config: TierConfig = serde_json::from_slice(
+ br#"{"type":"wasabi","wasabi":{"name":"WASABI-FIRST","accessKey":"ak","secretKey":"sk","bucket":"archive","region":"us-east-1"}}"#,
+ )
+ .expect("Wasabi AddTier payload should decode");
+ assert_eq!(wasabi_payload_name(&config).expect("Wasabi payload name should exist"), "WASABI-FIRST");
+
+ let missing: TierConfig = serde_json::from_slice(br#"{"type":"wasabi"}"#).expect("type-only payload should decode");
+ let err = wasabi_payload_name(&missing).expect_err("missing Wasabi payload must fail at the AddTier boundary");
+ assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+ assert_eq!(err.message(), Some("missing Wasabi configuration"));
+ }
+
#[test]
fn resolve_tier_name_prefers_path_parameter() {
let uri: Uri = "/rustfs/admin/v3/tier/HOT?tier=COLD".parse().expect("uri should parse");
diff --git a/scripts/check_logging_guardrails.sh b/scripts/check_logging_guardrails.sh
index cb3640e5e..7663ce82c 100755
--- a/scripts/check_logging_guardrails.sh
+++ b/scripts/check_logging_guardrails.sh
@@ -43,6 +43,7 @@ checked_files=(
"crates/targets/src/target/webhook.rs"
"crates/ecstore/src/store/peer.rs"
"crates/ecstore/src/store/init.rs"
+ "crates/ecstore/src/client/transition_api.rs"
"crates/ecstore/src/services/tier/tier.rs"
"crates/heal/src/heal/manager.rs"
"crates/heal/src/heal/storage.rs"
@@ -124,6 +125,8 @@ forbidden_patterns=(
'secret_key={}'
'Authorization={}'
'token={}'
+ 'debug!("http_client headers: {:?}"'
+ 'warn!("err_body: {}"'
'debug!("config: {:?}"'
'warn!("No audit targets configured for dispatch"'
'warn!("No audit targets configured for batch dispatch"'