feat(tier): support native Azure Blob endpoints

Cargo.lock

@@ -9862,6 +9862,7 @@ dependencies = [
  "path-absolutize",
  "pin-project-lite",
  "proptest",
+ "quick-xml",
  "rand 0.10.3",
  "ratelimit",
  "rcgen",
@@ -9894,7 +9895,9 @@ dependencies = [
  "rustfs-s3-ops",
  "rustfs-s3-types",
  "rustfs-scanner-metrics",
+ "rustfs-signer",
  "rustfs-storage-api",
+ "rustfs-tls-runtime",
  "rustfs-uring",
  "rustfs-utils",
  "rustix",
@@ -10896,6 +10899,7 @@ dependencies = [
  "time",
  "tracing",
  "tracing-subscriber",
+ "url",
 ]
 
 [[package]]

crates/ecstore/Cargo.toml

@@ -31,7 +31,7 @@ workspace = true
 
 [features]
 default = []
-fips = ["rustfs-s3-client/fips", "rustfs-utils/fips"]
+fips = ["rustfs-s3-client/fips", "rustfs-utils/fips", "rustfs-tls-runtime/fips"]
 diagnostic-read-version-caller = ["rustfs-io-metrics/diagnostic-read-version-caller"]
 # Sampled wall/CPU diagnostics only; does not change the read backend.
 diagnostic-local-batch = ["rustix/time"]
@@ -203,6 +203,9 @@ libc.workspace = true
 # (backlog#1178); "thread" provides cleanup affinity; "fs" is a workspace default.
 rustix = { workspace = true, features = ["process", "fs", "thread"] }
 rustfs-madmin.workspace = true
+rustfs-signer.workspace = true
+rustfs-tls-runtime.workspace = true
+quick-xml.workspace = true
 reqwest = { workspace = true }
 aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
 urlencoding = { workspace = true }

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

@@ -26,6 +26,7 @@ pub(crate) mod tier_probe_intent;
 pub mod warm_backend;
 pub mod warm_backend_aliyun;
 pub mod warm_backend_azure;
+pub mod warm_backend_azure_blob;
 #[cfg(feature = "gcs")]
 pub mod warm_backend_gcs;
 pub mod warm_backend_huaweicloud;

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

@@ -327,6 +327,7 @@ pub const TIER_CONFIG_VERSION: u16 = 2;
 const EXTERNAL_TIER_TYPE_UNSUPPORTED: i32 = 0;
 const EXTERNAL_TIER_TYPE_S3: i32 = 1;
 const EXTERNAL_TIER_TYPE_AZURE: i32 = 2;
+const EXTERNAL_TIER_AZURE_BLOB_ENDPOINT_SENTINEL: &str = "rustfs-azure-blob-v1:";
 const EXTERNAL_TIER_TYPE_GCS: i32 = 3;
 const EXTERNAL_TIER_TYPE_MINIO: i32 = 4;
 const EXTERNAL_TIER_VERSION_WASABI_V1: &str = "rustfs-wasabi-v1";
@@ -2147,6 +2148,7 @@ fn build_coordinator_tier_mutation_intent(
         return Ok(None);
     }
     Ok(Some(TierMutationIntent {
+        requires_native_azure: candidate.tiers.values().any(TierConfig::is_native_azure),
         mutation_id: uuid::Uuid::new_v4(),
         revision: 1,
         kind,
@@ -2515,11 +2517,11 @@ fn tier_backend_identity(config: &TierConfig) -> io::Result<TierDestinationId> {
         }),
         TierType::Azure => config.azure.as_ref().map(|value| {
             (
-                "azure",
+                if config.is_native_azure() { "azure-blob-v1" } else { "azure" },
                 value.endpoint.as_str(),
                 value.bucket.as_str(),
                 value.prefix.as_str(),
-                value.region.as_str(),
+                if config.is_native_azure() { "" } else { value.region.as_str() },
                 value.access_key.as_str(),
             )
         }),
@@ -2592,7 +2594,7 @@ fn normalized_tier_prefix<'a>(tier_type: &TierType, prefix: &'a str) -> &'a str
 }
 
 fn tier_exact_get_delete(config: &TierConfig) -> bool {
-    ProviderVersionCapabilities::for_tier_type(&config.tier_type.as_lowercase()).exact_get_delete
+    config.is_native_azure() || ProviderVersionCapabilities::for_tier_type(&config.tier_type.as_lowercase()).exact_get_delete
 }
 
 struct TierDriverGeneration {
@@ -3130,7 +3132,15 @@ fn to_external_tier_config(name: &str, tier: &TierConfig) -> io::Result<External
                 .ok_or_else(|| io::Error::other("tier config missing azure backend payload"))?;
             out.tier_type = EXTERNAL_TIER_TYPE_AZURE;
             out.azure = Some(ExternalTierAzure {
-                endpoint: az.endpoint.clone(),
+                // Keep the v2 Azure payload readable and rewriteable by older
+                // servers, but make their S3 constructor reject its endpoint.
+                // An unknown numeric type would discard the Azure payload on
+                // an unrelated configuration rewrite by an older server.
+                endpoint: if tier.is_native_azure() {
+                    format!("{EXTERNAL_TIER_AZURE_BLOB_ENDPOINT_SENTINEL}{}", az.endpoint)
+                } else {
+                    az.endpoint.clone()
+                },
                 account_name: az.access_key.clone(),
                 account_key: az.secret_key.clone(),
                 bucket: az.bucket.clone(),
@@ -3289,6 +3299,15 @@ fn from_external_tier_config(name: String, ext: ExternalTierConfig) -> io::Resul
         ..Default::default()
     };
 
+    let azure_blob = ext.version == super::tier_config::AZURE_BLOB_TIER_VERSION;
+    let native_endpoint = ext
+        .azure
+        .as_ref()
+        .is_some_and(|azure| azure.endpoint.starts_with(EXTERNAL_TIER_AZURE_BLOB_ENDPOINT_SENTINEL));
+    if azure_blob != native_endpoint || azure_blob && (ext.tier_type != EXTERNAL_TIER_TYPE_AZURE || ext.tier_type_hint.is_some())
+    {
+        return Err(io::Error::other("native Azure Blob tier has inconsistent protocol discriminators"));
+    }
     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 {
@@ -3309,7 +3328,9 @@ fn from_external_tier_config(name: String, ext: ExternalTierConfig) -> io::Resul
             cfg.name
         )));
     }
-    let tier_type = if wasabi_version {
+    let tier_type = if azure_blob {
+        TierType::Azure
+    } else if wasabi_version {
         TierType::Wasabi
     } else {
         tier_type_from_hint(ext.tier_type_hint.as_deref()).unwrap_or(match ext.tier_type {
@@ -3387,7 +3408,15 @@ fn from_external_tier_config(name: String, ext: ExternalTierConfig) -> io::Resul
                 .ok_or_else(|| io::Error::other(format!("tier config '{}' missing azure backend payload", cfg.name)))?;
             cfg.azure = Some(crate::services::tier::tier_config::TierAzure {
                 name: cfg.name.clone(),
-                endpoint: az.endpoint.clone(),
+                endpoint: if azure_blob {
+                    az.endpoint
+                        .strip_prefix(EXTERNAL_TIER_AZURE_BLOB_ENDPOINT_SENTINEL)
+                        .filter(|endpoint| !endpoint.is_empty())
+                        .ok_or_else(|| io::Error::other("native Azure Blob endpoint marker is missing"))?
+                        .to_string()
+                } else {
+                    az.endpoint.clone()
+                },
                 access_key: az.account_name.clone(),
                 secret_key: az.account_key.clone(),
                 bucket: az.bucket.clone(),
@@ -3558,6 +3587,9 @@ impl TierConfigMgr {
     }
 
     pub fn unmarshal(data: &[u8]) -> std::result::Result<TierConfigMgr, std::io::Error> {
+        if data.starts_with(&TIER_CONFIG_FORMAT.to_le_bytes()) {
+            return decode_external_tiering_config_blob(data);
+        }
         let mut cfg: TierConfigMgr = serde_json::from_slice(data)?;
         for (name, tier) in &mut cfg.tiers {
             if matches!(&tier.tier_type, TierType::Wasabi) {
@@ -3575,6 +3607,11 @@ impl TierConfigMgr {
     }
 
     pub fn marshal(&self) -> std::result::Result<Bytes, std::io::Error> {
+        // The legacy JSON shape omits the driver protocol version. Native
+        // Azure must retain the binary discriminator even through this helper.
+        if self.tiers.values().any(TierConfig::is_native_azure) {
+            return encode_external_tiering_config_blob(self);
+        }
         let data = serde_json::to_vec(&self)?;
         let mut data = Bytes::from(data);
 
@@ -3608,6 +3645,17 @@ impl TierConfigMgr {
         deadline: Option<Instant>,
     ) -> std::result::Result<(), AdminError> {
         normalize_s3_gcs_add_tier_name(&mut tier_config)?;
+        if tier_config.tier_type == TierType::Azure {
+            tier_config.version = super::tier_config::AZURE_BLOB_TIER_VERSION.to_string();
+            if let Some(azure) = tier_config.azure.as_mut() {
+                if azure.storage_class.eq_ignore_ascii_case("STANDARD") {
+                    azure.storage_class.clear();
+                }
+                if azure.endpoint.is_empty() {
+                    azure.endpoint = format!("https://{}.blob.core.windows.net", azure.access_key);
+                }
+            }
+        }
         self.ensure_generation_is_idle(&tier_config.name)?;
         let tier_name = tier_config.name.clone();
         if tier_name != tier_name.to_uppercase() {
@@ -9536,6 +9584,98 @@ mod tests {
         assert_eq!(az.sp_auth.client_secret, "secret");
     }
 
+    #[tokio::test]
+    async fn native_azure_add_preserves_protocol_through_restart_and_requires_peer_support() {
+        let mut manager = empty_mgr();
+        let mut tier = build_azure_tier("account");
+        let name = tier.name.clone();
+        let azure = tier.azure.as_mut().expect("Azure payload should exist");
+        azure.endpoint.clear();
+        azure.storage_class = "STANDARD".to_string();
+        azure.secret_key = base64_simd::STANDARD.encode_to_string([7; 32]);
+        TIER_DRIVER_TEST_FACTORY
+            .scope(healthy_driver_factory(), manager.add(tier, true))
+            .await
+            .expect("current Console Azure payload should be admitted");
+        let created = &manager.tiers[&name];
+        assert!(created.is_native_azure());
+        assert!(tier_exact_get_delete(created));
+        let azure = created.azure.as_ref().expect("Azure payload should survive");
+        assert_eq!(azure.endpoint, "https://account.blob.core.windows.net");
+        assert!(azure.storage_class.is_empty());
+
+        let bytes = manager
+            .marshal()
+            .expect("native configuration should serialize without losing its protocol");
+        let external: ExternalTierConfigMgr =
+            rmp_serde::from_slice(&bytes[4..]).expect("native config should use the binary shape");
+        assert_eq!(external.tiers[&name].tier_type, EXTERNAL_TIER_TYPE_AZURE);
+        let stored = external.tiers[&name]
+            .azure
+            .as_ref()
+            .expect("native storage keeps the existing Azure payload");
+        assert_eq!(stored.endpoint, "rustfs-azure-blob-v1:https://account.blob.core.windows.net");
+        let legacy_backend_config = crate::services::tier::tier_config::TierAzure {
+            endpoint: stored.endpoint.clone(),
+            access_key: stored.account_name.clone(),
+            secret_key: stored.account_key.clone(),
+            bucket: stored.bucket.clone(),
+            ..Default::default()
+        };
+        assert!(
+            crate::services::tier::warm_backend_azure::WarmBackendAzure::new(&legacy_backend_config, &name)
+                .await
+                .is_err(),
+            "the unchanged legacy S3 constructor must reject the native endpoint before any request"
+        );
+        assert!(
+            external.tiers[&name].tier_type_hint.is_none(),
+            "the protocol marker must not be overridden by an extended provider hint"
+        );
+        let restarted = TierConfigMgr::unmarshal(&bytes).expect("native configuration should reload");
+        assert!(restarted.tiers[&name].is_native_azure());
+        assert_eq!(
+            tier_backend_identity(created).expect("created identity"),
+            tier_backend_identity(&restarted.tiers[&name]).expect("reloaded identity")
+        );
+        let targets = vec![TierMutationIntentTarget {
+            tier_name: name,
+            old_backend_identity: None,
+            new_backend_identity: Some(tier_backend_identity(created).expect("native destination identity")),
+        }];
+        let intent = build_coordinator_tier_mutation_intent(TierMutationIntentKind::Add, None, &manager, targets)
+            .expect("native intent should build")
+            .expect("native Add must use peer prepare");
+        assert!(intent.requires_native_azure);
+    }
+
+    #[test]
+    fn native_azure_protocol_is_distinct_from_legacy_and_rejects_corrupt_discriminators() {
+        let legacy = build_azure_tier("account");
+        let mut native = legacy.clone_with_credentials();
+        native.version = super::super::tier_config::AZURE_BLOB_TIER_VERSION.to_string();
+        assert_ne!(
+            tier_backend_identity(&legacy).expect("legacy identity"),
+            tier_backend_identity(&native).expect("native identity"),
+            "native Blob must never acquire a legacy S3 destination's cleanup authority"
+        );
+        let stored = to_external_tier_config(&legacy.name, &legacy).expect("legacy encoding");
+        assert_eq!(stored.tier_type, EXTERNAL_TIER_TYPE_AZURE);
+        let decoded = from_external_tier_config(legacy.name, stored).expect("legacy decoding");
+        assert!(!decoded.is_native_azure());
+        assert!(!tier_exact_get_delete(&decoded));
+        let stored = to_external_tier_config(&native.name, &native).expect("native encoding");
+        for mutate in [
+            |config: &mut ExternalTierConfig| config.version = "v1".to_string(),
+            |config: &mut ExternalTierConfig| config.tier_type = EXTERNAL_TIER_TYPE_S3,
+            |config: &mut ExternalTierConfig| config.tier_type_hint = Some("azure".to_string()),
+        ] {
+            let mut corrupt = stored.clone();
+            mutate(&mut corrupt);
+            assert!(from_external_tier_config(native.name.clone(), corrupt).is_err());
+        }
+    }
+
     #[test]
     fn test_external_blob_gcs_writer_stays_raw_for_v2_old_readers() {
         const RAW_JSON: &str = r#"{"type":"service_account","project_id":"tier-௿"}"#;
@@ -9944,6 +10084,7 @@ mod tests {
 
     fn prepared_remove_intent(tier_name: &str, mutation_id: uuid::Uuid) -> TierMutationIntent {
         TierMutationIntent {
+            requires_native_azure: false,
             mutation_id,
             revision: 1,
             kind: TierMutationIntentKind::Remove,

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

@@ -139,6 +139,7 @@ impl TierType {
 }
 
 pub(crate) const TIER_CREDENTIAL_REDACTED: &str = "REDACTED";
+pub(crate) const AZURE_BLOB_TIER_VERSION: &str = "rustfs-azure-blob-v1";
 
 #[derive(Default, Serialize, Deserialize)]
 #[serde(default)]
@@ -178,6 +179,10 @@ impl Clone for TierConfig {
 }
 
 impl TierConfig {
+    pub(crate) fn is_native_azure(&self) -> bool {
+        self.tier_type == TierType::Azure && self.version == AZURE_BLOB_TIER_VERSION
+    }
+
     pub(crate) fn redacted(&self) -> Self {
         let mut redacted = Self {
             version: self.version.clone(),

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

@@ -127,6 +127,10 @@ impl TierMutationIntentTarget {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub(crate) struct TierMutationIntent {
+    // Old peers deny unknown intent fields before installing a prepare block.
+    // Omit this extension for existing drivers to retain their wire checksum.
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub requires_native_azure: bool,
     pub mutation_id: Uuid,
     pub revision: u64,
     pub kind: TierMutationIntentKind,
@@ -282,7 +286,8 @@ impl TierMutationIntent {
     }
 
     pub(crate) fn same_identity_as(&self, other: &Self) -> bool {
-        self.mutation_id == other.mutation_id
+        self.requires_native_azure == other.requires_native_azure
+            && self.mutation_id == other.mutation_id
             && self.kind == other.kind
             && self.old_config_etag == other.old_config_etag
             && self.candidate_digest == other.candidate_digest
@@ -825,6 +830,10 @@ fn identity_is_empty(identity: &TierDestinationId) -> bool {
     identity.iter().all(|byte| *byte == 0)
 }
 
+fn is_false(value: &bool) -> bool {
+    !value
+}
+
 fn digest_is_empty(digest: &TierMutationDigest) -> bool {
     digest.iter().all(|byte| *byte == 0)
 }
@@ -840,6 +849,7 @@ mod tests {
 
     fn prepared_intent() -> TierMutationIntent {
         TierMutationIntent {
+            requires_native_azure: false,
             mutation_id: Uuid::new_v4(),
             revision: 1,
             kind: TierMutationIntentKind::Edit,
@@ -867,6 +877,49 @@ mod tests {
         assert_eq!(shard, tier_mutation_mutex_shard_index(mutation_id));
     }
 
+    #[test]
+    fn native_azure_requirement_is_identity_bound_and_rejected_by_the_previous_wire_reader() {
+        // Preserve the previous strict wire shape as an independent reader.
+        #[derive(Serialize, Deserialize)]
+        #[serde(deny_unknown_fields)]
+        struct LegacyIntent {
+            mutation_id: Uuid,
+            revision: u64,
+            kind: TierMutationIntentKind,
+            state: TierMutationIntentState,
+            #[serde(default, skip_serializing_if = "Option::is_none")]
+            old_config_etag: Option<String>,
+            #[serde(default, skip_serializing_if = "Option::is_none")]
+            committed_config_etag: Option<String>,
+            candidate_digest: TierMutationDigest,
+            affected_targets: Vec<TierMutationIntentTarget>,
+            expires_at_unix_nanos: i64,
+        }
+        let legacy = prepared_intent();
+        let legacy_bytes = serde_json::to_vec(&legacy).expect("legacy intent should encode");
+        let old_reader: LegacyIntent =
+            serde_json::from_slice(&legacy_bytes).expect("ordinary mutations remain readable by old peers");
+        assert_eq!(
+            serde_json::to_vec(&old_reader).expect("old encoding"),
+            legacy_bytes,
+            "ordinary mutation checksums must stay unchanged"
+        );
+
+        let mut native = legacy.clone();
+        native.requires_native_azure = true;
+        let native_bytes = serde_json::to_vec(&native).expect("native intent should encode");
+        assert!(
+            serde_json::from_slice::<LegacyIntent>(&native_bytes).is_err(),
+            "an old peer must reject native Azure before prepare"
+        );
+        assert!(!legacy.same_identity_as(&native), "an abort cannot drop the native-driver requirement");
+        let encoded = native.encode().expect("native durable intent should encode");
+        assert_eq!(
+            TierMutationIntent::decode(native.mutation_id, &encoded).expect("new peer should decode"),
+            native
+        );
+    }
+
     #[tokio::test]
     async fn mutation_mutex_serializes_the_same_id() {
         let mutation_id = Uuid::new_v4();

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

@@ -25,6 +25,7 @@ use crate::services::tier::{
     tier_handlers::{ERR_TIER_BUCKET_NOT_FOUND, ERR_TIER_INVALID_CREDENTIALS, ERR_TIER_NOT_FOUND, ERR_TIER_PERM_ERR},
     warm_backend_aliyun::WarmBackendAliyun,
     warm_backend_azure::WarmBackendAzure,
+    warm_backend_azure_blob::{AzureBlobError, WarmBackendAzureBlob},
     warm_backend_huaweicloud::WarmBackendHuaweicloud,
     warm_backend_minio::WarmBackendMinIO,
     warm_backend_r2::WarmBackendR2,
@@ -309,10 +310,10 @@ pub fn build_transition_put_options(storage_class: String, mut metadata: HashMap
 
 /// Connection parameters every S3-compatible warm backend provider supplies.
 ///
-/// The Aliyun, Azure, Huaweicloud, Tencent, MinIO, R2, and RustFS backends all
+/// The Aliyun, legacy Azure, Huaweicloud, Tencent, MinIO, R2, and RustFS backends all
 /// wrap [`WarmBackendS3`] around a statically-credentialed [`TransitionClient`]
 /// built from exactly these values. `bucket_lookup` is a parameter rather than a
-/// constant because the providers split into two families: Aliyun, Azure,
+/// constant because the providers split into two families: Aliyun, legacy Azure,
 /// Huaweicloud, and Tencent pin [`BucketLookupType::BucketLookupDNS`], while
 /// MinIO, R2, and RustFS leave it at [`BucketLookupType::BucketLookupAuto`].
 pub(crate) struct S3CompatibleWarmBackendParams<'a> {
@@ -673,6 +674,9 @@ fn probe_cleanup_incomplete_error() -> AdminError {
 }
 
 fn is_remote_tier_auth_rejection(err: &std::io::Error) -> bool {
+    if AzureBlobError::from_io(err).is_some() && err.kind() == std::io::ErrorKind::PermissionDenied {
+        return true;
+    }
     let Some(response) = err.get_ref().and_then(|err| err.downcast_ref::<ErrorResponse>()) else {
         return false;
     };
@@ -706,10 +710,27 @@ async fn check_warm_backend_with_deadlines(
         Ok(Err(err)) if is_remote_tier_auth_rejection(&err) => {
             return Err(ERR_TIER_INVALID_CREDENTIALS.clone());
         }
-        Ok(Err(_)) => {
+        Ok(Err(err)) => {
+            if let Some(rejection) = AzureBlobError::from_io(&err).filter(|error| error.rejects_write()) {
+                let mut admin_err = ERR_TIER_INVALID_CONFIG.clone();
+                admin_err.message = rejection.to_string();
+                return Err(admin_err);
+            }
             return Err(match compensate_uncertain_probe_put(w, &probe_object, cleanup_deadline).await {
-                Ok(()) => ERR_TIER_PERM_ERR.clone(),
-                Err(_) => probe_cleanup_incomplete_error(),
+                Ok(()) => {
+                    let mut admin_err = ERR_TIER_PERM_ERR.clone();
+                    if let Some(original) = AzureBlobError::from_io(&err) {
+                        admin_err.message = original.to_string();
+                    }
+                    admin_err
+                }
+                Err(_) => {
+                    let mut admin_err = probe_cleanup_incomplete_error();
+                    if let Some(original) = AzureBlobError::from_io(&err) {
+                        admin_err.message.push_str(&format!(": {original}"));
+                    }
+                    admin_err
+                }
             });
         }
         Err(_) => {
@@ -951,24 +972,23 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
             }
         }
         TierType::Azure => {
-            if let Some(azure_config) = tier.azure.as_ref() {
-                let dd = WarmBackendAzure::new(azure_config, &tier.name).await;
-                if let Err(err) = dd {
-                    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,
-                    });
-                }
-                d = Some(Box::new(dd.expect("Failed to create Azure backend")));
+            let azure = tier.azure.as_ref().ok_or_else(|| AdminError {
+                code: "XRustFSAdminTierInvalidConfig".to_string(),
+                message: "Azure tier configuration not found".to_string(),
+                status_code: StatusCode::BAD_REQUEST,
+            })?;
+            let driver = if tier.is_native_azure() {
+                WarmBackendAzureBlob::new(azure).map(|driver| -> WarmBackendImpl { Box::new(driver) })
             } else {
-                return Err(AdminError {
-                    code: "XRustFSAdminTierInvalidConfig".to_string(),
-                    message: "Azure tier configuration not found".to_string(),
-                    status_code: StatusCode::BAD_REQUEST,
-                });
-            }
+                WarmBackendAzure::new(azure, &tier.name)
+                    .await
+                    .map(|driver| -> WarmBackendImpl { Box::new(driver) })
+            };
+            d = Some(driver.map_err(|err| AdminError {
+                code: "XRustFSAdminTierInvalidConfig".to_string(),
+                message: format!("Unable to setup remote tier, check tier configuration: {err}"),
+                status_code: StatusCode::BAD_REQUEST,
+            })?);
         }
         #[cfg(not(feature = "gcs"))]
         TierType::GCS => {
@@ -1042,6 +1062,13 @@ pub(crate) async fn new_transition_candidate_reconciler(
     tier: &TierConfig,
 ) -> Result<Option<Box<dyn TransitionCandidateReconciler + Send + Sync + 'static>>, AdminError> {
     let reconciler: Box<dyn TransitionCandidateReconciler + Send + Sync + 'static> = match tier.tier_type {
+        TierType::Azure if tier.is_native_azure() => Box::new(
+            WarmBackendAzureBlob::new(tier.azure.as_ref().ok_or_else(|| ERR_TIER_INVALID_CONFIG.clone())?).map_err(|err| {
+                let mut admin_err = ERR_TIER_INVALID_CONFIG.clone();
+                admin_err.message = err.to_string();
+                admin_err
+            })?,
+        ),
         TierType::S3 => Box::new(
             WarmBackendS3::new(tier.s3.as_ref().ok_or_else(|| ERR_TIER_INVALID_CONFIG.clone())?, &tier.name)
                 .await

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

@@ -88,7 +88,7 @@ impl WarmBackend for WarmBackendAzure {
     }
 
     async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-        // Azure currently uses the shared S3/SigV4 transport, but its normal
+        // Legacy Azure uses the shared S3/SigV4 transport, but its normal
         // object path cannot persist exact remote versions across mixed
         // RustFS releases. The mutation probe may still detect and precisely
         // remove a versioned test object before rejecting that configuration.

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

@@ -0,0 +1,828 @@
+// 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.
+
+//! Native Azure Blob tier transport, including Azure Stack Hub 2301 and later.
+//!
+//! Writes use one conditional commit and never replay a failed HTTP request.
+//! Larger bodies are staged in bounded blocks with a unique upload namespace;
+//! an interrupted staging operation leaves only Azure's expiring uncommitted
+//! blocks. Lifecycle uploads use the existing transaction journal after a lost
+//! commit acknowledgement; validation probes retain in-process compensation.
+//! GET and DELETE carry an exact version only when the persisted version is
+//! non-empty. Unversioned DELETE uses an ETag condition to refuse replacement.
+
+use super::tier_config::TierAzure;
+use super::warm_backend::{
+    TransitionCandidateIdentity, TransitionCandidateProbe, TransitionCandidateReconciler, WarmBackend, WarmBackendGetOpts,
+    build_transition_put_options, transition_client_timeouts_from_env,
+};
+use super::warm_backend_s3::transition_candidate_metadata_matches;
+use bytes::Bytes;
+use futures_util::stream;
+use http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
+use quick_xml::{Reader, events::Event};
+use rustfs_s3_client::transition_api::{ReadCloser, ReaderImpl};
+use rustfs_signer::azure_shared_key_signature;
+use rustfs_utils::crypto::hex;
+use rustfs_utils::egress::validate_outbound_url;
+use rustfs_utils::hash::FastMd5;
+use rustfs_utils::http::metadata_compat::{
+    MINIO_INTERNAL_PREFIX, SUFFIX_TRANSITION_TIER_DESTINATION_ID, SUFFIX_TRANSITION_TRANSACTION_ID, internal_key_rustfs,
+};
+use std::collections::HashMap;
+use std::io::{self, Cursor, ErrorKind};
+use std::time::Duration;
+use time::OffsetDateTime;
+use tokio::io::{AsyncRead, AsyncReadExt};
+use url::Url;
+
+const API_VERSION: &str = "2020-10-02";
+const MIN_BLOCK_BYTES: u64 = 16 * 1024 * 1024;
+const MAX_BLOCK_BYTES: u64 = 100 * 1024 * 1024;
+const MAX_BLOCKS: u64 = 50_000;
+const MAX_CONTROL_BYTES: usize = 64 * 1024;
+const MAX_METADATA_BYTES: usize = 8 * 1024;
+const METADATA_PREFIX: &str = "x-ms-meta-rustfs_";
+const HTTP_DATE: &[time::format_description::BorrowedFormatItem<'static>] =
+    time::macros::format_description!("[weekday repr:short], [day] [month repr:short] [year] [hour]:[minute]:[second] GMT");
+
+/// A bounded diagnostic that never contains credentials, URLs or response bodies.
+#[derive(Debug, thiserror::Error)]
+#[error("Azure Blob {operation} failed (HTTP {status}, code {code}, request ID {request_id})")]
+pub(crate) struct AzureBlobError {
+    operation: &'static str,
+    status: StatusCode,
+    code: String,
+    request_id: String,
+}
+
+impl AzureBlobError {
+    pub(super) fn from_io(error: &io::Error) -> Option<&Self> {
+        error.get_ref()?.downcast_ref()
+    }
+
+    pub(super) fn rejects_write(&self) -> bool {
+        matches!(
+            self.status,
+            StatusCode::BAD_REQUEST | StatusCode::NOT_FOUND | StatusCode::CONFLICT | StatusCode::PRECONDITION_FAILED
+        )
+    }
+}
+
+pub struct WarmBackendAzureBlob {
+    client: reqwest::Client,
+    endpoint: Url,
+    account: String,
+    key: Vec<u8>,
+    container: String,
+    prefix: String,
+    request_timeout: Duration,
+}
+
+impl WarmBackendAzureBlob {
+    /// Build a native Blob backend without performing remote I/O.
+    ///
+    /// # Errors
+    ///
+    /// Rejects unsupported credentials or storage classes, an invalid account,
+    /// container, prefix, endpoint or account key, and TLS/client setup failures.
+    pub fn new(conf: &TierAzure) -> io::Result<Self> {
+        validate_config(conf)?;
+        let endpoint = if conf.endpoint.is_empty() {
+            format!("https://{}.blob.core.windows.net", conf.access_key)
+        } else {
+            conf.endpoint.clone()
+        };
+        let endpoint = Url::parse(&endpoint).map_err(|_| invalid_input("Azure Blob endpoint is not a valid URL"))?;
+        if !matches!(endpoint.scheme(), "https" | "http")
+            || endpoint.host_str().is_none()
+            || !endpoint.username().is_empty()
+            || endpoint.password().is_some()
+            || !matches!(endpoint.path(), "" | "/")
+            || endpoint.query().is_some()
+            || endpoint.fragment().is_some()
+        {
+            return Err(invalid_input(
+                "Azure Blob endpoint must be an HTTP(S) origin without credentials, path or query",
+            ));
+        }
+        validate_outbound_url(&endpoint).map_err(|_| invalid_input("Azure Blob endpoint is not allowed"))?;
+        let key = decode_key(&conf.secret_key)?;
+        rustfs_tls_runtime::ensure_crypto_provider().map_err(io::Error::other)?;
+        let timeouts = transition_client_timeouts_from_env();
+        let client = reqwest::Client::builder()
+            .connect_timeout(timeouts.connect_timeout)
+            .read_timeout(timeouts.response_body_idle_timeout)
+            .redirect(reqwest::redirect::Policy::none())
+            .retry(reqwest::retry::never())
+            .no_gzip()
+            .no_brotli()
+            .no_zstd()
+            .no_deflate()
+            .build()
+            .map_err(transport_error)?;
+        Ok(Self {
+            client,
+            endpoint,
+            account: conf.access_key.clone(),
+            key,
+            container: conf.bucket.clone(),
+            prefix: conf.prefix.strip_suffix('/').unwrap_or(&conf.prefix).to_string(),
+            request_timeout: timeouts.request_timeout,
+        })
+    }
+
+    fn object_url(&self, object: &str, version: &str) -> io::Result<Url> {
+        validate_version(version)?;
+        let name = if self.prefix.is_empty() {
+            object.to_string()
+        } else {
+            format!("{}/{object}", self.prefix)
+        };
+        validate_object_name(&name)?;
+        let mut url = self.endpoint.clone();
+        {
+            let mut segments = url
+                .path_segments_mut()
+                .map_err(|_| invalid_input("Azure Blob endpoint has no path"))?;
+            segments.clear().push(&self.container);
+            for segment in name.split('/') {
+                segments.push(segment);
+            }
+        }
+        if !version.is_empty() {
+            url.query_pairs_mut().append_pair("versionid", version);
+        }
+        Ok(url)
+    }
+
+    fn container_url(&self) -> io::Result<Url> {
+        let mut url = self.endpoint.clone();
+        url.path_segments_mut()
+            .map_err(|_| invalid_input("Azure Blob endpoint has no path"))?
+            .clear()
+            .push(&self.container);
+        Ok(url)
+    }
+
+    async fn send(&self, method: Method, url: Url, mut headers: HeaderMap, body: Option<Bytes>) -> io::Result<reqwest::Response> {
+        if let Some(body) = &body {
+            insert_header(&mut headers, "content-length", &body.len().to_string())?;
+            let bytes = body.clone();
+            let digest = tokio::task::spawn_blocking(move || FastMd5::digest(&bytes))
+                .await
+                .map_err(|_| io::Error::other("Azure Blob content checksum task failed"))?;
+            insert_header(&mut headers, "content-md5", &base64_simd::STANDARD.encode_to_string(digest))?;
+        }
+        headers.insert("x-ms-version", HeaderValue::from_static(API_VERSION));
+        headers.insert("accept-encoding", HeaderValue::from_static("identity"));
+        let date = OffsetDateTime::now_utc()
+            .format(HTTP_DATE)
+            .map_err(|_| io::Error::other("Azure Blob request date is not representable"))?;
+        insert_header(&mut headers, "x-ms-date", &date)?;
+        let signature = azure_shared_key_signature(&self.key, &self.account, method.as_str(), &url, &headers)
+            .map_err(|err| io::Error::new(ErrorKind::InvalidInput, err))?;
+        insert_header(&mut headers, "authorization", &format!("SharedKey {}:{signature}", self.account))?;
+        let mut request = reqwest::Request::new(method, url);
+        *request.headers_mut() = headers;
+        *request.body_mut() = body.map(reqwest::Body::from);
+        tokio::time::timeout(self.request_timeout, self.client.execute(request))
+            .await
+            .map_err(|_| io::Error::new(ErrorKind::TimedOut, "Azure Blob request timed out"))?
+            .map_err(transport_error)
+    }
+
+    async fn expect_status(
+        &self,
+        response: reqwest::Response,
+        expected: StatusCode,
+        operation: &'static str,
+    ) -> io::Result<reqwest::Response> {
+        if response.status() == expected {
+            return Ok(response);
+        }
+        Err(service_error(response, operation, self.request_timeout).await)
+    }
+
+    async fn head(&self, object: &str, version: &str) -> io::Result<Option<HeaderMap>> {
+        let response = self
+            .send(Method::HEAD, self.object_url(object, version)?, HeaderMap::new(), None)
+            .await?;
+        if response.status() == StatusCode::OK {
+            if !version.is_empty() {
+                validate_response_version(response.headers(), version)?;
+            }
+            return Ok(Some(response.headers().clone()));
+        }
+        let err = service_error(response, "HEAD", self.request_timeout).await;
+        if err.kind() == ErrorKind::NotFound {
+            return Ok(None);
+        }
+        Err(err)
+    }
+
+    async fn write(
+        &self,
+        object: &str,
+        reader: ReaderImpl,
+        length: i64,
+        metadata: HashMap<String, String>,
+    ) -> io::Result<String> {
+        let size = u64::try_from(length).map_err(|_| invalid_input("Azure Blob upload requires a known, non-negative length"))?;
+        let block_size = block_size(size)?;
+        let url = self.object_url(object, "")?;
+        let mut headers = metadata_headers(metadata)?;
+        headers.insert("if-none-match", HeaderValue::from_static("*"));
+        let mut reader: Box<dyn AsyncRead + Send + Sync + Unpin> = match reader {
+            ReaderImpl::Body(bytes) => Box::new(Cursor::new(bytes)),
+            ReaderImpl::ObjectBody(reader) => Box::new(reader),
+        };
+        if size <= block_size {
+            let body = read_part(&mut reader, size).await?;
+            ensure_eof(&mut reader).await?;
+            headers.insert("x-ms-blob-type", HeaderValue::from_static("BlockBlob"));
+            let response = self.send(Method::PUT, url, headers, Some(body)).await?;
+            let response = self.expect_status(response, StatusCode::CREATED, "PUT").await?;
+            return response_version(response.headers());
+        }
+
+        let upload = uuid::Uuid::new_v4().simple().to_string();
+        let mut blocks = Vec::new();
+        let mut remaining = size;
+        while remaining > 0 {
+            let count = remaining.min(block_size);
+            let body = read_part(&mut reader, count).await?;
+            let block_id = base64_simd::STANDARD.encode_to_string(format!("{upload}-{:05}", blocks.len()));
+            let mut block_url = url.clone();
+            block_url
+                .query_pairs_mut()
+                .append_pair("comp", "block")
+                .append_pair("blockid", &block_id);
+            let response = self.send(Method::PUT, block_url, HeaderMap::new(), Some(body)).await?;
+            self.expect_status(response, StatusCode::CREATED, "PUT block").await?;
+            blocks.push(block_id);
+            remaining -= count;
+        }
+        ensure_eof(&mut reader).await?;
+        let mut xml = String::from("<?xml version=\"1.0\" encoding=\"utf-8\"?><BlockList>");
+        for block in blocks {
+            // IDs are generated base64, so they contain no XML delimiters.
+            xml.push_str("<Uncommitted>");
+            xml.push_str(&block);
+            xml.push_str("</Uncommitted>");
+        }
+        xml.push_str("</BlockList>");
+        let mut commit_url = url;
+        commit_url.query_pairs_mut().append_pair("comp", "blocklist");
+        headers.insert("content-type", HeaderValue::from_static("application/xml"));
+        let response = self.send(Method::PUT, commit_url, headers, Some(Bytes::from(xml))).await?;
+        let response = self.expect_status(response, StatusCode::CREATED, "PUT block list").await?;
+        response_version(response.headers())
+    }
+}
+
+#[async_trait::async_trait]
+impl WarmBackend for WarmBackendAzureBlob {
+    fn validate_remote_version_id(&self, version: &str) -> io::Result<()> {
+        validate_version(version)
+    }
+
+    async fn put(&self, object: &str, reader: ReaderImpl, length: i64) -> io::Result<String> {
+        self.write(object, reader, length, HashMap::new()).await
+    }
+
+    async fn put_with_meta(
+        &self,
+        object: &str,
+        reader: ReaderImpl,
+        length: i64,
+        metadata: HashMap<String, String>,
+    ) -> io::Result<String> {
+        self.write(object, reader, length, metadata).await
+    }
+
+    async fn get(&self, object: &str, version: &str, opts: WarmBackendGetOpts) -> io::Result<ReadCloser> {
+        let range = requested_range(&opts)?;
+        let mut headers = HeaderMap::new();
+        if let Some((start, end)) = range {
+            let value = end.map_or_else(|| format!("bytes={start}-"), |end| format!("bytes={start}-{end}"));
+            insert_header(&mut headers, "range", &value)?;
+        }
+        let response = self
+            .send(Method::GET, self.object_url(object, version)?, headers, None)
+            .await?;
+        let expected = if range.is_some() {
+            StatusCode::PARTIAL_CONTENT
+        } else {
+            StatusCode::OK
+        };
+        let response = self.expect_status(response, expected, "GET").await?;
+        validate_response_version(response.headers(), version)?;
+        let length = response
+            .headers()
+            .get("content-length")
+            .and_then(|value| value.to_str().ok())
+            .and_then(|value| value.parse::<u64>().ok())
+            .ok_or_else(|| invalid_data("Azure Blob GET omitted a valid Content-Length"))?;
+        if let Some(range) = range {
+            validate_content_range(response.headers(), range, length)?;
+        }
+        let body = stream::try_unfold((response, length), |(mut response, remaining)| async move {
+            loop {
+                match response.chunk().await.map_err(transport_error)? {
+                    Some(chunk) if chunk.is_empty() => continue,
+                    Some(chunk) => {
+                        let count =
+                            u64::try_from(chunk.len()).map_err(|_| invalid_data("Azure Blob response length overflow"))?;
+                        let remaining = remaining
+                            .checked_sub(count)
+                            .ok_or_else(|| invalid_data("Azure Blob returned more bytes than declared"))?;
+                        return Ok::<_, io::Error>(Some((chunk, (response, remaining))));
+                    }
+                    None if remaining == 0 => return Ok(None),
+                    None => {
+                        return Err(io::Error::new(
+                            ErrorKind::UnexpectedEof,
+                            "Azure Blob response ended before its declared length",
+                        ));
+                    }
+                }
+            }
+        });
+        Ok(ReadCloser::new(tokio_util::io::StreamReader::new(Box::pin(body))))
+    }
+
+    async fn remove(&self, object: &str, version: &str) -> io::Result<()> {
+        let mut headers = HeaderMap::new();
+        if version.is_empty() {
+            let Some(current) = self.head(object, "").await? else {
+                return Ok(());
+            };
+            if !response_version(&current)?.is_empty() {
+                return Err(invalid_data("Azure Blob became versioned; refusing an unversioned delete"));
+            }
+            let etag = current
+                .get("etag")
+                .ok_or_else(|| invalid_data("Azure Blob HEAD omitted the delete precondition"))?;
+            headers.insert("if-match", etag.clone());
+        }
+        let response = self
+            .send(Method::DELETE, self.object_url(object, version)?, headers, None)
+            .await?;
+        if response.status() == StatusCode::ACCEPTED {
+            return Ok(());
+        }
+        let err = service_error(response, "DELETE", self.request_timeout).await;
+        if err.kind() == ErrorKind::NotFound {
+            return Ok(());
+        }
+        Err(err)
+    }
+
+    async fn probe_transition_candidate(&self, object: &str) -> io::Result<TransitionCandidateProbe> {
+        Ok(match self.head(object, "").await? {
+            None => TransitionCandidateProbe::Missing,
+            Some(headers) => classify_head(&headers)?,
+        })
+    }
+
+    async fn probe_transition_version(&self, object: &str, version: &str) -> io::Result<TransitionCandidateProbe> {
+        if version.is_empty() {
+            return Err(invalid_input("Azure Blob exact probe requires a version ID"));
+        }
+        Ok(match self.head(object, version).await? {
+            None => TransitionCandidateProbe::Missing,
+            Some(_) => TransitionCandidateProbe::VersionedPresent(version.to_string()),
+        })
+    }
+
+    async fn in_use(&self) -> io::Result<bool> {
+        let mut url = self.container_url()?;
+        let prefix = if self.prefix.is_empty() {
+            String::new()
+        } else {
+            format!("{}/", self.prefix)
+        };
+        url.query_pairs_mut()
+            .append_pair("restype", "container")
+            .append_pair("comp", "list")
+            .append_pair("prefix", &prefix)
+            .append_pair("maxresults", "1");
+        let response = self.send(Method::GET, url, HeaderMap::new(), None).await?;
+        let response = self.expect_status(response, StatusCode::OK, "LIST").await?;
+        list_has_objects(&read_control_body(response, self.request_timeout).await?)
+    }
+}
+
+#[async_trait::async_trait]
+impl TransitionCandidateReconciler for WarmBackendAzureBlob {
+    async fn probe_transition_candidate_for(
+        &self,
+        object: &str,
+        identity: TransitionCandidateIdentity,
+    ) -> io::Result<TransitionCandidateProbe> {
+        let Some(headers) = self.head(object, "").await? else {
+            return Ok(TransitionCandidateProbe::Missing);
+        };
+        let mut metadata = HashMap::new();
+        for suffix in [SUFFIX_TRANSITION_TRANSACTION_ID, SUFFIX_TRANSITION_TIER_DESTINATION_ID] {
+            for key in [internal_key_rustfs(suffix), format!("{MINIO_INTERNAL_PREFIX}{suffix}")] {
+                if let Some(value) = headers.get(metadata_header_name(&key)) {
+                    let decoded = base64_simd::STANDARD
+                        .decode_to_vec(value.as_bytes())
+                        .map_err(|_| invalid_data("Azure Blob candidate metadata is not valid base64"))?;
+                    let value =
+                        String::from_utf8(decoded).map_err(|_| invalid_data("Azure Blob candidate metadata is not UTF-8"))?;
+                    metadata.insert(key, value);
+                }
+            }
+        }
+        if !transition_candidate_metadata_matches(&metadata, identity)? {
+            return Ok(TransitionCandidateProbe::Ambiguous);
+        }
+        classify_head(&headers)
+    }
+}
+
+fn validate_config(conf: &TierAzure) -> io::Result<()> {
+    if !(3..=24).contains(&conf.access_key.len())
+        || !conf
+            .access_key
+            .bytes()
+            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit())
+    {
+        return Err(invalid_input("Azure Blob account name must contain 3 to 24 lowercase letters or digits"));
+    }
+    let container = conf.bucket.as_bytes();
+    if !(3..=63).contains(&container.len())
+        || !container
+            .iter()
+            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || *byte == b'-')
+        || container.first() == Some(&b'-')
+        || container.last() == Some(&b'-')
+        || conf.bucket.contains("--")
+    {
+        return Err(invalid_input("Azure Blob container name is invalid"));
+    }
+    if !(conf.storage_class.is_empty() || conf.storage_class == "STANDARD")
+        || !conf.sp_auth.tenant_id.is_empty()
+        || !conf.sp_auth.client_id.is_empty()
+        || !conf.sp_auth.client_secret.is_empty()
+    {
+        return Err(invalid_input(
+            "Azure Blob tiers currently require an account key and the default storage class",
+        ));
+    }
+    if !conf.prefix.is_empty() {
+        validate_object_name(&conf.prefix)?;
+    }
+    Ok(())
+}
+
+fn decode_key(key: &str) -> io::Result<Vec<u8>> {
+    let key = base64_simd::STANDARD
+        .decode_to_vec(key.trim().as_bytes())
+        .map_err(|_| invalid_input("Azure Blob account key must be base64"))?;
+    if key.is_empty() {
+        return Err(invalid_input("Azure Blob account key must not be empty"));
+    }
+    Ok(key)
+}
+
+fn validate_object_name(name: &str) -> io::Result<()> {
+    if name.is_empty()
+        || name.len() > 1760
+        || name.chars().count() > 880
+        || name.split('/').any(|segment| matches!(segment, "." | ".."))
+    {
+        return Err(invalid_input(
+            "Azure Blob tier name exceeds Stack Hub limits or contains a standalone dot segment",
+        ));
+    }
+    Ok(())
+}
+
+fn validate_version(version: &str) -> io::Result<()> {
+    if version.len() > 1024 || version.bytes().any(|byte| !byte.is_ascii_graphic()) {
+        return Err(invalid_input("Azure Blob version ID is invalid"));
+    }
+    Ok(())
+}
+
+fn response_version(headers: &HeaderMap) -> io::Result<String> {
+    let Some(version) = headers.get("x-ms-version-id") else {
+        return Ok(String::new());
+    };
+    let version = version
+        .to_str()
+        .map_err(|_| invalid_data("Azure Blob version ID is not ASCII"))?;
+    validate_version(version)?;
+    Ok(version.to_string())
+}
+
+fn validate_response_version(headers: &HeaderMap, requested: &str) -> io::Result<()> {
+    if response_version(headers)? != requested {
+        return Err(invalid_data("Azure Blob did not return the requested exact version"));
+    }
+    Ok(())
+}
+
+fn classify_head(headers: &HeaderMap) -> io::Result<TransitionCandidateProbe> {
+    let version = response_version(headers)?;
+    Ok(if version.is_empty() {
+        TransitionCandidateProbe::UnversionedPresent
+    } else {
+        TransitionCandidateProbe::VersionedPresent(version)
+    })
+}
+
+fn block_size(size: u64) -> io::Result<u64> {
+    if size > MAX_BLOCK_BYTES * MAX_BLOCKS {
+        return Err(invalid_input("Azure Blob exceeds the Stack Hub block count or object size limit"));
+    }
+    Ok(size.div_ceil(MAX_BLOCKS).max(MIN_BLOCK_BYTES))
+}
+
+async fn read_part(reader: &mut (dyn AsyncRead + Send + Sync + Unpin), length: u64) -> io::Result<Bytes> {
+    let length = usize::try_from(length).map_err(|_| invalid_input("Azure Blob block length is not representable"))?;
+    let mut bytes = vec![0; length];
+    reader.read_exact(&mut bytes).await?;
+    Ok(Bytes::from(bytes))
+}
+
+async fn ensure_eof(reader: &mut (dyn AsyncRead + Send + Sync + Unpin)) -> io::Result<()> {
+    if reader.read(&mut [0_u8; 1]).await? != 0 {
+        return Err(invalid_input("Azure Blob upload body exceeds its declared length"));
+    }
+    Ok(())
+}
+
+fn metadata_header_name(key: &str) -> String {
+    format!("{METADATA_PREFIX}{}", hex(key.as_bytes()))
+}
+
+fn metadata_headers(metadata: HashMap<String, String>) -> io::Result<HeaderMap> {
+    let opts = build_transition_put_options(String::new(), metadata);
+    let mut headers = HeaderMap::new();
+    for (name, value) in [
+        ("x-ms-blob-content-type", opts.content_type),
+        ("x-ms-blob-content-encoding", opts.content_encoding),
+        ("x-ms-blob-content-language", opts.content_language),
+        ("x-ms-blob-content-disposition", opts.content_disposition),
+        ("x-ms-blob-cache-control", opts.cache_control),
+    ] {
+        if !value.is_empty() {
+            insert_header(&mut headers, name, &value)?;
+        }
+    }
+    let mut size = 0_usize;
+    for (key, value) in opts.user_metadata {
+        let key = key.to_ascii_lowercase();
+        let key = key.strip_prefix("x-amz-meta-").unwrap_or(&key);
+        let name = metadata_header_name(key);
+        let value = base64_simd::STANDARD.encode_to_string(value.as_bytes());
+        size = size.saturating_add(name.len()).saturating_add(value.len());
+        if size > MAX_METADATA_BYTES {
+            return Err(invalid_input("Azure Blob encoded metadata exceeds 8 KiB"));
+        }
+        if headers.contains_key(&name) {
+            return Err(invalid_input("Azure Blob metadata contains conflicting names"));
+        }
+        insert_header(&mut headers, &name, &value)?;
+    }
+    Ok(headers)
+}
+
+fn requested_range(opts: &WarmBackendGetOpts) -> io::Result<Option<(u64, Option<u64>)>> {
+    let start = u64::try_from(opts.start_offset).map_err(|_| invalid_input("Azure Blob range starts before zero"))?;
+    let length = u64::try_from(opts.length).map_err(|_| invalid_input("Azure Blob range length is negative"))?;
+    if length > 0 {
+        let end = start
+            .checked_add(length - 1)
+            .ok_or_else(|| invalid_input("Azure Blob range overflows"))?;
+        return Ok(Some((start, Some(end))));
+    }
+    Ok((start > 0).then_some((start, None)))
+}
+
+fn validate_content_range(headers: &HeaderMap, requested: (u64, Option<u64>), length: u64) -> io::Result<()> {
+    let value = headers
+        .get("content-range")
+        .and_then(|value| value.to_str().ok())
+        .unwrap_or_default();
+    let range = value
+        .strip_prefix("bytes ")
+        .and_then(|value| value.split_once('/'))
+        .and_then(|(range, total)| {
+            let (start, end) = range.split_once('-')?;
+            Some((start.parse::<u64>().ok()?, end.parse::<u64>().ok()?, total.parse::<u64>().ok()?))
+        });
+    let Some((start, end, total)) = range else {
+        return Err(invalid_data("Azure Blob returned an invalid Content-Range"));
+    };
+    if start != requested.0
+        || end < start
+        || end >= total
+        || end != requested.1.unwrap_or(total - 1).min(total - 1)
+        || end.checked_sub(start).and_then(|size| size.checked_add(1)) != Some(length)
+    {
+        return Err(invalid_data("Azure Blob returned a different range from the one requested"));
+    }
+    Ok(())
+}
+
+async fn read_control_body(mut response: reqwest::Response, timeout: Duration) -> io::Result<Vec<u8>> {
+    tokio::time::timeout(timeout, async move {
+        let mut body = Vec::new();
+        while let Some(chunk) = response.chunk().await.map_err(transport_error)? {
+            if body.len().saturating_add(chunk.len()) > MAX_CONTROL_BYTES {
+                return Err(invalid_data("Azure Blob control response exceeds 64 KiB"));
+            }
+            body.extend_from_slice(&chunk);
+        }
+        Ok(body)
+    })
+    .await
+    .map_err(|_| io::Error::new(ErrorKind::TimedOut, "Azure Blob control response timed out"))?
+}
+
+async fn service_error(response: reqwest::Response, operation: &'static str, timeout: Duration) -> io::Error {
+    let status = response.status();
+    let request_id = diagnostic_token(
+        response
+            .headers()
+            .get("x-ms-request-id")
+            .and_then(|value| value.to_str().ok()),
+    );
+    let mut code = diagnostic_token(
+        response
+            .headers()
+            .get("x-ms-error-code")
+            .and_then(|value| value.to_str().ok()),
+    );
+    if code == "unknown"
+        && let Ok(body) = read_control_body(response, timeout).await
+        && let Some(xml_code) = xml_error_code(&body)
+    {
+        code = diagnostic_token(Some(&xml_code));
+    }
+    let kind = if status == StatusCode::NOT_FOUND && code == "BlobNotFound" {
+        ErrorKind::NotFound
+    } else if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) {
+        ErrorKind::PermissionDenied
+    } else if status == StatusCode::PRECONDITION_FAILED {
+        ErrorKind::AlreadyExists
+    } else {
+        ErrorKind::Other
+    };
+    io::Error::new(
+        kind,
+        AzureBlobError {
+            operation,
+            status,
+            code,
+            request_id,
+        },
+    )
+}
+
+fn diagnostic_token(value: Option<&str>) -> String {
+    value
+        .filter(|value| {
+            !value.is_empty() && value.len() <= 128 && value.bytes().all(|byte| byte.is_ascii_alphanumeric() || byte == b'-')
+        })
+        .unwrap_or("unknown")
+        .to_string()
+}
+
+fn xml_error_code(body: &[u8]) -> Option<String> {
+    let mut reader = Reader::from_reader(body);
+    let mut depth = 0_u32;
+    let mut root = false;
+    let mut code = None;
+    loop {
+        match reader.read_event().ok()? {
+            Event::Start(start) => {
+                if depth == 0 {
+                    if root || start.local_name().as_ref() != "Error" {
+                        return None;
+                    }
+                    root = true;
+                }
+                if depth == 1 && start.local_name().as_ref() == "Code" {
+                    if code.is_some() {
+                        return None;
+                    }
+                    let text = reader.read_text(start.name()).ok()?;
+                    code = Some(quick_xml::escape::unescape(text.as_ref()).ok()?.into_owned());
+                    continue;
+                }
+                depth += 1;
+                if depth > 16 {
+                    return None;
+                }
+            }
+            Event::End(_) => depth = depth.checked_sub(1)?,
+            Event::Text(text) if depth == 0 && !text.as_ref().trim().is_empty() => return None,
+            Event::CData(_) if depth == 0 => return None,
+            Event::Eof => return (root && depth == 0).then_some(code).flatten(),
+            Event::DocType(_) => return None,
+            _ => {}
+        }
+    }
+}
+
+fn list_has_objects(body: &[u8]) -> io::Result<bool> {
+    let mut reader = Reader::from_reader(body);
+    let mut depth = 0_u32;
+    let mut root = false;
+    let mut blobs = false;
+    let mut found = false;
+    loop {
+        let event = reader
+            .read_event()
+            .map_err(|_| invalid_data("Azure Blob listing is not valid XML"))?;
+        let empty = matches!(event, Event::Empty(_));
+        match event {
+            Event::Start(start) | Event::Empty(start) => {
+                if depth == 0 && (root || start.local_name().as_ref() != "EnumerationResults") {
+                    return Err(invalid_data("Azure Blob listing has an unexpected root"));
+                }
+                root |= start.local_name().as_ref() == "EnumerationResults";
+                if start.local_name().as_ref() == "Blobs" {
+                    if blobs || depth != 1 {
+                        return Err(invalid_data("Azure Blob listing has an invalid Blobs element"));
+                    }
+                    blobs = true;
+                }
+                found |= matches!(start.local_name().as_ref(), "Blob" | "BlobPrefix");
+                if !empty && start.local_name().as_ref() == "NextMarker" {
+                    let marker = reader
+                        .read_text(start.name())
+                        .map_err(|_| invalid_data("Azure Blob listing marker is invalid"))?;
+                    found |= !marker.is_empty();
+                    continue;
+                }
+                if !empty {
+                    depth += 1;
+                    if depth > 16 {
+                        return Err(invalid_data("Azure Blob listing exceeds the XML depth limit"));
+                    }
+                }
+            }
+            Event::End(_) => {
+                depth = depth
+                    .checked_sub(1)
+                    .ok_or_else(|| invalid_data("Azure Blob listing has an invalid closing tag"))?;
+            }
+            Event::DocType(_) => return Err(invalid_data("Azure Blob listing must not contain a document type")),
+            Event::Text(text) if depth == 0 && !text.as_ref().trim().is_empty() => {
+                return Err(invalid_data("Azure Blob listing contains text outside its root"));
+            }
+            Event::CData(_) if depth == 0 => return Err(invalid_data("Azure Blob listing contains text outside its root")),
+            Event::Eof if root && blobs && depth == 0 => return Ok(found),
+            Event::Eof => return Err(invalid_data("Azure Blob listing is incomplete")),
+            _ => {}
+        }
+    }
+}
+
+fn insert_header(headers: &mut HeaderMap, name: &str, value: &str) -> io::Result<()> {
+    let name =
+        HeaderName::from_bytes(name.as_bytes()).map_err(|_| invalid_input("Azure Blob metadata header name is invalid"))?;
+    let value = HeaderValue::from_str(value).map_err(|_| invalid_input("Azure Blob metadata header value is invalid"))?;
+    headers.insert(name, value);
+    Ok(())
+}
+
+fn invalid_input(message: &'static str) -> io::Error {
+    io::Error::new(ErrorKind::InvalidInput, message)
+}
+
+fn invalid_data(message: &'static str) -> io::Error {
+    io::Error::new(ErrorKind::InvalidData, message)
+}
+
+fn transport_error(err: reqwest::Error) -> io::Error {
+    let kind = if err.is_timeout() {
+        ErrorKind::TimedOut
+    } else {
+        ErrorKind::Other
+    };
+    io::Error::new(kind, err.without_url())
+}
+
+#[cfg(test)]
+#[path = "warm_backend_azure_blob_tests.rs"]
+mod tests;

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

@@ -0,0 +1,649 @@
+// 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 super::*;
+use crate::services::tier::warm_backend::{WarmBackendImpl, check_warm_backend};
+use rustfs_utils::http::metadata_compat::insert_str;
+use tokio::io::AsyncWriteExt;
+use tokio::net::{TcpListener, TcpStream};
+use tokio::task::JoinHandle;
+
+struct Reply {
+    status: StatusCode,
+    headers: HeaderMap,
+    body: Bytes,
+    disconnect: bool,
+}
+
+fn reply(status: StatusCode, headers: &[(&str, &str)], body: &[u8]) -> Reply {
+    let mut map = HeaderMap::new();
+    for (name, value) in headers {
+        insert_header(&mut map, name, value).expect("fixture headers should be valid");
+    }
+    Reply {
+        status,
+        headers: map,
+        body: Bytes::copy_from_slice(body),
+        disconnect: false,
+    }
+}
+
+fn present(version: Option<&str>) -> Reply {
+    let mut response = reply(StatusCode::OK, &[("content-length", "6"), ("etag", "\"original\"")], b"");
+    if let Some(version) = version {
+        insert_header(&mut response.headers, "x-ms-version-id", version).expect("fixture version should be valid");
+    }
+    response
+}
+
+fn missing() -> Reply {
+    reply(StatusCode::NOT_FOUND, &[("x-ms-error-code", "BlobNotFound")], b"")
+}
+
+#[derive(Debug)]
+struct Captured {
+    method: String,
+    target: String,
+    headers: HeaderMap,
+    body: Vec<u8>,
+}
+
+async fn capture(stream: &mut TcpStream) -> Captured {
+    let mut bytes = Vec::new();
+    let boundary = loop {
+        let mut buffer = [0_u8; 4096];
+        let count = stream.read(&mut buffer).await.expect("fixture request should be readable");
+        assert_ne!(count, 0, "request must include a complete HTTP head");
+        bytes.extend_from_slice(&buffer[..count]);
+        if let Some(index) = bytes.windows(4).position(|window| window == b"\r\n\r\n") {
+            break index + 4;
+        }
+        assert!(bytes.len() <= 32 * 1024, "request headers must remain bounded");
+    };
+    let text = std::str::from_utf8(&bytes[..boundary]).expect("request headers should be ASCII");
+    let mut lines = text.split("\r\n");
+    let mut request_line = lines.next().expect("request line").split_whitespace();
+    let method = request_line.next().expect("request method").to_string();
+    let target = request_line.next().expect("request target").to_string();
+    let mut headers = HeaderMap::new();
+    for line in lines.filter(|line| !line.is_empty()) {
+        let (name, value) = line.split_once(':').expect("request header should include a colon");
+        insert_header(&mut headers, name, value.trim()).expect("request headers should parse");
+    }
+    let length = headers
+        .get("content-length")
+        .map(|value| {
+            value
+                .to_str()
+                .expect("length should be ASCII")
+                .parse::<usize>()
+                .expect("length should be numeric")
+        })
+        .unwrap_or_default();
+    assert!(length <= 17 * 1024 * 1024, "fixture only accepts its bounded test bodies");
+    let mut body = bytes[boundary..].to_vec();
+    let received = body.len();
+    body.resize(length, 0);
+    stream
+        .read_exact(&mut body[received..])
+        .await
+        .expect("request body should match its declared length");
+    Captured {
+        method,
+        target,
+        headers,
+        body,
+    }
+}
+
+async fn fixture(responses: Vec<Reply>) -> (WarmBackendAzureBlob, JoinHandle<Vec<Captured>>) {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("fixture listener should bind");
+    let address = listener.local_addr().expect("fixture address");
+    let task = tokio::spawn(async move {
+        let mut requests = Vec::new();
+        for mut response in responses {
+            let (mut stream, _) = tokio::time::timeout(Duration::from_secs(10), listener.accept())
+                .await
+                .expect("expected request must arrive")
+                .expect("fixture should accept");
+            requests.push(capture(&mut stream).await);
+            if response.disconnect {
+                continue;
+            }
+            if !response.headers.contains_key("content-length") {
+                insert_header(&mut response.headers, "content-length", &response.body.len().to_string())
+                    .expect("fixture length should be valid");
+            }
+            let mut wire = format!(
+                "HTTP/1.1 {} {}\r\nConnection: close\r\n",
+                response.status.as_u16(),
+                response.status.canonical_reason().unwrap_or("")
+            );
+            for (name, value) in &response.headers {
+                wire.push_str(&format!("{name}: {}\r\n", value.to_str().expect("fixture header is ASCII")));
+            }
+            wire.push_str("\r\n");
+            stream.write_all(wire.as_bytes()).await.expect("fixture should send headers");
+            stream.write_all(&response.body).await.expect("fixture should send body");
+        }
+        requests
+    });
+    rustfs_tls_runtime::ensure_crypto_provider().expect("fixture TLS provider should initialize");
+    let client = reqwest::Client::builder()
+        .no_proxy()
+        .redirect(reqwest::redirect::Policy::none())
+        .retry(reqwest::retry::never())
+        .read_timeout(Duration::from_secs(5))
+        .build()
+        .expect("fixture client should build");
+    (
+        WarmBackendAzureBlob {
+            client,
+            endpoint: Url::parse(&format!("http://{address}")).expect("fixture endpoint"),
+            account: "account".to_string(),
+            key: vec![7; 32],
+            container: "container".to_string(),
+            prefix: "cold".to_string(),
+            request_timeout: Duration::from_secs(5),
+        },
+        task,
+    )
+}
+
+#[tokio::test]
+async fn mutation_probe_uses_blob_shared_key_and_cleans_up_with_an_etag() {
+    let responses = vec![
+        reply(StatusCode::CREATED, &[], b""),
+        present(None),
+        reply(StatusCode::PARTIAL_CONTENT, &[("content-range", "bytes 0-5/6")], b"RustFS"),
+        present(None),
+        reply(StatusCode::ACCEPTED, &[], b""),
+        missing(),
+    ];
+    let (backend, task) = fixture(responses).await;
+    let backend: WarmBackendImpl = Box::new(backend);
+    check_warm_backend(Some(&backend))
+        .await
+        .expect("native Blob mutation probe should succeed");
+    let requests = task.await.expect("fixture should join");
+    assert_eq!(
+        requests.iter().map(|request| request.method.as_str()).collect::<Vec<_>>(),
+        ["PUT", "HEAD", "GET", "HEAD", "DELETE", "HEAD"]
+    );
+    assert!(requests[0].target.starts_with("/container/cold/rustfs-tier-probe-"));
+    assert_eq!(requests[0].body, b"RustFS");
+    assert_eq!(requests[0].headers["x-ms-blob-type"], "BlockBlob");
+    assert_eq!(requests[0].headers["if-none-match"], "*");
+    assert_eq!(requests[4].headers["if-match"], "\"original\"");
+    for request in &requests {
+        assert_eq!(request.target, requests[0].target, "the whole probe must address the same blob");
+        assert_eq!(request.headers["x-ms-version"], API_VERSION);
+        assert!(
+            request.headers["authorization"]
+                .to_str()
+                .expect("authorization is ASCII")
+                .starts_with("SharedKey account:")
+        );
+        assert!(!request.headers.contains_key("x-amz-date"));
+        assert!(!request.target.contains("location"));
+        assert!(!request.target.contains("versionid"));
+    }
+}
+
+#[tokio::test]
+async fn rejected_put_preserves_azure_error_and_never_deletes_an_existing_blob() {
+    for (status, code) in [
+        (StatusCode::BAD_REQUEST, "InvalidHeaderValue"),
+        (StatusCode::PRECONDITION_FAILED, "ConditionNotMet"),
+    ] {
+        let (backend, task) = fixture(vec![reply(
+            status,
+            &[("x-ms-error-code", code), ("x-ms-request-id", "abc-123")],
+            b"<Error><Code>PRIVATE-RESPONSE-BODY</Code></Error>",
+        )])
+        .await;
+        let backend: WarmBackendImpl = Box::new(backend);
+        let err = check_warm_backend(Some(&backend))
+            .await
+            .expect_err("deterministic rejected write must fail");
+        assert!(err.message.contains(code));
+        assert!(err.message.contains("abc-123"));
+        assert!(!err.message.contains("PRIVATE-RESPONSE-BODY"));
+        assert!(!err.message.contains("cleanup is incomplete"));
+        let requests = task.await.expect("fixture should join");
+        assert_eq!(requests.len(), 1, "a rejected conditional PUT must not trigger cleanup");
+    }
+}
+
+#[tokio::test]
+async fn empty_version_headers_use_the_unversioned_read_and_delete_paths() {
+    let mut head = present(Some(""));
+    head.headers.insert("content-length", HeaderValue::from_static("0"));
+    let (backend, task) = fixture(vec![
+        reply(StatusCode::CREATED, &[("x-ms-version-id", "")], b""),
+        reply(StatusCode::OK, &[("x-ms-version-id", "")], b""),
+        head,
+        reply(StatusCode::ACCEPTED, &[], b""),
+    ])
+    .await;
+    let version = backend
+        .put("empty", ReaderImpl::Body(Bytes::new()), 0)
+        .await
+        .expect("empty version is unversioned");
+    assert!(version.is_empty());
+    let mut reader = backend
+        .get("empty", &version, WarmBackendGetOpts::default())
+        .await
+        .expect("unversioned read");
+    let mut body = Vec::new();
+    reader.read_to_end(&mut body).await.expect("empty blob should read");
+    assert!(body.is_empty());
+    backend
+        .remove("empty", &version)
+        .await
+        .expect("unversioned conditional delete");
+    for request in task.await.expect("fixture should join") {
+        assert_eq!(request.target, "/container/cold/empty");
+    }
+}
+
+#[tokio::test]
+async fn rejected_credentials_are_reported_without_an_uncertain_cleanup() {
+    let (backend, task) = fixture(vec![reply(
+        StatusCode::FORBIDDEN,
+        &[("x-ms-error-code", "AuthenticationFailed")],
+        b"",
+    )])
+    .await;
+    let backend: WarmBackendImpl = Box::new(backend);
+    let err = check_warm_backend(Some(&backend))
+        .await
+        .expect_err("invalid credentials must fail");
+    assert_eq!(err.code, crate::services::tier::tier_handlers::ERR_TIER_INVALID_CREDENTIALS.code);
+    assert_eq!(task.await.expect("fixture should join").len(), 1);
+}
+
+#[tokio::test]
+async fn lost_put_response_is_not_replayed_and_its_candidate_is_cleaned() {
+    let mut lost = reply(StatusCode::CREATED, &[], b"");
+    lost.disconnect = true;
+    let (backend, task) = fixture(vec![
+        lost,
+        present(None),
+        present(None),
+        reply(StatusCode::ACCEPTED, &[], b""),
+        missing(),
+        missing(),
+    ])
+    .await;
+    let backend: WarmBackendImpl = Box::new(backend);
+    let err = check_warm_backend(Some(&backend))
+        .await
+        .expect_err("a lost write acknowledgement must remain a failure");
+    assert!(!err.message.contains("cleanup is incomplete"));
+    let requests = task.await.expect("fixture should join");
+    assert_eq!(requests.iter().filter(|request| request.method == "PUT").count(), 1);
+    assert_eq!(requests.iter().filter(|request| request.method == "DELETE").count(), 1);
+}
+
+#[tokio::test]
+async fn block_upload_commits_only_the_complete_declared_body() {
+    let (backend, task) = fixture(vec![
+        reply(StatusCode::CREATED, &[], b""),
+        reply(StatusCode::CREATED, &[], b""),
+        reply(StatusCode::CREATED, &[], b""),
+    ])
+    .await;
+    let length = usize::try_from(MIN_BLOCK_BYTES + 5).expect("fixture size");
+    let bytes = Bytes::from(vec![42; length]);
+    assert_eq!(
+        backend
+            .put(
+                "large",
+                ReaderImpl::ObjectBody(ReadCloser::new(Cursor::new(bytes.clone()))),
+                i64::try_from(length).expect("fixture length")
+            )
+            .await
+            .expect("block upload"),
+        ""
+    );
+    let requests = task.await.expect("fixture should join");
+    assert_eq!(requests.len(), 3);
+    let mut block_ids = Vec::new();
+    let mut body = Vec::new();
+    for request in &requests[..2] {
+        let url = Url::parse(&format!("http://fixture{}", request.target)).expect("captured URL");
+        let query: HashMap<_, _> = url.query_pairs().into_owned().collect();
+        assert_eq!(query["comp"], "block");
+        block_ids.push(query["blockid"].clone());
+        assert!(request.headers.contains_key("content-md5"));
+        body.extend_from_slice(&request.body);
+    }
+    assert_eq!(body, bytes);
+    assert_ne!(block_ids[0], block_ids[1]);
+    assert!(requests[2].target.ends_with("?comp=blocklist"));
+    assert_eq!(requests[2].headers["if-none-match"], "*");
+    let expected = format!(
+        "<?xml version=\"1.0\" encoding=\"utf-8\"?><BlockList><Uncommitted>{}</Uncommitted><Uncommitted>{}</Uncommitted></BlockList>",
+        block_ids[0], block_ids[1],
+    );
+    assert_eq!(requests[2].body, expected.as_bytes());
+}
+
+#[tokio::test]
+async fn server_error_remains_visible_after_successful_probe_compensation() {
+    let (backend, task) = fixture(vec![
+        reply(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            &[("x-ms-error-code", "InternalError"), ("x-ms-request-id", "error-123")],
+            b"",
+        ),
+        present(None),
+        present(None),
+        reply(StatusCode::ACCEPTED, &[], b""),
+        missing(),
+        missing(),
+    ])
+    .await;
+    let backend: WarmBackendImpl = Box::new(backend);
+    let error = check_warm_backend(Some(&backend))
+        .await
+        .expect_err("failed upload remains an error after cleanup");
+    assert!(error.message.contains("500"));
+    assert!(error.message.contains("InternalError"));
+    assert!(error.message.contains("error-123"));
+    assert!(!error.message.contains("cleanup is incomplete"));
+    let requests = task.await.expect("fixture should join");
+    assert_eq!(requests.iter().filter(|request| request.method == "PUT").count(), 1);
+    assert_eq!(requests.iter().filter(|request| request.method == "DELETE").count(), 1);
+}
+
+#[tokio::test]
+async fn empty_blob_and_declared_length_failures_are_handled_before_commit() {
+    let (backend, task) = fixture(vec![reply(StatusCode::CREATED, &[], b"")]).await;
+    backend
+        .put("empty", ReaderImpl::Body(Bytes::new()), 0)
+        .await
+        .expect("empty upload");
+    let requests = task.await.expect("fixture should join");
+    assert_eq!(requests[0].headers["content-length"], "0");
+    assert_eq!(requests[0].headers["content-md5"], "1B2M2Y8AsgTpgAmY7PhCfg==");
+    for (body, kind) in [
+        (b"short".as_slice(), ErrorKind::UnexpectedEof),
+        (b"too-long".as_slice(), ErrorKind::InvalidInput),
+    ] {
+        let (backend, task) = fixture(vec![]).await;
+        let err = backend
+            .put("bad", ReaderImpl::Body(Bytes::copy_from_slice(body)), 6)
+            .await
+            .expect_err("length mismatch");
+        assert_eq!(err.kind(), kind);
+        assert!(task.await.expect("fixture should join").is_empty());
+    }
+    let (backend, task) = fixture(vec![reply(StatusCode::CREATED, &[], b"")]).await;
+    let body = Bytes::from(vec![0; usize::try_from(MIN_BLOCK_BYTES + 1).expect("fixture size")]);
+    let err = backend
+        .put(
+            "short-block",
+            ReaderImpl::Body(body),
+            i64::try_from(MIN_BLOCK_BYTES + 2).expect("fixture length"),
+        )
+        .await
+        .expect_err("short final block");
+    assert_eq!(err.kind(), ErrorKind::UnexpectedEof);
+    assert_eq!(
+        task.await.expect("fixture should join").len(),
+        1,
+        "short input must never commit its staged blocks"
+    );
+}
+
+#[tokio::test]
+async fn range_get_preserves_key_encoding_and_exact_version_for_delete() {
+    let version = "2026-11-06T01:02:03.0000000Z";
+    let (backend, task) = fixture(vec![
+        reply(
+            StatusCode::PARTIAL_CONTENT,
+            &[("content-range", "bytes 1-3/6"), ("x-ms-version-id", version)],
+            b"ust",
+        ),
+        reply(StatusCode::ACCEPTED, &[], b""),
+    ])
+    .await;
+    let object = "dir/a b+?#%2F.bin";
+    let mut reader = backend
+        .get(
+            object,
+            version,
+            WarmBackendGetOpts {
+                start_offset: 1,
+                length: 3,
+            },
+        )
+        .await
+        .expect("versioned range read");
+    let mut body = Vec::new();
+    reader.read_to_end(&mut body).await.expect("range body");
+    assert_eq!(body, b"ust");
+    backend.remove_exact(object, version).await.expect("exact delete");
+    let requests = task.await.expect("fixture should join");
+    assert_eq!(requests.len(), 2, "exact delete needs no current-version HEAD");
+    for request in requests {
+        let url = Url::parse(&format!("http://fixture{}", request.target)).expect("captured URL");
+        assert_eq!(
+            urlencoding::decode(url.path()).expect("decoded path"),
+            format!("/container/cold/{object}")
+        );
+        let query: Vec<_> = url.query_pairs().into_owned().collect();
+        assert_eq!(query, [("versionid".to_string(), version.to_string())]);
+    }
+}
+
+#[tokio::test]
+async fn version_drift_and_incorrect_ranges_fail_closed() {
+    let (backend, task) = fixture(vec![present(Some("new-version"))]).await;
+    let err = backend
+        .remove("changed", "")
+        .await
+        .expect_err("unversioned delete must reject version drift");
+    assert!(err.to_string().contains("became versioned"));
+    assert_eq!(task.await.expect("fixture should join").len(), 1);
+    for headers in [
+        vec![("content-range", "bytes 0-2/6")],
+        vec![("content-range", "bytes 1-3/6"), ("x-ms-version-id", "unexpected-version")],
+    ] {
+        let (backend, task) = fixture(vec![reply(StatusCode::PARTIAL_CONTENT, &headers, b"ust")]).await;
+        let result = backend
+            .get(
+                "object",
+                "",
+                WarmBackendGetOpts {
+                    start_offset: 1,
+                    length: 3,
+                },
+            )
+            .await;
+        assert!(result.is_err(), "wrong offset or an unexpected version cannot satisfy the read");
+        task.await.expect("fixture should join");
+    }
+}
+
+#[tokio::test]
+async fn unversioned_delete_keeps_the_head_precondition_when_the_blob_changes() {
+    let (backend, task) = fixture(vec![
+        present(None),
+        reply(StatusCode::PRECONDITION_FAILED, &[("x-ms-error-code", "ConditionNotMet")], b""),
+    ])
+    .await;
+    assert!(backend.remove("changed", "").await.is_err());
+    let requests = task.await.expect("fixture should join");
+    assert_eq!(requests[1].headers["if-match"], "\"original\"");
+}
+
+#[tokio::test]
+async fn absence_requires_blob_not_found_and_container_listing_is_bounded() {
+    for (response, missing_expected) in [
+        (missing(), true),
+        (reply(StatusCode::NOT_FOUND, &[("x-ms-error-code", "ContainerNotFound")], b""), false),
+        (reply(StatusCode::NOT_FOUND, &[], b""), false),
+    ] {
+        let (backend, task) = fixture(vec![response]).await;
+        let result = backend.probe_transition_candidate("object").await;
+        if missing_expected {
+            assert_eq!(result.expect("explicit blob absence"), TransitionCandidateProbe::Missing);
+        } else {
+            assert!(result.is_err(), "an absent container or an ambiguous HEAD is not blob absence");
+        }
+        task.await.expect("fixture should join");
+    }
+    for (body, in_use) in [
+        (b"<EnumerationResults><Blobs/><NextMarker/></EnumerationResults>".as_slice(), false),
+        (
+            b"<EnumerationResults><Blobs><Blob><Name>cold/object</Name></Blob></Blobs><NextMarker/></EnumerationResults>"
+                .as_slice(),
+            true,
+        ),
+        (
+            b"<EnumerationResults><Blobs/><NextMarker>next</NextMarker></EnumerationResults>".as_slice(),
+            true,
+        ),
+    ] {
+        let (backend, task) = fixture(vec![reply(StatusCode::OK, &[], body)]).await;
+        assert_eq!(backend.in_use().await.expect("native container listing"), in_use);
+        let requests = task.await.expect("fixture should join");
+        assert!(
+            requests[0]
+                .target
+                .contains("restype=container&comp=list&prefix=cold%2F&maxresults=1")
+        );
+    }
+    let oversized = vec![b' '; MAX_CONTROL_BYTES + 1];
+    let (backend, task) = fixture(vec![reply(StatusCode::OK, &[], &oversized)]).await;
+    assert!(backend.in_use().await.is_err(), "oversized listing cannot establish an empty destination");
+    task.await.expect("fixture should join");
+}
+
+#[tokio::test]
+async fn recovery_requires_the_uploaded_transaction_and_destination_identity() {
+    let identity = TransitionCandidateIdentity {
+        transaction_id: uuid::Uuid::new_v4(),
+        destination_id: [3; 32],
+    };
+    let mut metadata = HashMap::new();
+    insert_str(&mut metadata, SUFFIX_TRANSITION_TRANSACTION_ID, identity.transaction_id.to_string());
+    insert_str(&mut metadata, SUFFIX_TRANSITION_TIER_DESTINATION_ID, hex(identity.destination_id));
+    let headers = metadata_headers(metadata).expect("identity metadata should encode for Azure");
+    assert_eq!(
+        headers
+            .keys()
+            .filter(|name| name.as_str().starts_with(METADATA_PREFIX))
+            .count(),
+        4
+    );
+    let mut owned = present(Some("owned-version"));
+    owned.headers.extend(headers);
+    let (backend, task) = fixture(vec![owned, present(None)]).await;
+    assert_eq!(
+        backend
+            .probe_transition_candidate_for("object", identity)
+            .await
+            .expect("owned candidate"),
+        TransitionCandidateProbe::VersionedPresent("owned-version".to_string())
+    );
+    assert_eq!(
+        backend
+            .probe_transition_candidate_for("object", identity)
+            .await
+            .expect("foreign candidate classification"),
+        TransitionCandidateProbe::Ambiguous
+    );
+    task.await.expect("fixture should join");
+}
+
+#[test]
+fn configuration_xml_and_size_limits_fail_before_network_io() {
+    let mut config = TierAzure {
+        access_key: "account".to_string(),
+        secret_key: base64_simd::STANDARD.encode_to_string([7; 32]),
+        bucket: "container".to_string(),
+        ..Default::default()
+    };
+    assert!(
+        WarmBackendAzureBlob::new(&config).is_ok(),
+        "public endpoint should be derived from the account"
+    );
+    config.endpoint = "http://127.0.0.1:10000".to_string();
+    assert!(
+        WarmBackendAzureBlob::new(&config).is_err(),
+        "production constructor must apply the outbound guard"
+    );
+    config.endpoint = "https://account.blob.example/path".to_string();
+    assert!(
+        WarmBackendAzureBlob::new(&config).is_err(),
+        "an endpoint path cannot rewrite object addressing"
+    );
+    for name in [".", "..", "a/../b", "a/./b"] {
+        assert!(validate_object_name(name).is_err());
+    }
+    assert_eq!(
+        block_size(MAX_BLOCK_BYTES * MAX_BLOCKS).expect("largest supported object"),
+        MAX_BLOCK_BYTES
+    );
+    assert!(block_size(MAX_BLOCK_BYTES * MAX_BLOCKS + 1).is_err());
+    for xml in [
+        b"<html>not a container</html>".as_slice(),
+        b"<EnumerationResults><Blobs>".as_slice(),
+        b"<!DOCTYPE x><EnumerationResults><Blobs/></EnumerationResults>".as_slice(),
+        b"<EnumerationResults><Other><Blobs/></Other></EnumerationResults>".as_slice(),
+        b"garbage<EnumerationResults><Blobs/></EnumerationResults>".as_slice(),
+    ] {
+        assert!(list_has_objects(xml).is_err());
+    }
+    assert_eq!(
+        xml_error_code(b"<Error><Code>BlobNotFound</Code></Error>").as_deref(),
+        Some("BlobNotFound")
+    );
+    assert!(xml_error_code(b"<Error><Code>BlobNotFound</Code>").is_none());
+    assert!(xml_error_code(b"<Error><Code>BlobNotFound</Code></Error>garbage").is_none());
+}
+
+#[tokio::test]
+async fn control_response_has_a_total_deadline_after_headers_arrive() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("control fixture should bind");
+    let address = listener.local_addr().expect("control fixture address");
+    let (release, wait) = tokio::sync::oneshot::channel();
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.expect("control fixture should accept");
+        capture(&mut stream).await;
+        stream
+            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 16\r\n\r\n")
+            .await
+            .expect("control response headers");
+        let _ = wait.await;
+    });
+    let (mut backend, empty) = fixture(Vec::new()).await;
+    empty.await.expect("unused fixture should close");
+    backend.endpoint = Url::parse(&format!("http://{address}")).expect("control endpoint should parse");
+    let response = backend
+        .send(Method::GET, backend.container_url().expect("container URL"), HeaderMap::new(), None)
+        .await
+        .expect("headers should arrive before testing the body deadline");
+    let err = tokio::time::timeout(Duration::from_secs(1), read_control_body(response, Duration::from_millis(20)))
+        .await
+        .expect("body deadline must be shorter than the transport idle timeout")
+        .expect_err("an unfinished control response must time out");
+    assert_eq!(err.kind(), ErrorKind::TimedOut);
+    let _ = release.send(());
+    server.await.expect("control fixture should join");
+}

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

@@ -351,7 +351,7 @@ fn transition_candidate_metadata(info: ObjectInfo) -> HashMap<String, String> {
     metadata
 }
 
-fn transition_candidate_metadata_matches(
+pub(super) fn transition_candidate_metadata_matches(
     metadata: &HashMap<String, String>,
     identity: TransitionCandidateIdentity,
 ) -> Result<bool, std::io::Error> {

crates/ecstore/src/store/init.rs

@@ -24035,6 +24035,7 @@ mod tests {
             without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-mutation-intent-record", &[4])).await;
         let mutation_id = uuid::Uuid::new_v4();
         let intent = TierMutationIntent {
+            requires_native_azure: false,
             mutation_id,
             revision: 1,
             kind: TierMutationIntentKind::Edit,
@@ -24220,6 +24221,7 @@ mod tests {
         let (_ctx, store, _shutdown) =
             without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-mutation-intent-scan", &[4])).await;
         let build_intent = |mutation_id: uuid::Uuid, tier_name: &str| TierMutationIntent {
+            requires_native_azure: false,
             mutation_id,
             revision: 1,
             kind: TierMutationIntentKind::Edit,
@@ -24281,6 +24283,7 @@ mod tests {
             without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-mutation-intent-advance", &[4])).await;
         let mutation_id = uuid::Uuid::new_v4();
         let intent = TierMutationIntent {
+            requires_native_azure: false,
             mutation_id,
             revision: 1,
             kind: TierMutationIntentKind::Edit,
@@ -24355,6 +24358,7 @@ mod tests {
 
         let abort_id = uuid::Uuid::new_v4();
         let abort_intent = TierMutationIntent {
+            requires_native_azure: false,
             mutation_id: abort_id,
             revision: 1,
             kind: TierMutationIntentKind::Edit,
@@ -24397,6 +24401,7 @@ mod tests {
         candidate_digest: [u8; 32],
     ) -> TierMutationIntent {
         TierMutationIntent {
+            requires_native_azure: false,
             mutation_id,
             revision: 1,
             kind: TierMutationIntentKind::Edit,
@@ -25473,6 +25478,7 @@ mod tests {
         let (_ctx, store, _shutdown) =
             without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-mutation-intent-scan-page", &[4])).await;
         let build_intent = |mutation_id: uuid::Uuid, tier_name: &str| TierMutationIntent {
+            requires_native_azure: false,
             mutation_id,
             revision: 1,
             kind: TierMutationIntentKind::Edit,

crates/signer/Cargo.toml

@@ -43,6 +43,7 @@ rustfs-utils = { workspace = true, features = ["full"] }
 s3s = { workspace = true, features = ["minio"] }
 base64-simd.workspace = true
 thiserror.workspace = true
+url.workspace = true
 
 [dev-dependencies]
 tracing-subscriber = { workspace = true, features = ["fmt"] }

crates/signer/src/lib.rs

@@ -13,12 +13,14 @@
 // limitations under the License.
 
 pub mod constants;
+pub mod request_signature_azure;
 pub mod request_signature_streaming;
 pub mod request_signature_streaming_unsigned_trailer;
 pub mod request_signature_v2;
 pub mod request_signature_v4;
 pub mod utils;
 
+pub use request_signature_azure::{AzureSignError, azure_shared_key_signature};
 pub use request_signature_streaming::streaming_sign_v4;
 pub use request_signature_streaming::try_streaming_sign_v4;
 pub use request_signature_v2::SignV2Error;

crates/signer/src/request_signature_azure.rs

@@ -0,0 +1,175 @@
+// 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.
+
+//! Shared Key signing for Azure Blob REST requests.
+
+use http::HeaderMap;
+use rustfs_utils::crypto::hmac_sha256;
+use std::collections::BTreeMap;
+use url::Url;
+
+#[derive(Debug, thiserror::Error)]
+pub enum AzureSignError {
+    #[error("Azure Shared Key signing requires a non-empty account key")]
+    EmptyKey,
+    #[error("Azure Shared Key header {0} is not valid ASCII")]
+    InvalidHeader(String),
+    #[error("Azure Shared Key header {0} has multiple values")]
+    DuplicateHeader(&'static str),
+}
+
+/// Sign an Azure Blob request using a decoded storage-account key.
+///
+/// The caller supplies the final URL and headers, including x-ms-date and
+/// x-ms-version. Content headers and conditions are signed for writes as well
+/// as reads. This implements the Shared Key format for service versions
+/// 2015-02-21 and newer, where a zero Content-Length is signed as an empty
+/// field. The returned base64 signature does not include the account name.
+///
+/// # Errors
+///
+/// Rejects an empty key, non-ASCII signed headers and repeated standard
+/// headers instead of signing a different request from the one on the wire.
+pub fn azure_shared_key_signature(
+    key: &[u8],
+    account: &str,
+    method: &str,
+    url: &Url,
+    headers: &HeaderMap,
+) -> Result<String, AzureSignError> {
+    if key.is_empty() {
+        return Err(AzureSignError::EmptyKey);
+    }
+    let mut canonical = String::with_capacity(512);
+    canonical.push_str(method);
+    canonical.push('\n');
+    for name in [
+        "content-encoding",
+        "content-language",
+        "content-length",
+        "content-md5",
+        "content-type",
+        "date",
+        "if-modified-since",
+        "if-match",
+        "if-none-match",
+        "if-unmodified-since",
+        "range",
+    ] {
+        let mut values = headers.get_all(name).iter();
+        let value = values.next();
+        if values.next().is_some() {
+            return Err(AzureSignError::DuplicateHeader(name));
+        }
+        let value = value
+            .map(|value| {
+                value
+                    .to_str()
+                    .map(str::trim)
+                    .map_err(|_| AzureSignError::InvalidHeader(name.to_string()))
+            })
+            .transpose()?
+            .unwrap_or_default();
+        if !(name == "content-length" && value == "0" || name == "date" && headers.contains_key("x-ms-date")) {
+            canonical.push_str(value);
+        }
+        canonical.push('\n');
+    }
+
+    let mut service_headers: BTreeMap<&str, Vec<String>> = BTreeMap::new();
+    for (name, value) in headers {
+        if name.as_str().starts_with("x-ms-") {
+            let value = value.to_str().map_err(|_| AzureSignError::InvalidHeader(name.to_string()))?;
+            service_headers
+                .entry(name.as_str())
+                .or_default()
+                .push(value.split_ascii_whitespace().collect::<Vec<_>>().join(" "));
+        }
+    }
+    for (name, values) in service_headers {
+        canonical.push_str(name);
+        canonical.push(':');
+        canonical.push_str(&values.join(","));
+        canonical.push('\n');
+    }
+    canonical.push('/');
+    canonical.push_str(account);
+    canonical.push_str(url.path());
+    let mut query: BTreeMap<String, Vec<String>> = BTreeMap::new();
+    for (name, value) in url.query_pairs() {
+        query.entry(name.to_ascii_lowercase()).or_default().push(value.into_owned());
+    }
+    for (name, mut values) in query {
+        values.sort();
+        canonical.push('\n');
+        canonical.push_str(&name);
+        canonical.push(':');
+        canonical.push_str(&values.join(","));
+    }
+    Ok(base64_simd::STANDARD.encode_to_string(hmac_sha256(key, canonical.as_bytes())))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use http::HeaderValue;
+
+    #[test]
+    fn signs_write_content_and_conditions_with_independent_hmac_vectors() {
+        let url = Url::parse("https://account.blob.core.windows.net/container/blob?comp=block&blockid=AAAA")
+            .expect("fixture URL should parse");
+        let mut headers = HeaderMap::new();
+        for (name, value) in [
+            ("content-length", "3"),
+            ("content-md5", "YWJj"),
+            ("content-type", "application/octet-stream"),
+            ("if-none-match", "*"),
+            ("x-ms-date", "Fri, 06 Nov 2026 01:00:00 GMT"),
+            ("x-ms-version", "2020-10-02"),
+        ] {
+            headers.insert(name, HeaderValue::from_static(value));
+        }
+        // Fixed vectors generated with Python's hmac/hashlib from the Azure
+        // documented canonical fields, independently of this implementation.
+        for (length, expected) in [
+            ("3", "TMvdha8PzCSf0SB3v9EQSv0Se1gpdu1HusHBamwkyws="),
+            ("0", "kxBwUHkI/MK9beMd85Mij0MvaAlBtQQVYo9Qvl9rams="),
+        ] {
+            headers.insert("content-length", HeaderValue::from_static(length));
+            let signature = azure_shared_key_signature(b"0123456789abcdef0123456789abcdef", "account", "PUT", &url, &headers)
+                .expect("valid write should sign");
+            assert_eq!(signature, expected);
+        }
+    }
+
+    #[test]
+    fn rejects_ambiguous_headers_without_exposing_their_values() {
+        let url = Url::parse("https://account.blob.core.windows.net/container/blob").expect("fixture URL should parse");
+        let mut headers = HeaderMap::new();
+        headers.append("content-length", HeaderValue::from_static("1"));
+        headers.append("content-length", HeaderValue::from_static("2"));
+        assert!(matches!(
+            azure_shared_key_signature(b"key", "account", "PUT", &url, &headers),
+            Err(AzureSignError::DuplicateHeader("content-length"))
+        ));
+        headers.clear();
+        headers.insert(
+            "x-ms-meta-value",
+            HeaderValue::from_bytes(b"\xff-private").expect("opaque header is legal HTTP"),
+        );
+        let err =
+            azure_shared_key_signature(b"key", "account", "PUT", &url, &headers).expect_err("opaque signed header must fail");
+        assert_eq!(err.to_string(), "Azure Shared Key header x-ms-meta-value is not valid ASCII");
+    }
+}

docs/architecture/ilm-tiering-persistence-contracts.md

@@ -187,6 +187,8 @@ The current background loop runs every 60 seconds, scans at most 1,000 records p
 
 `TierMutationIntent` models Add/Edit/Remove/Clear with `Prepared`, `Committed`, and `Aborted`. The identity includes the old tier-config ETag, candidate digest, canonical sorted affected targets and their old/new backend identities, and expiry. Revision and committed config ETag are the only mutable generation data.
 
+A candidate configuration containing a native Azure Blob tier adds the immutable `requires_native_azure: true` intent extension. The flag participates in intent identity and checksum validation. Older strict readers reject it before installing a peer Prepare block, so all-node Prepare gates publication on native-driver support. The flag is omitted for ordinary mutations, preserving their prior wire bytes and checksums. Native Azure's persisted type and upgrade restrictions are documented in [Azure Blob remote tiers](../operations/azure-blob-tiering.md#upgrade-and-rollback).
+
 New intents use a 15-minute expiry. A peer-only terminal tombstone is retained until that expiry plus five minutes of clock-skew allowance and until no coordinator record remains. Expiry bounds replay protection; it is not config commit/abort evidence.
 
 The coordinator creates its durable record and peer `Prepare` blocks new reference creation, drains exact tier-operation leases, and proves that edit/remove/clear will not strand authoritative references. Prepare, Commit, and Abort use all-node fanout rather than quorum: independent peer calls use a work-conserving concurrency limit of four, a 30-second per-peer deadline, and a 30-second fanout-wide deadline; Prepare is additionally capped by the intent expiry. The coordinator collects every completed outcome. A timed-out or otherwise ambiguous started Prepare is included in compensating Abort because cancellation does not prove the peer failed to persist its fence; peers not started before the fanout deadline make Prepare fail but do not require Abort. The coordinator then conditionally writes tier config, durably commits the coordinator intent, releases its exclusive guards, requires every prepared peer to commit, publishes the runtime candidate, and clears the block. Per-mutation sharded mutexes serialize local phases only; persisted intent plus tier-config ETag is authoritative.

docs/operations/azure-blob-tiering.md

@@ -0,0 +1,66 @@
+# Azure Blob Remote Tiers
+
+New Azure tiers use the native Azure Blob REST API with storage-account Shared Key authentication. Azure public cloud and Azure Stack Hub use the same driver; Stack Hub requires its own account Blob endpoint. The driver uses service version 2020-10-02, which Microsoft lists for Stack Hub 2301 and later. See [Microsoft's storage compatibility matrix](https://learn.microsoft.com/en-us/azure-stack/user/azure-stack-acs-differences).
+
+The implementation is in [warm_backend_azure_blob.rs](../../crates/ecstore/src/services/tier/warm_backend_azure_blob.rs). Account-key signing is shared with the on-demand migration reader through [request_signature_azure.rs](../../crates/signer/src/request_signature_azure.rs); the two features have different storage-operation and service-version contracts.
+
+## Configuration
+
+Create the destination container first. Reserve its selected prefix for RustFS tier data, and do not modify the generated blobs with another writer.
+
+| Console/API field | Azure meaning |
+|---|---|
+| Azure Blob | Select the Azure tier provider |
+| Access Key / accessKey | Storage account name |
+| Secret Key / secretKey | Base64 storage-account key |
+| Bucket / bucket | Existing container name |
+| Endpoint / endpoint | Account Blob service origin; an empty value derives the public-cloud endpoint |
+| Prefix / prefix | Optional prefix reserved for this tier |
+| Region / region | Not used by native Blob signing or routing |
+| Storage Class / storageClass | Leave empty; the Console placeholder STANDARD is normalized to the default |
+
+For Stack Hub, use the complete account endpoint supplied by its portal, such as https://account.blob.region.example. Do not supply a container path or a SAS query string as part of the endpoint. HTTPS uses the server's trusted CA roots, including the deployment's installed private CA bundle.
+
+The existing Add Tier API accepts this shape:
+
+~~~json
+{
+  "type": "azure",
+  "azure": {
+    "name": "AZURE_COLD",
+    "endpoint": "https://account.blob.region.example",
+    "accessKey": "account",
+    "secretKey": "<base64-account-key>",
+    "bucket": "archive",
+    "prefix": "rustfs"
+  }
+}
+~~~
+
+Only storage-account keys are supported by this tier driver. Service-principal credentials, SAS tokens and explicit Hot/Cool/Archive access-tier selection are not enabled. Stack Hub's own support matrix also differs from public Azure for authentication and access tiers; selecting Azure Blob does not promise every Azure feature.
+
+## Upload, Read, and Cleanup
+
+The driver writes Block Blobs. Small objects use Put Blob; larger objects use bounded Put Block requests followed by one conditional Put Block List. Each block has Content-MD5 and a unique upload-specific ID. The complete declared input must be consumed before commit, and the driver does not automatically replay failed requests. The Stack Hub limit of 50,000 blocks with at most 100 MiB per block bounds the accepted object size.
+
+RustFS retains an opaque Azure version ID when the service returns one. Reads and deletes of that object explicitly address that version. An empty remote version sends no versionid parameter. Unversioned cleanup obtains an ETag and uses If-Match; a versioning change or a failed condition is an error, not permission to delete a replacement. Snapshots and immutability policies may prevent deletion and leave the normal tier cleanup record pending.
+
+Metadata is encoded into legal Azure metadata names and values. The transaction and destination identities retain both RustFS and MinIO aliases. Recovery checks those identities before recognizing an uncertain upload as its candidate. Encoded metadata is limited to 8 KiB. The complete destination blob name, including the configured prefix, must fit the Stack Hub limit of 880 characters and 1,760 UTF-8 bytes; standalone dot path segments are rejected before I/O.
+
+Interrupted block staging can leave uncommitted blocks that Azure expires according to its own policy. Lifecycle uploads retain the existing transaction-journal recovery contract for a lost commit acknowledgement. Add/Edit/Verify probes retain the existing bounded in-process compensation: killing the process during a validation probe is not covered by a durable probe recovery worker. See [validation probe recovery status](tier-ilm-debugging.md#validation-probe-crash-recovery-status).
+
+## Upgrade and Rollback
+
+Existing Azure configurations retain their legacy S3-compatible transport. New Add operations select native Blob and persist a distinct protocol version and an encoded endpoint marker inside the existing binary configuration shape. A legacy destination and a native destination have different backend identities even if their account, endpoint, container and prefix are identical. Credential edits do not change the stored protocol.
+
+Upgrade every topology member before adding a native Azure tier. A mutation whose candidate configuration contains a native Azure tier carries the immutable requires_native_azure intent extension. Earlier peers reject that extension during Prepare, before the new configuration can be committed. Ordinary tier mutations without native Azure keep their previous wire shape and checksums.
+
+A rejected attempt during a mixed-version upgrade may retain a coordinator recovery record until every peer can read the intent and complete its normal abort/recovery exchange. Finish upgrading the remaining nodes before retrying; do not manually remove mutation records.
+
+Do not downgrade a cluster while a native Azure tier or its recovery records remain in use. An older server can preserve and rewrite the existing Azure payload, but its S3 driver rejects the encoded native endpoint before making a request. Restore or migrate the referenced data and complete normal tier removal and recovery before rollback. Do not manually remove the native endpoint marker or protocol version, or delete tier metadata to bypass reference checks.
+
+## Diagnostics and Verification
+
+Native service failures retain a bounded HTTP status, Azure service error code and x-ms-request-id. Credential failures and deterministic rejected writes are reported directly; they do not cause a cleanup attempt against an object the write did not create. A transport failure or lost acknowledgement still requires candidate reconciliation, and an unconfirmed cleanup remains an error. Credentials, signed authorization values and raw response bodies are excluded from these diagnostics.
+
+Local regression tests cover the signed Blob wire requests, block commit boundaries, empty and truncated input, ranged and versioned reads, conditional cleanup, identity-bound recovery, service errors, bounded XML parsing, persisted protocol selection and rejection by the previous peer reader. Before certifying a Stack Hub release, run Add/Verify, lifecycle transition, read-back/restore, expiry and restart/failure tests against that actual Stack Hub version. Mock protocol tests and Azure public-cloud results alone do not establish Stack Hub interoperability.

docs/operations/outbound-connection-policy.md

@@ -22,7 +22,7 @@ RustFS validates every operator-configured outbound destination to close a serve
 | On-demand migration sources | Literal check, relaxed | Same guard and same escape hatch as replication targets (see below) |
 | Site replication peers | Literal check | `rustfs/src/site_replication/mod.rs` |
 | Self-hosted tiering warm backends (S3, MinIO, RustFS) | Literal check, relaxed | `validate_outbound_url_allow_private` allows private storage addresses except known IPv6 metadata addresses; RustFS retains its own debug-only loopback exception |
-| Cloud tiering warm backends (Azure, GCS, Aliyun, Tencent, Huawei, R2) | Literal check | `crates/ecstore/src/services/tier/warm_backend.rs` `validate_endpoint`; Wasabi additionally requires its canonical provider endpoint before using the S3 constructor |
+| Cloud tiering warm backends (Azure, GCS, Aliyun, Tencent, Huawei, R2) | Literal check | `crates/ecstore/src/services/tier/warm_backend.rs` `validate_endpoint`, or the direct literal check in `warm_backend_azure_blob.rs` for native Azure; Wasabi additionally requires its canonical provider endpoint before using the S3 constructor |
 | Keystone `auth_url` | Literal check | `crates/keystone/src/config.rs` |
 
 The allowlist affects only the "Full policy" rows. A literal-check subsystem rejects a hostname that is itself a restricted IP literal, does not re-check what a hostname resolves to, and cannot be widened by `RUSTFS_OUTBOUND_ALLOW_ORIGINS`.

docs/operations/tier-ilm-debugging.md

@@ -24,6 +24,8 @@
 
 ## Lifecycle rule limits and evaluation
 
+For native Azure configuration, protocol compatibility and the supported Stack Hub baseline, see [Azure Blob remote tiers](azure-blob-tiering.md).
+
 Each lifecycle rule supports at most one `Transition` and one `NoncurrentVersionTransition`. A version can make one initial transition; chaining additional tiers after it reaches `complete` is not supported. Splitting stages across overlapping rules does not enable a transition chain. `PutBucketLifecycleConfiguration` rejects multiple entries in either transition array with `InvalidArgument`, including in disabled rules. Existing stored multi-entry arrays are not executed; replace each with a single intended destination. Independent expiration actions in the rule remain eligible.
 
 `Expiration.Days` and `Expiration.Date` are mutually exclusive. A request containing both is rejected instead of silently selecting the date. When expiration and transition are both eligible, expiration takes precedence; a failed earlier transition does not keep an expired object indefinitely. Deadlines select the earliest action within the same action class.

rustfs/src/on_demand_migration/azure.rs

@@ -42,15 +42,18 @@ use super::source_client::{
 };
 use super::storage_api::HTTPRangeSpec;
 use super::storage_api::remote_s3_client::RemoteS3ClientError;
+#[cfg(test)]
 use hmac::{Hmac, Mac, digest::KeyInit};
 use http::{HeaderMap, HeaderValue, Method};
 use percent_encoding::percent_decode_str;
 use quick_xml::Reader;
 use quick_xml::events::{BytesStart, Event};
+#[cfg(test)]
 use sha2::Sha256;
-use std::collections::{BTreeMap, HashMap};
+use std::collections::HashMap;
 use url::Url;
 
+#[cfg(test)]
 type HmacSha256 = Hmac<Sha256>;
 
 /// Blob REST version this backend pins. Every response field it reads exists
@@ -301,61 +304,8 @@ impl SourceBackend for AzureSourceBackend {
 /// canonicalized resource. GET and HEAD carry no body, so every `Content-*`
 /// slot stays empty.
 fn shared_key_signature(key: &[u8], account: &str, method: &str, url: &Url, headers: &HeaderMap) -> Result<String, SourceError> {
-    let mut string_to_sign = String::with_capacity(256);
-    string_to_sign.push_str(method);
-    string_to_sign.push('\n');
-    // Content-Encoding, Content-Language, Content-Length, Content-MD5,
-    // Content-Type, Date, If-Modified-Since, If-Match, If-None-Match,
-    // If-Unmodified-Since: all empty. `Date` stays empty because `x-ms-date`
-    // carries the timestamp and Azure then ignores this slot.
-    for _ in 0..10 {
-        string_to_sign.push('\n');
-    }
-    string_to_sign.push_str(header(headers, "range").unwrap_or_default());
-    string_to_sign.push('\n');
-
-    // Canonicalized headers: every `x-ms-*` header, lowercased and sorted.
-    let mut canonical_headers = BTreeMap::new();
-    for (name, value) in headers {
-        let name = name.as_str();
-        if let Some(rest) = name.strip_prefix("x-ms-")
-            && !rest.is_empty()
-            && let Ok(value) = value.to_str()
-        {
-            canonical_headers.insert(name.to_string(), value.trim().to_string());
-        }
-    }
-    for (name, value) in &canonical_headers {
-        string_to_sign.push_str(name);
-        string_to_sign.push(':');
-        string_to_sign.push_str(value);
-        string_to_sign.push('\n');
-    }
-
-    // Canonicalized resource: the account, the encoded path, then every query
-    // parameter lowercased and sorted, with repeated values joined by commas.
-    string_to_sign.push('/');
-    string_to_sign.push_str(account);
-    string_to_sign.push_str(url.path());
-    let mut canonical_query: BTreeMap<String, Vec<String>> = BTreeMap::new();
-    for (name, value) in url.query_pairs() {
-        canonical_query
-            .entry(name.to_ascii_lowercase())
-            .or_default()
-            .push(value.into_owned());
-    }
-    for (name, mut values) in canonical_query {
-        values.sort();
-        string_to_sign.push('\n');
-        string_to_sign.push_str(&name);
-        string_to_sign.push(':');
-        string_to_sign.push_str(&values.join(","));
-    }
-
-    let mut mac = HmacSha256::new_from_slice(key)
-        .map_err(|_| SourceError::Other("azure account key has an unusable length".to_string()))?;
-    mac.update(string_to_sign.as_bytes());
-    Ok(base64_simd::STANDARD.encode_to_string(mac.finalize().into_bytes()))
+    rustfs_signer::azure_shared_key_signature(key, account, method, url, headers)
+        .map_err(|err| SourceError::Other(err.to_string()))
 }
 
 #[derive(Debug)]