test(s3): cover delete marker list visibility

* feat(kms): record operation metrics in the retry policy engine

Instrument policy::execute — the single choke point every outbound Vault
call and credential exchange already flows through — so no call site
needs its own instrumentation:

- rustfs_kms_backend_operations_total (counter): operation, op_class,
  outcome (success / fatal / budget_exhausted / deadline_exceeded /
  cancelled)
- rustfs_kms_backend_attempt_failures_total (counter): operation,
  error_class (retryable_conn / retryable_status / fatal /
  attempt_timeout)
- rustfs_kms_backend_operation_duration_seconds (histogram): wall-clock
  duration including retries and backoff
- rustfs_kms_backend_operation_attempts (histogram): attempts used

Metric labels carry only static enum values (operation names, classes,
outcomes) — never key identifiers, key material, ciphertext, or tokens.
Emission goes through the process-global metrics facade recorder, the
same pattern the rest of the workspace uses, so no new wiring is needed
in rustfs/src.

Tests drive a paused-clock runtime under a thread-local debugging
recorder, so counts, attempts, and even the recorded (virtual-clock)
durations are asserted deterministically with zero real sleeps.

Refs rustfs/backlog#1569 (part of rustfs/backlog#1562)

* test(kms): add Vault fault-injection matrix

Offline cases inject transport faults locally and are fully
deterministic: a refused connection is retried up to the configured
budget, and a stalled connection is cut off by the per-attempt timeout
instead of hanging. Ignored cases run against a real dev Vault
(RUSTFS_KMS_VAULT_ADDR) and pin the fail-closed auth behavior: an
invalid token and a missing key each resolve in exactly one attempt.

Every case asserts through the policy metrics recorded by a
thread-local debugging recorder, which doubles as the request-count
assertion even against a real server. Throttling and recoverable 5xx
responses cannot be forced on a stock dev Vault; those paths stay
pinned by the scripted-Vault wiring tests and the engine tests.

Refs rustfs/backlog#1569 (part of rustfs/backlog#1562)

Cargo.lock

@@ -9502,6 +9502,8 @@ dependencies = [
  "insta",
  "jiff",
  "md-5 0.11.0",
+ "metrics",
+ "metrics-util",
  "moka",
  "rand 0.10.2",
  "reqwest",

crates/e2e_test/src/delete_marker_migration_semantics_test.rs

@@ -56,6 +56,21 @@ mod tests {
         );
     }
 
+    async fn assert_current_list_hides_delete_marker(client: &Client, bucket: &str, key: &str) {
+        let listed = client
+            .list_objects_v2()
+            .bucket(bucket)
+            .prefix(key)
+            .send()
+            .await
+            .expect("list current objects after delete marker");
+
+        assert!(
+            listed.contents().iter().all(|object| object.key() != Some(key)),
+            "ListObjectsV2 must hide an object whose latest version is a delete marker"
+        );
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_versioning_only_delete_marker_has_minio_compatible_visibility_for_migration_proof() {
@@ -94,6 +109,7 @@ mod tests {
         assert_eq!(markers[0].version_id(), Some(delete_marker_version_id));
         assert_eq!(markers[0].is_latest(), Some(true));
         assert_current_get_is_delete_marker_not_found(&client, bucket, key).await;
+        assert_current_list_hides_delete_marker(&client, bucket, key).await;
     }
 
     #[tokio::test]
@@ -118,6 +134,17 @@ mod tests {
             .await
             .expect("put historical version");
         let data_version_id = put.version_id().expect("put should return data version id");
+        let listed_before_delete = client
+            .list_objects_v2()
+            .bucket(bucket)
+            .prefix(key)
+            .send()
+            .await
+            .expect("list current object before creating delete marker");
+        assert!(
+            listed_before_delete.contents().iter().any(|object| object.key() == Some(key)),
+            "ListObjectsV2 must include the current object before it is deleted"
+        );
 
         let delete_marker = client
             .delete_object()
@@ -145,6 +172,7 @@ mod tests {
         assert_eq!(markers[0].version_id(), Some(delete_marker_version_id));
         assert_eq!(markers[0].is_latest(), Some(true));
         assert_current_get_is_delete_marker_not_found(&client, bucket, key).await;
+        assert_current_list_hides_delete_marker(&client, bucket, key).await;
 
         let historical = client
             .get_object()

crates/kms/Cargo.toml

@@ -37,6 +37,8 @@ serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true, features = ["raw_value"] }
 tracing = { workspace = true }
 thiserror = { workspace = true }
+# Operation metrics emitted by the retry policy engine (crate::policy).
+metrics = { workspace = true }
 
 # Cryptography
 aes-gcm = { workspace = true, features = ["rand_core"] }
@@ -72,6 +74,8 @@ tokio-util = { workspace = true }
 
 [dev-dependencies]
 anyhow = { workspace = true }
+# Debugging recorder for asserting emitted metrics in tests.
+metrics-util = { version = "0.20", features = ["debugging"] }
 insta = { workspace = true, features = ["yaml", "json"] }
 tempfile = { workspace = true }
 temp-env = { workspace = true }

crates/kms/src/policy.rs

@@ -27,6 +27,12 @@
 //! retried automatically: a response lost after the server applied the write
 //! would otherwise be replayed into duplicate side effects (extra key versions,
 //! repeated deletes).
+//!
+//! Every execution also records operation metrics (attempt failures by retry
+//! class, terminal outcome, attempts used, wall-clock duration) through the
+//! process-global `metrics` recorder. Metric labels carry only static enum
+//! values — operation names, classes, outcomes — never key identifiers, key
+//! material, ciphertext, or tokens.
 
 use std::future::Future;
 use std::time::Duration;
@@ -200,6 +206,130 @@ fn equal_jitter(rng: &mut impl RngExt, cap: Duration) -> Duration {
     half + Duration::from_nanos(rng.random_range(0..=spread))
 }
 
+// ---------------------------------------------------------------------------
+// Metrics
+//
+// Every execution is recorded here, at the single choke point all backend
+// calls flow through, so instrumenting a new call site costs nothing beyond
+// naming its operation. Label values are exclusively static enum strings
+// (operation names, classes, outcomes) — key identifiers, key material,
+// ciphertext, and tokens must never reach a metric label.
+// ---------------------------------------------------------------------------
+
+/// Counter: operations executed, by `operation`, `op_class`, and `outcome`.
+const METRIC_OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
+/// Counter: failed attempts, by `operation` and `error_class` (including
+/// `attempt_timeout` for attempts cut off by the per-attempt timeout).
+const METRIC_ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
+/// Histogram: wall-clock duration of a whole operation (attempts plus
+/// backoff), in seconds, by `operation` and `outcome`.
+const METRIC_OPERATION_DURATION_SECONDS: &str = "rustfs_kms_backend_operation_duration_seconds";
+/// Histogram: attempts one operation used before completing, by `operation`
+/// and `outcome`.
+const METRIC_OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
+
+impl OpClass {
+    fn as_label(self) -> &'static str {
+        match self {
+            OpClass::ReadIdempotent => "read_idempotent",
+            OpClass::MutatingNonIdempotent => "mutating_non_idempotent",
+            OpClass::Auth => "auth",
+        }
+    }
+}
+
+impl ErrorClass {
+    fn as_label(self) -> &'static str {
+        match self {
+            ErrorClass::RetryableConn => "retryable_conn",
+            ErrorClass::RetryableStatus => "retryable_status",
+            ErrorClass::Fatal => "fatal",
+        }
+    }
+}
+
+/// How one policy execution terminated, for the `outcome` metric label.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Outcome {
+    Success,
+    /// A fatal-classified failure ended the operation on its first observation.
+    Fatal,
+    /// The attempt budget ran out; the last failure was retryable (including a
+    /// timed-out final attempt).
+    BudgetExhausted,
+    /// The operation deadline ran out before another attempt could complete.
+    DeadlineExceeded,
+    Cancelled,
+}
+
+impl Outcome {
+    fn as_label(self) -> &'static str {
+        match self {
+            Outcome::Success => "success",
+            Outcome::Fatal => "fatal",
+            Outcome::BudgetExhausted => "budget_exhausted",
+            Outcome::DeadlineExceeded => "deadline_exceeded",
+            Outcome::Cancelled => "cancelled",
+        }
+    }
+}
+
+/// Register metric descriptions once per process.
+fn describe_metrics() {
+    static DESCRIBE: std::sync::Once = std::sync::Once::new();
+    DESCRIBE.call_once(|| {
+        metrics::describe_counter!(
+            METRIC_OPERATIONS_TOTAL,
+            "Total KMS backend operations executed under the operation policy, by operation, operation class, and outcome"
+        );
+        metrics::describe_counter!(
+            METRIC_ATTEMPT_FAILURES_TOTAL,
+            "Total failed KMS backend attempts, by operation and retry classification"
+        );
+        metrics::describe_histogram!(
+            METRIC_OPERATION_DURATION_SECONDS,
+            "Wall-clock duration of KMS backend operations including retries and backoff, in seconds"
+        );
+        metrics::describe_histogram!(
+            METRIC_OPERATION_ATTEMPTS,
+            "Number of attempts a KMS backend operation used before completing"
+        );
+    });
+}
+
+/// Record one failed attempt with its retry classification.
+fn record_attempt_failure(operation: &'static str, error_class: &'static str) {
+    metrics::counter!(
+        METRIC_ATTEMPT_FAILURES_TOTAL,
+        "operation" => operation,
+        "error_class" => error_class
+    )
+    .increment(1);
+}
+
+/// Record the terminal outcome of one policy execution.
+fn record_operation(operation: &'static str, class: OpClass, outcome: Outcome, attempts: u32, elapsed: Duration) {
+    metrics::counter!(
+        METRIC_OPERATIONS_TOTAL,
+        "operation" => operation,
+        "op_class" => class.as_label(),
+        "outcome" => outcome.as_label()
+    )
+    .increment(1);
+    metrics::histogram!(
+        METRIC_OPERATION_DURATION_SECONDS,
+        "operation" => operation,
+        "outcome" => outcome.as_label()
+    )
+    .record(elapsed.as_secs_f64());
+    metrics::histogram!(
+        METRIC_OPERATION_ATTEMPTS,
+        "operation" => operation,
+        "outcome" => outcome.as_label()
+    )
+    .record(f64::from(attempts));
+}
+
 /// Run `attempt` under the policy.
 ///
 /// Each attempt is bounded by `attempt_timeout` (further capped by whatever is
@@ -230,13 +360,37 @@ where
 /// [`execute`] with an injectable jitter source so tests can pin deterministic
 /// backoff durations instead of asserting around random sleeps.
 pub(crate) async fn execute_with_jitter<T, F, Fut, J>(
+    operation: &'static str,
+    class: OpClass,
+    policy: &RetryPolicy,
+    cancel: &CancellationToken,
+    jitter: J,
+    attempt: F,
+) -> Result<T>
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = std::result::Result<T, AttemptError>>,
+    J: FnMut(Duration) -> Duration,
+{
+    describe_metrics();
+    let started = Instant::now();
+    let mut attempts_made = 0u32;
+    let (outcome, result) = drive_attempts(operation, class, policy, cancel, jitter, attempt, &mut attempts_made).await;
+    record_operation(operation, class, outcome, attempts_made, started.elapsed());
+    result
+}
+
+/// The attempt loop behind [`execute_with_jitter`], returning the terminal
+/// outcome alongside the result so the caller can record it exactly once.
+async fn drive_attempts<T, F, Fut, J>(
     operation: &'static str,
     class: OpClass,
     policy: &RetryPolicy,
     cancel: &CancellationToken,
     mut jitter: J,
     mut attempt: F,
-) -> Result<T>
+    attempts_made: &mut u32,
+) -> (Outcome, Result<T>)
 where
     F: FnMut() -> Fut,
     Fut: Future<Output = std::result::Result<T, AttemptError>>,
@@ -247,48 +401,68 @@ where
 
     let mut attempt_no = 0u32;
     loop {
-        attempt_no += 1;
         if cancel.is_cancelled() {
-            return Err(KmsError::operation_cancelled(format!(
-                "{operation} cancelled before attempt {attempt_no}"
-            )));
+            return (
+                Outcome::Cancelled,
+                Err(KmsError::operation_cancelled(format!(
+                    "{operation} cancelled before attempt {}",
+                    attempt_no + 1
+                ))),
+            );
         }
         let remaining = deadline.saturating_duration_since(Instant::now());
         if remaining.is_zero() {
-            return Err(KmsError::operation_timed_out(format!(
-                "{operation} exceeded operation deadline of {:?}",
-                policy.op_deadline
-            )));
+            return (
+                Outcome::DeadlineExceeded,
+                Err(KmsError::operation_timed_out(format!(
+                    "{operation} exceeded operation deadline of {:?}",
+                    policy.op_deadline
+                ))),
+            );
         }
 
+        attempt_no += 1;
+        *attempts_made = attempt_no;
         let attempt_budget = policy.attempt_timeout.min(remaining);
         let outcome = tokio::select! {
             biased;
             _ = cancel.cancelled() => {
-                return Err(KmsError::operation_cancelled(format!("{operation} cancelled during attempt {attempt_no}")));
+                return (
+                    Outcome::Cancelled,
+                    Err(KmsError::operation_cancelled(format!("{operation} cancelled during attempt {attempt_no}"))),
+                );
             }
             outcome = tokio::time::timeout(attempt_budget, attempt()) => outcome,
         };
 
         let failure = match outcome {
-            Ok(Ok(value)) => return Ok(value),
-            Ok(Err(failure)) => failure,
-            Err(_) => AttemptError {
-                class: ErrorClass::RetryableConn,
-                error: KmsError::operation_timed_out(format!(
-                    "{operation} attempt {attempt_no} timed out after {attempt_budget:?}"
-                )),
-            },
+            Ok(Ok(value)) => return (Outcome::Success, Ok(value)),
+            Ok(Err(failure)) => {
+                record_attempt_failure(operation, failure.class.as_label());
+                failure
+            }
+            Err(_) => {
+                record_attempt_failure(operation, "attempt_timeout");
+                AttemptError {
+                    class: ErrorClass::RetryableConn,
+                    error: KmsError::operation_timed_out(format!(
+                        "{operation} attempt {attempt_no} timed out after {attempt_budget:?}"
+                    )),
+                }
+            }
         };
 
-        if failure.class == ErrorClass::Fatal || attempt_no >= max_attempts {
-            return Err(failure.error);
+        if failure.class == ErrorClass::Fatal {
+            return (Outcome::Fatal, Err(failure.error));
+        }
+        if attempt_no >= max_attempts {
+            return (Outcome::BudgetExhausted, Err(failure.error));
         }
 
         let backoff = jitter(backoff_cap(policy, attempt_no));
         if backoff >= deadline.saturating_duration_since(Instant::now()) {
             // Not enough deadline budget left for another attempt.
-            return Err(failure.error);
+            return (Outcome::DeadlineExceeded, Err(failure.error));
         }
         tracing::warn!(
             operation,
@@ -300,7 +474,10 @@ where
         tokio::select! {
             biased;
             _ = cancel.cancelled() => {
-                return Err(KmsError::operation_cancelled(format!("{operation} cancelled during retry backoff")));
+                return (
+                    Outcome::Cancelled,
+                    Err(KmsError::operation_cancelled(format!("{operation} cancelled during retry backoff"))),
+                );
             }
             _ = tokio::time::sleep(backoff) => {}
         }
@@ -628,4 +805,313 @@ mod tests {
         assert_eq!(classify_vaultrs(&ClientError::ResponseEmptyError), ErrorClass::Fatal);
         assert_eq!(classify_vaultrs(&ClientError::InvalidLoginMethodError), ErrorClass::Fatal);
     }
+
+    // -- Metric emission ----------------------------------------------------
+    //
+    // Each test installs a thread-local debugging recorder and drives a
+    // paused-clock current-thread runtime inside it, so the emitted metrics
+    // (including virtual-clock durations) are fully deterministic.
+
+    use metrics_util::MetricKind;
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+
+    type MetricEntry = (
+        metrics_util::CompositeKey,
+        Option<metrics::Unit>,
+        Option<metrics::SharedString>,
+        DebugValue,
+    );
+
+    /// Run `test` on a paused current-thread runtime under a debugging
+    /// recorder and return one snapshot of everything it emitted.
+    ///
+    /// A single snapshot per test on purpose: `Snapshotter::snapshot` drains
+    /// the recorded state, so taking it per assertion would only show the
+    /// first assertion any data.
+    fn record_metrics<Out>(test: impl FnOnce() -> std::pin::Pin<Box<dyn Future<Output = Out>>>) -> (Vec<MetricEntry>, Out) {
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let out = metrics::with_local_recorder(&recorder, || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_time()
+                .start_paused(true)
+                .build()
+                .expect("current-thread runtime must build");
+            runtime.block_on(test())
+        });
+        (snapshotter.snapshot().into_vec(), out)
+    }
+
+    fn labels_match(key: &metrics::Key, labels: &[(&str, &str)]) -> bool {
+        labels
+            .iter()
+            .all(|(label, expected)| key.labels().any(|l| l.key() == *label && l.value() == *expected))
+    }
+
+    fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> u64 {
+        snapshot
+            .iter()
+            .filter_map(|(composite, _unit, _description, value)| {
+                let matches = composite.kind() == MetricKind::Counter
+                    && composite.key().name() == name
+                    && labels_match(composite.key(), labels);
+                match (matches, value) {
+                    (true, DebugValue::Counter(count)) => Some(*count),
+                    _ => None,
+                }
+            })
+            .sum()
+    }
+
+    fn histogram_values(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> Vec<f64> {
+        snapshot
+            .iter()
+            .filter_map(|(composite, _unit, _description, value)| {
+                let matches = composite.kind() == MetricKind::Histogram
+                    && composite.key().name() == name
+                    && labels_match(composite.key(), labels);
+                match (matches, value) {
+                    (true, DebugValue::Histogram(values)) => Some(values),
+                    _ => None,
+                }
+            })
+            .flatten()
+            .map(|value| value.into_inner())
+            .collect()
+    }
+
+    #[test]
+    fn metrics_record_retried_success_with_attempts_and_duration() {
+        let calls_in_test = Arc::new(AtomicU32::new(0));
+        let (snapshot, ()) = record_metrics(move || {
+            Box::pin(async move {
+                let policy = policy_of(1_000, 60_000, 3, 100, 2_000);
+                let cancel = CancellationToken::new();
+                let calls_in_attempt = calls_in_test.clone();
+                execute_with_jitter("metrics_read", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, move || {
+                    let calls = calls_in_attempt.clone();
+                    async move {
+                        if calls.fetch_add(1, Ordering::SeqCst) < 2 {
+                            Err(AttemptError {
+                                class: ErrorClass::RetryableStatus,
+                                error: KmsError::backend_error("throttled (429)"),
+                            })
+                        } else {
+                            Ok(())
+                        }
+                    }
+                })
+                .await
+                .expect("retries within budget must succeed");
+            })
+        });
+
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_OPERATIONS_TOTAL,
+                &[
+                    ("operation", "metrics_read"),
+                    ("op_class", "read_idempotent"),
+                    ("outcome", "success")
+                ]
+            ),
+            1
+        );
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_ATTEMPT_FAILURES_TOTAL,
+                &[("operation", "metrics_read"), ("error_class", "retryable_status")]
+            ),
+            2
+        );
+        assert_eq!(
+            histogram_values(
+                &snapshot,
+                METRIC_OPERATION_ATTEMPTS,
+                &[("operation", "metrics_read"), ("outcome", "success")]
+            ),
+            vec![3.0]
+        );
+        // Full-cap backoffs of 100ms and 200ms on the paused clock.
+        let durations = histogram_values(
+            &snapshot,
+            METRIC_OPERATION_DURATION_SECONDS,
+            &[("operation", "metrics_read"), ("outcome", "success")],
+        );
+        assert_eq!(durations.len(), 1);
+        assert!((durations[0] - 0.3).abs() < 1e-9, "expected 0.3s of virtual backoff, got {durations:?}");
+    }
+
+    #[test]
+    fn metrics_record_fatal_outcome_with_single_attempt() {
+        let (snapshot, ()) = record_metrics(|| {
+            Box::pin(async {
+                let policy = policy_of(1_000, 60_000, 5, 100, 2_000);
+                let cancel = CancellationToken::new();
+                let result: Result<()> =
+                    execute_with_jitter("metrics_fatal", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, || async {
+                        Err(AttemptError {
+                            class: ErrorClass::Fatal,
+                            error: KmsError::access_denied("permission denied (403)"),
+                        })
+                    })
+                    .await;
+                result.expect_err("a fatal failure must end the operation");
+            })
+        });
+
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_OPERATIONS_TOTAL,
+                &[("operation", "metrics_fatal"), ("outcome", "fatal")]
+            ),
+            1
+        );
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_ATTEMPT_FAILURES_TOTAL,
+                &[("operation", "metrics_fatal"), ("error_class", "fatal")]
+            ),
+            1
+        );
+        assert_eq!(
+            histogram_values(
+                &snapshot,
+                METRIC_OPERATION_ATTEMPTS,
+                &[("operation", "metrics_fatal"), ("outcome", "fatal")]
+            ),
+            vec![1.0]
+        );
+    }
+
+    #[test]
+    fn metrics_record_mutating_budget_exhausted_after_one_attempt() {
+        let (snapshot, ()) = record_metrics(|| {
+            Box::pin(async {
+                let policy = policy_of(1_000, 60_000, 5, 100, 2_000);
+                let cancel = CancellationToken::new();
+                let result: Result<()> = execute_with_jitter(
+                    "metrics_rotate",
+                    OpClass::MutatingNonIdempotent,
+                    &policy,
+                    &cancel,
+                    full_jitter,
+                    || async { Err(retryable_conn_error()) },
+                )
+                .await;
+                result.expect_err("a mutating operation must not retry a retryable failure");
+            })
+        });
+
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_OPERATIONS_TOTAL,
+                &[
+                    ("operation", "metrics_rotate"),
+                    ("op_class", "mutating_non_idempotent"),
+                    ("outcome", "budget_exhausted")
+                ]
+            ),
+            1
+        );
+        assert_eq!(
+            histogram_values(
+                &snapshot,
+                METRIC_OPERATION_ATTEMPTS,
+                &[("operation", "metrics_rotate"), ("outcome", "budget_exhausted")]
+            ),
+            vec![1.0]
+        );
+    }
+
+    #[test]
+    fn metrics_record_timeouts_and_deadline_outcome() {
+        let (snapshot, ()) = record_metrics(|| {
+            Box::pin(async {
+                // Hung attempts: 10s each against a 25s deadline (see
+                // total_duration_never_exceeds_deadline for the timeline).
+                let policy = policy_of(10_000, 25_000, 5, 100, 2_000);
+                let cancel = CancellationToken::new();
+                let result: Result<()> =
+                    execute_with_jitter("metrics_hung", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, || {
+                        std::future::pending::<AttemptResult<()>>()
+                    })
+                    .await;
+                result.expect_err("hung attempts must exhaust the deadline");
+            })
+        });
+
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_OPERATIONS_TOTAL,
+                &[("operation", "metrics_hung"), ("outcome", "deadline_exceeded")]
+            ),
+            1
+        );
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_ATTEMPT_FAILURES_TOTAL,
+                &[("operation", "metrics_hung"), ("error_class", "attempt_timeout")]
+            ),
+            3
+        );
+        let durations = histogram_values(
+            &snapshot,
+            METRIC_OPERATION_DURATION_SECONDS,
+            &[("operation", "metrics_hung"), ("outcome", "deadline_exceeded")],
+        );
+        assert_eq!(durations.len(), 1);
+        assert!((durations[0] - 25.0).abs() < 1e-9, "expected the full 25s deadline, got {durations:?}");
+    }
+
+    #[test]
+    fn metrics_record_cancelled_outcome() {
+        let calls_in_test = Arc::new(AtomicU32::new(0));
+        let (snapshot, ()) = record_metrics(move || {
+            Box::pin(async move {
+                let policy = policy_of(1_000, 600_000, 5, 10_000, 10_000);
+                let cancel = CancellationToken::new();
+                let canceller = cancel.clone();
+                tokio::spawn(async move {
+                    tokio::time::sleep(Duration::from_millis(500)).await;
+                    canceller.cancel();
+                });
+                let calls_in_attempt = calls_in_test.clone();
+                let result: Result<()> =
+                    execute_with_jitter("metrics_cancel", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, move || {
+                        let calls = calls_in_attempt.clone();
+                        async move {
+                            calls.fetch_add(1, Ordering::SeqCst);
+                            Err(retryable_conn_error())
+                        }
+                    })
+                    .await;
+                result.expect_err("cancellation must abort the backoff");
+            })
+        });
+
+        assert_eq!(
+            counter_value(
+                &snapshot,
+                METRIC_OPERATIONS_TOTAL,
+                &[("operation", "metrics_cancel"), ("outcome", "cancelled")]
+            ),
+            1
+        );
+        assert_eq!(
+            histogram_values(
+                &snapshot,
+                METRIC_OPERATION_ATTEMPTS,
+                &[("operation", "metrics_cancel"), ("outcome", "cancelled")]
+            ),
+            vec![1.0]
+        );
+    }
 }

crates/kms/tests/vault_fault_injection.rs

@@ -0,0 +1,255 @@
+// 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.
+
+//! Fault-injection matrix for the Vault backend operation policy.
+//!
+//! Offline cases run against locally injected transport faults (a closed
+//! port, a listener that never responds) — deterministic, no external
+//! dependencies. Real-Vault cases are `#[ignore]`d and need a dev Vault
+//! (default `http://127.0.0.1:8200`, override with `RUSTFS_KMS_VAULT_ADDR`).
+//!
+//! Throttling (429) and recoverable 5xx responses cannot be forced on a stock
+//! dev Vault, so their retry and metric behavior is pinned deterministically
+//! by the scripted-Vault wiring tests in `backends::vault` and the engine
+//! tests in `policy.rs`. Pointing `RUSTFS_KMS_VAULT_ADDR` at a
+//! fault-injecting proxy reuses the ignored cases here unchanged.
+//!
+//! Every case installs a thread-local debugging metrics recorder and drives a
+//! current-thread runtime inside it, so the policy metrics double as the
+//! request-count assertion even against a real server.
+
+use std::time::Duration;
+
+use metrics_util::MetricKind;
+use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+use rustfs_kms::backends::KmsClient;
+use rustfs_kms::backends::vault::VaultKmsClient;
+use rustfs_kms::{KmsConfig, KmsError, VaultAuthMethod, VaultConfig};
+
+const OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
+const ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
+
+fn vault_config(address: &str, token: &str) -> VaultConfig {
+    VaultConfig {
+        address: address.to_string(),
+        auth_method: VaultAuthMethod::Token {
+            token: token.to_string(),
+        },
+        namespace: None,
+        mount_path: "transit".to_string(),
+        kv_mount: "secret".to_string(),
+        key_path_prefix: "rustfs/kms/fault-injection".to_string(),
+        tls: None,
+    }
+}
+
+fn kms_config(attempt_timeout: Duration, retry_attempts: u32) -> KmsConfig {
+    KmsConfig {
+        timeout: attempt_timeout,
+        retry_attempts,
+        ..KmsConfig::default()
+    }
+}
+
+type MetricEntry = (
+    metrics_util::CompositeKey,
+    Option<metrics::Unit>,
+    Option<metrics::SharedString>,
+    DebugValue,
+);
+
+/// Run `test` on a current-thread runtime under a debugging metrics recorder
+/// and return one snapshot of everything it emitted.
+///
+/// A single snapshot per test on purpose: `Snapshotter::snapshot` drains the
+/// recorded state, so taking it per assertion would only show the first
+/// assertion any data.
+fn record_metrics(test: impl FnOnce() -> std::pin::Pin<Box<dyn std::future::Future<Output = ()>>>) -> Vec<MetricEntry> {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+    metrics::with_local_recorder(&recorder, || {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime must build");
+        runtime.block_on(test());
+    });
+    snapshotter.snapshot().into_vec()
+}
+
+/// Sum of counters with `name` whose labels include every `(key, value)` pair.
+fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> u64 {
+    snapshot
+        .iter()
+        .filter_map(|(composite, _unit, _description, value)| {
+            let key = composite.key();
+            let matches = composite.kind() == MetricKind::Counter
+                && key.name() == name
+                && labels
+                    .iter()
+                    .all(|(label, expected)| key.labels().any(|l| l.key() == *label && l.value() == *expected));
+            match (matches, value) {
+                (true, DebugValue::Counter(count)) => Some(*count),
+                _ => None,
+            }
+        })
+        .sum()
+}
+
+/// Connection refused: connection-class failures are retried up to the
+/// configured budget, then surface as a backend error.
+#[test]
+fn connection_refused_is_retried_within_budget() {
+    // Reserve a loopback port and release it so nothing is listening there.
+    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve a loopback port");
+    let address = format!("http://{}", listener.local_addr().expect("reserved port addr"));
+    drop(listener);
+
+    let snapshot = record_metrics(|| {
+        Box::pin(async move {
+            let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_secs(2), 2))
+                .await
+                .expect("client construction performs no network calls");
+            let error = client
+                .describe_key("fault-injection-refused", None)
+                .await
+                .expect_err("a refused connection must fail the operation");
+            assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
+        })
+    });
+
+    assert_eq!(
+        counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_conn")]),
+        2,
+        "both budgeted attempts must observe the refused connection"
+    );
+    assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), 1);
+    // The static-token login records its own success; the Vault read must not.
+    assert_eq!(
+        counter_value(
+            &snapshot,
+            OPERATIONS_TOTAL,
+            &[("operation", "vault_kv2_read_key"), ("outcome", "success")]
+        ),
+        0
+    );
+}
+
+/// Stalled connection: a server that accepts but never responds is cut off by
+/// the per-attempt timeout (either the policy timer or the equally sized HTTP
+/// client timeout, whichever fires first) instead of hanging forever.
+#[test]
+fn stalled_connection_is_cut_off_by_the_attempt_timeout() {
+    let snapshot = record_metrics(|| {
+        Box::pin(async move {
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+                .await
+                .expect("bind stall listener");
+            let address = format!("http://{}", listener.local_addr().expect("stall listener addr"));
+            // Accept and park every connection without ever responding.
+            tokio::spawn(async move {
+                let mut parked = Vec::new();
+                loop {
+                    let Ok((socket, _)) = listener.accept().await else { return };
+                    parked.push(socket);
+                }
+            });
+
+            let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_millis(250), 1))
+                .await
+                .expect("client construction performs no network calls");
+            let error = client
+                .describe_key("fault-injection-stalled", None)
+                .await
+                .expect_err("a stalled request must be cut off by the attempt timeout");
+            assert!(
+                matches!(error, KmsError::OperationTimedOut { .. } | KmsError::BackendError { .. }),
+                "got {error:?}"
+            );
+        })
+    });
+
+    // The policy timer reports attempt_timeout; the client-level HTTP timeout
+    // surfaces as a connection-class failure. Either way it is exactly one
+    // attempt that was cut off.
+    let cut_off = counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "attempt_timeout")])
+        + counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_conn")]);
+    assert_eq!(cut_off, 1, "the single budgeted attempt must be cut off by a timeout");
+    assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), 1);
+}
+
+fn real_vault_address() -> String {
+    std::env::var("RUSTFS_KMS_VAULT_ADDR").unwrap_or_else(|_| "http://127.0.0.1:8200".to_string())
+}
+
+/// Invalid token against a real Vault: the 403 is fatal — exactly one
+/// attempt, no retry, and the operation fails closed.
+#[test]
+#[ignore] // Requires a running Vault dev server
+fn real_vault_invalid_token_is_fatal_and_never_retried() {
+    let snapshot = record_metrics(|| {
+        Box::pin(async {
+            let config = vault_config(&real_vault_address(), "fault-injection-invalid-token");
+            let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
+                .await
+                .expect("client construction performs no network calls");
+            let error = client
+                .describe_key("fault-injection-forbidden", None)
+                .await
+                .expect_err("an invalid token must be rejected");
+            assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
+        })
+    });
+
+    assert_eq!(
+        counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "fatal")]),
+        1,
+        "a 403 must be observed by exactly one attempt"
+    );
+    assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "fatal")]), 1);
+    assert_eq!(
+        counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_status")]),
+        0,
+        "an auth failure must never be classified as retryable"
+    );
+}
+
+/// Healthy read against a real Vault: a missing key resolves in one attempt
+/// (404 is fatal for retry purposes) and records a fatal outcome rather than
+/// burning the retry budget.
+#[test]
+#[ignore] // Requires a running Vault dev server
+fn real_vault_missing_key_is_resolved_in_one_attempt() {
+    let token = std::env::var("RUSTFS_KMS_VAULT_TOKEN").unwrap_or_else(|_| "dev-only-token".to_string());
+    let snapshot = record_metrics(|| {
+        Box::pin(async move {
+            let config = vault_config(&real_vault_address(), &token);
+            let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
+                .await
+                .expect("client construction performs no network calls");
+            let error = client
+                .describe_key("fault-injection-definitely-missing", None)
+                .await
+                .expect_err("a missing key must resolve to key-not-found");
+            assert!(matches!(error, KmsError::KeyNotFound { .. }), "got {error:?}");
+        })
+    });
+
+    assert_eq!(
+        counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "fatal")]),
+        1,
+        "a 404 must be observed by exactly one attempt"
+    );
+    assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "fatal")]), 1);
+}