fix(ilm): harden signer failures and guard remote tier delete storms (#2706)

crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs

@@ -20,15 +20,144 @@
 
 use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryOp, GLOBAL_ExpiryState, TransitionedObject};
 use crate::bucket::lifecycle::lifecycle::{self, ObjectOpts};
+use crate::client::signer_error::error_chain_contains_signer_header_marker;
 use crate::global::GLOBAL_TierConfigMgr;
+use rustfs_utils::get_env_usize;
 use sha2::{Digest, Sha256};
 use std::any::Any;
+use std::collections::VecDeque;
 use std::io::Write;
+use std::sync::LazyLock;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::{Duration, Instant};
+use tokio::sync::{Mutex, Semaphore};
+use tracing::warn;
 use uuid::Uuid;
 use xxhash_rust::xxh64;
 
 static XXHASH_SEED: u64 = 0;
 
+const ENV_REMOTE_DELETE_MAX_CONCURRENCY: &str = "RUSTFS_REMOTE_DELETE_MAX_CONCURRENCY";
+const ENV_REMOTE_DELETE_BREAKER_THRESHOLD: &str = "RUSTFS_REMOTE_DELETE_BREAKER_THRESHOLD";
+const ENV_REMOTE_DELETE_BREAKER_WINDOW_SECS: &str = "RUSTFS_REMOTE_DELETE_BREAKER_WINDOW_SECS";
+const DEFAULT_REMOTE_DELETE_BREAKER_THRESHOLD: usize = 50;
+const DEFAULT_REMOTE_DELETE_BREAKER_WINDOW_SECS: usize = 30;
+const METRIC_DELETE_REMOTE_FAILED_TOTAL: &str = "rustfs_delete_remote_failed_total";
+const METRIC_DELETE_REMOTE_BREAKER_TOTAL: &str = "rustfs_delete_remote_breaker_total";
+const METRIC_DELETE_REMOTE_INFLIGHT: &str = "rustfs_delete_remote_inflight";
+
+static REMOTE_DELETE_INFLIGHT: AtomicUsize = AtomicUsize::new(0);
+
+static REMOTE_DELETE_LIMITER: LazyLock<Semaphore> = LazyLock::new(|| {
+    let default_limit = std::cmp::min(num_cpus::get(), 16).max(1);
+    let concurrency = get_env_usize(ENV_REMOTE_DELETE_MAX_CONCURRENCY, default_limit).max(1);
+    Semaphore::new(concurrency)
+});
+
+static REMOTE_DELETE_BREAKER: LazyLock<Mutex<RemoteDeleteBreaker>> = LazyLock::new(|| {
+    Mutex::new(RemoteDeleteBreaker::new(
+        get_env_usize(ENV_REMOTE_DELETE_BREAKER_THRESHOLD, DEFAULT_REMOTE_DELETE_BREAKER_THRESHOLD).max(1),
+        Duration::from_secs(
+            get_env_usize(ENV_REMOTE_DELETE_BREAKER_WINDOW_SECS, DEFAULT_REMOTE_DELETE_BREAKER_WINDOW_SECS) as u64,
+        ),
+    ))
+});
+
+#[derive(Debug)]
+struct RemoteDeleteBreaker {
+    threshold: usize,
+    window: Duration,
+    failures: VecDeque<Instant>,
+}
+
+impl RemoteDeleteBreaker {
+    fn new(threshold: usize, window: Duration) -> Self {
+        Self {
+            threshold: threshold.max(1),
+            window: window.max(Duration::from_secs(1)),
+            failures: VecDeque::new(),
+        }
+    }
+
+    fn should_short_circuit(&mut self, now: Instant) -> bool {
+        self.prune(now);
+        self.failures.len() >= self.threshold
+    }
+
+    fn record_signer_failure(&mut self, now: Instant) -> bool {
+        self.prune(now);
+        let was_open = self.failures.len() >= self.threshold;
+        self.failures.push_back(now);
+        !was_open && self.failures.len() >= self.threshold
+    }
+
+    fn prune(&mut self, now: Instant) {
+        while let Some(ts) = self.failures.front().copied() {
+            if now.duration_since(ts) > self.window {
+                self.failures.pop_front();
+            } else {
+                break;
+            }
+        }
+    }
+}
+
+struct RemoteDeleteInflightGuard;
+
+impl RemoteDeleteInflightGuard {
+    fn new() -> Self {
+        let inflight = REMOTE_DELETE_INFLIGHT.fetch_add(1, Ordering::Relaxed) + 1;
+        metrics::gauge!(METRIC_DELETE_REMOTE_INFLIGHT).set(inflight as f64);
+        Self
+    }
+}
+
+impl Drop for RemoteDeleteInflightGuard {
+    fn drop(&mut self) {
+        let inflight = REMOTE_DELETE_INFLIGHT.fetch_sub(1, Ordering::Relaxed) - 1;
+        metrics::gauge!(METRIC_DELETE_REMOTE_INFLIGHT).set(inflight as f64);
+    }
+}
+
+fn is_signer_header_error(err: &std::io::Error) -> bool {
+    if err.kind() != std::io::ErrorKind::InvalidInput {
+        return false;
+    }
+
+    if let Some(source) = err.get_ref() {
+        if error_chain_contains_signer_header_marker(source) {
+            return true;
+        }
+    }
+
+    let message = err.to_string().to_ascii_lowercase();
+    message.contains("invalid utf-8 header value")
+        || message.contains("invalidheadervalue")
+        || (message.contains("sign v4") && message.contains("header value"))
+}
+
+async fn remote_delete_breaker_is_open(now: Instant) -> bool {
+    let mut breaker = REMOTE_DELETE_BREAKER.lock().await;
+    breaker.should_short_circuit(now)
+}
+
+async fn record_remote_delete_failure(err: &std::io::Error, now: Instant) {
+    metrics::counter!(METRIC_DELETE_REMOTE_FAILED_TOTAL).increment(1);
+
+    if !is_signer_header_error(err) {
+        return;
+    }
+
+    let mut breaker = REMOTE_DELETE_BREAKER.lock().await;
+    if breaker.record_signer_failure(now) {
+        warn!(
+            threshold = breaker.threshold,
+            window_secs = breaker.window.as_secs(),
+            "remote tier delete breaker opened by signer/header failures"
+        );
+    }
+}
+
 #[derive(Default)]
 #[allow(dead_code)]
 struct ObjSweeper {
@@ -148,12 +277,31 @@ impl ExpiryOp for Jentry {
 }
 
 pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
+    if remote_delete_breaker_is_open(Instant::now()).await {
+        metrics::counter!(METRIC_DELETE_REMOTE_BREAKER_TOTAL).increment(1);
+        return Err(std::io::Error::other("remote tier delete breaker is open due to signer/header failures"));
+    }
+
+    let _permit = REMOTE_DELETE_LIMITER
+        .acquire()
+        .await
+        .map_err(|_| std::io::Error::other("remote tier delete limiter is closed"))?;
+    let _inflight = RemoteDeleteInflightGuard::new();
+
     let mut config_mgr = GLOBAL_TierConfigMgr.write().await;
     let w = match config_mgr.get_driver(tier_name).await {
         Ok(w) => w,
-        Err(e) => return Err(std::io::Error::other(e)),
+        Err(e) => {
+            let err = std::io::Error::other(e);
+            record_remote_delete_failure(&err, Instant::now()).await;
+            return Err(err);
+        }
     };
-    w.remove(obj_name, rv_id).await
+    let result = w.remove(obj_name, rv_id).await;
+    if let Err(err) = &result {
+        record_remote_delete_failure(err, Instant::now()).await;
+    }
+    result
 }
 
 pub fn transitioned_delete_journal_entry(
@@ -189,4 +337,44 @@ pub fn transitioned_force_delete_journal_entry(transitioned: &TransitionedObject
 }
 
 #[cfg(test)]
-mod test {}
+mod test {
+    use crate::client::signer_error::invalid_utf8_header_error;
+
+    use super::{RemoteDeleteBreaker, is_signer_header_error};
+    use std::io::{Error, ErrorKind};
+    use std::time::{Duration, Instant};
+
+    #[test]
+    fn signer_header_error_detection_matches_utf8_failures() {
+        let err = Error::new(
+            ErrorKind::InvalidInput,
+            "failed to sign v4 request: invalid UTF-8 header value for `x-amz-meta-invalid`",
+        );
+        assert!(is_signer_header_error(&err));
+    }
+
+    #[test]
+    fn signer_header_error_detection_rejects_unrelated_errors() {
+        let err = Error::other("dial tcp: i/o timeout");
+        assert!(!is_signer_header_error(&err));
+    }
+
+    #[test]
+    fn signer_header_error_detection_matches_structured_marker() {
+        let err = invalid_utf8_header_error("failed to sign v4 request", "x-amz-meta-invalid");
+        assert!(is_signer_header_error(&err));
+    }
+
+    #[test]
+    fn breaker_opens_at_threshold_and_recovers_after_window() {
+        let mut breaker = RemoteDeleteBreaker::new(3, Duration::from_secs(30));
+        let start = Instant::now();
+
+        assert!(!breaker.should_short_circuit(start));
+        assert!(!breaker.record_signer_failure(start));
+        assert!(!breaker.record_signer_failure(start + Duration::from_secs(1)));
+        assert!(breaker.record_signer_failure(start + Duration::from_secs(2)));
+        assert!(breaker.should_short_circuit(start + Duration::from_secs(3)));
+        assert!(!breaker.should_short_circuit(start + Duration::from_secs(40)));
+    }
+}

crates/ecstore/src/client/bucket_cache.rs

@@ -22,6 +22,7 @@ use super::constants::UNSIGNED_PAYLOAD;
 use super::credentials::SignatureType;
 use crate::client::{
     api_error_response::http_resp_to_error_response,
+    signer_error,
     transition_api::{CreateBucketConfiguration, LocationConstraint, TransitionClient},
 };
 use http::Request;
@@ -35,6 +36,10 @@ use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
 use s3s::S3ErrorCode;
 use std::collections::HashMap;
 
+fn signer_error_to_io_error(scope: &str, error: rustfs_signer::SignV4Error) -> std::io::Error {
+    signer_error::signer_error_to_io_error(scope, error)
+}
+
 #[derive(Debug, Clone)]
 pub struct BucketLocationCache {
     items: HashMap<String, String>,
@@ -179,10 +184,15 @@ impl TransitionClient {
             content_sha256 = UNSIGNED_PAYLOAD.to_string();
         }
 
-        if let Ok(content_sha256_value) = content_sha256.parse() {
-            req.headers_mut().insert("X-Amz-Content-Sha256", content_sha256_value);
-        }
-        let req = rustfs_signer::sign_v4(req, 0, &access_key_id, &secret_access_key, &session_token, "us-east-1");
+        let content_sha256_value = content_sha256.parse().map_err(|err| {
+            std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                format!("invalid X-Amz-Content-Sha256 header value: {err}"),
+            )
+        })?;
+        req.headers_mut().insert("X-Amz-Content-Sha256", content_sha256_value);
+        let req = rustfs_signer::try_sign_v4(req, 0, &access_key_id, &secret_access_key, &session_token, "us-east-1")
+            .map_err(|err| signer_error_to_io_error("failed to sign bucket location request", err))?;
         Ok(req)
     }
 }

crates/ecstore/src/client/mod.rs

@@ -35,5 +35,6 @@ pub mod constants;
 pub mod credentials;
 pub mod object_api_utils;
 pub mod object_handlers_common;
+pub mod signer_error;
 pub mod transition_api;
 pub mod utils;

crates/ecstore/src/client/signer_error.rs

@@ -0,0 +1,74 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::error::Error as StdError;
+use std::fmt::{Display, Formatter};
+use std::io::{Error, ErrorKind};
+
+pub(crate) const SIGNER_HEADER_ERROR_MARKER: &str = "rustfs_signer_header_error";
+
+#[derive(Debug)]
+struct SignerHeaderError {
+    scope: String,
+    header_name: String,
+}
+
+impl SignerHeaderError {
+    fn new(scope: &str, header_name: &str) -> Self {
+        Self {
+            scope: scope.to_string(),
+            header_name: header_name.to_string(),
+        }
+    }
+}
+
+impl Display for SignerHeaderError {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{}: invalid UTF-8 header value for `{}` [{}]",
+            self.scope, self.header_name, SIGNER_HEADER_ERROR_MARKER
+        )
+    }
+}
+
+impl StdError for SignerHeaderError {}
+
+pub(crate) fn invalid_utf8_header_error(scope: &str, header_name: &str) -> Error {
+    Error::new(ErrorKind::InvalidInput, SignerHeaderError::new(scope, header_name))
+}
+
+pub(crate) fn signer_error_to_io_error(scope: &str, error: rustfs_signer::SignV4Error) -> Error {
+    match error {
+        rustfs_signer::SignV4Error::InvalidHeaderValue { name } => invalid_utf8_header_error(scope, &name),
+        other => Error::other(format!("{scope}: {other}")),
+    }
+}
+
+pub(crate) fn error_chain_contains_signer_header_marker(err: &(dyn StdError + 'static)) -> bool {
+    let mut current = Some(err);
+    while let Some(source) = current {
+        if source.downcast_ref::<SignerHeaderError>().is_some() {
+            return true;
+        }
+
+        if source.to_string().contains(SIGNER_HEADER_ERROR_MARKER) {
+            return true;
+        }
+
+        current = source.source();
+    }
+
+    false
+}

crates/ecstore/src/client/transition_api.rs

@@ -31,6 +31,7 @@ use crate::client::{
     },
     constants::{UNSIGNED_PAYLOAD, UNSIGNED_PAYLOAD_TRAILER},
     credentials::{CredContext, Credentials, SignatureType, Static},
+    signer_error,
 };
 use crate::{client::checksum::ChecksumMode, store_api::GetObjectReader};
 use futures::{Future, StreamExt};
@@ -85,6 +86,21 @@ const C_UNKNOWN: i32 = -1;
 const C_OFFLINE: i32 = 0;
 const C_ONLINE: i32 = 1;
 
+fn invalid_utf8_header_error(scope: &str, header_name: &str) -> std::io::Error {
+    signer_error::invalid_utf8_header_error(scope, header_name)
+}
+
+fn validate_header_values(headers: &HeaderMap, scope: &str) -> Result<(), std::io::Error> {
+    for (name, value) in headers {
+        value.to_str().map_err(|_| invalid_utf8_header_error(scope, name.as_str()))?;
+    }
+    Ok(())
+}
+
+fn signer_error_to_io_error(scope: &str, error: rustfs_signer::SignV4Error) -> std::io::Error {
+    signer_error::signer_error_to_io_error(scope, error)
+}
+
 //pub type ReaderImpl = Box<dyn Reader + Send + Sync + 'static>;
 pub enum ReaderImpl {
     Body(Bytes),
@@ -560,8 +576,9 @@ impl TransitionClient {
                         "extra signed headers for presign with signature v2 is not supported.",
                     )));
                 }
-                let headers = req.headers_mut();
                 if let Some(extra_headers) = metadata.extra_pre_sign_header.as_ref() {
+                    validate_header_values(extra_headers, "presign extra header")?;
+                    let headers = req.headers_mut();
                     for (k, v) in extra_headers {
                         headers.insert(k, v.clone());
                     }
@@ -570,7 +587,7 @@ impl TransitionClient {
             if signer_type == SignatureType::SignatureV2 {
                 req = rustfs_signer::pre_sign_v2(req, &access_key_id, &secret_access_key, metadata.expires, is_virtual_host);
             } else if signer_type == SignatureType::SignatureV4 {
-                req = rustfs_signer::pre_sign_v4(
+                req = rustfs_signer::try_pre_sign_v4(
                     req,
                     &access_key_id,
                     &secret_access_key,
@@ -578,12 +595,14 @@ impl TransitionClient {
                     &location,
                     metadata.expires,
                     OffsetDateTime::now_utc(),
-                );
+                )
+                .map_err(|err| signer_error_to_io_error("failed to presign v4 request", err))?;
             }
             return Ok(req);
         }
 
         self.set_user_agent(&mut req);
+        validate_header_values(&metadata.custom_header, "request custom header")?;
 
         for (k, v) in metadata.custom_header.clone() {
             if let Some(key) = k {
@@ -593,15 +612,15 @@ impl TransitionClient {
 
         //req.content_length = metadata.content_length;
         if metadata.content_length <= -1 {
-            if let Ok(chunked_value) = HeaderValue::from_str(&vec!["chunked"].join(",")) {
-                req.headers_mut().insert(http::header::TRANSFER_ENCODING, chunked_value);
-            }
+            req.headers_mut()
+                .insert(http::header::TRANSFER_ENCODING, HeaderValue::from_static("chunked"));
         }
 
-        if metadata.content_md5_base64.len() > 0 {
-            if let Ok(md5_value) = HeaderValue::from_str(&metadata.content_md5_base64) {
-                req.headers_mut().insert("Content-Md5", md5_value);
-            }
+        if !metadata.content_md5_base64.is_empty() {
+            let md5_value = HeaderValue::from_str(&metadata.content_md5_base64).map_err(|err| {
+                std::io::Error::new(std::io::ErrorKind::InvalidInput, format!("invalid Content-Md5 header value: {err}"))
+            })?;
+            req.headers_mut().insert("Content-Md5", md5_value);
         }
 
         if signer_type == SignatureType::SignatureAnonymous {
@@ -634,14 +653,15 @@ impl TransitionClient {
                 .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?;
             req.headers_mut().insert(header_name, header_value);
 
-            req = rustfs_signer::sign_v4_trailer(
+            req = rustfs_signer::try_sign_v4_trailer(
                 req,
                 &access_key_id,
                 &secret_access_key,
                 &session_token,
                 &location,
                 metadata.trailer.clone(),
-            );
+            )
+            .map_err(|err| signer_error_to_io_error("failed to sign v4 request", err))?;
         }
 
         if metadata.content_length > 0 {
@@ -1354,7 +1374,10 @@ pub struct CreateBucketConfiguration {
 
 #[cfg(test)]
 mod tests {
-    use super::{build_tls_config, load_root_store_from_tls_path, with_rustls_init_guard};
+    use super::{
+        build_tls_config, load_root_store_from_tls_path, signer_error_to_io_error, validate_header_values, with_rustls_init_guard,
+    };
+    use http::{HeaderMap, HeaderValue};
 
     #[test]
     fn rustls_guard_converts_panics_to_io_errors() {
@@ -1404,4 +1427,29 @@ mod tests {
         });
         assert!(outcome.is_ok(), "provider install guard must not panic when a provider is already set");
     }
+
+    #[test]
+    fn validate_header_values_returns_header_name_for_non_utf8_values() {
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            "x-amz-meta-invalid",
+            HeaderValue::from_bytes(&[0xFF]).expect("invalid utf8 bytes should be accepted by HeaderValue"),
+        );
+
+        let err =
+            validate_header_values(&headers, "request custom header").expect_err("invalid header value should fail validation");
+        assert!(err.to_string().contains("x-amz-meta-invalid"));
+    }
+
+    #[test]
+    fn signer_error_mapping_preserves_header_name() {
+        let err = signer_error_to_io_error(
+            "failed to sign v4 request",
+            rustfs_signer::SignV4Error::InvalidHeaderValue {
+                name: "x-amz-meta-invalid".to_string(),
+            },
+        );
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+        assert!(err.to_string().contains("x-amz-meta-invalid"));
+    }
 }