test(e2e): pin bounded physical reads for compressed multipart range GETs

Byte-exactness tests stay green if the compressed range seek regresses into
decoding from byte zero: the returned bytes are still correct and only the
read amplification explodes. Assert the cost side as well.

The observation reuses rustfs_io_get_object_shard_read_observed_bytes_total,
already emitted per shard read by the erasure layer, so no production code is
instrumented. The OTLP collector learns to accumulate a second counter, keyed
by its path/role/outcome labels rather than by data-point position, which is
not stable across exports.

Two failure modes the assertions guard against:

- With RUSTFS_OBS_METER_INTERVAL=1, treating one unchanged sample as settled
  measures a delta of zero, because the range read's counter has not been
  exported yet. Settling now requires several consecutive equal samples.
- An upper bound alone passes vacuously on a zero delta, so a lower bound
  turns "measured nothing" into a failure instead of a green run.

Refs rustfs/rustfs#5957, backlog#1848.

crates/e2e_test/src/fake_s3_target/README.md

@@ -4,8 +4,8 @@ This module is the shared failure-injection boundary for replication end-to-end
 
 `FakeS3Target::start()` creates the listener. Add target buckets with `create_bucket`, point a RustFS remote target at `address()`, use `FAKE_ACCESS_KEY` / `FAKE_SECRET_KEY`, then enqueue per-operation faults with `inject`. Faults for one operation are consumed in FIFO order and do not consume faults queued for another operation. A fault is consumed only after `s3s` verifies the full request signature, so anonymous, other-access-key, and bad-signature traffic cannot disturb a script.
 
-Supported data operations are HeadBucket, GetBucketVersioning, PUT/GET/HEAD/DELETE Object, Get/Put/Delete ObjectTagging (tags live per version; Put replaces the whole set, Delete clears it), and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB.
+Supported data operations are HeadBucket, GetBucketVersioning, PUT/GET/HEAD/DELETE Object, and create/upload/complete/abort multipart upload. `create_bucket` models general-purpose buckets in S3's shared global namespace; account-regional namespace buckets and their `-an` names are intentionally out of scope. Buckets are versioned: PUT creates a version, DELETE without `versionId` creates a delete marker, and DELETE with `versionId` removes exactly that version. Internal source version IDs must be UUIDs and are stored canonically. Source mtime is honored only for source-replication PUT/DELETE requests; absent or invalid values use receipt time, matching RustFS, while multipart completion always uses receipt time. Replicated versions are ordered newest-first by source mtime so late older versions and delete markers do not become current. Equal mtimes prefer objects over delete markers, then canonical UUID order; RustFS's internal FileMeta signature tie-break is intentionally out of scope because it is not part of the target S3 protocol. Multipart part numbers follow S3's `1..=10000` range, and every completed part except the final part must be at least 5 MiB.
 
-Fault actions cover HTTP 401/403/503 responses, pre-dispatch delay, connection abort when a logical request-body threshold is reached, streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions. Each record also journals a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract.
+Fault actions cover HTTP 401/403/503 responses, pre-dispatch delay, connection abort when a logical request-body threshold is reached, streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions.
 
 The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type at 1 KiB. A PUT or uploaded part is capped at 64 MiB; a completed multipart object and all stored object/part data are capped at 128 MiB. Body drain, body-permit waits, delay, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound.

crates/e2e_test/src/fake_s3_target/mod.rs

@@ -30,12 +30,10 @@ use s3s::access::{S3Access, S3AccessContext};
 use s3s::auth::SimpleAuth;
 use s3s::dto::{
     AbortMultipartUploadInput, AbortMultipartUploadOutput, CompleteMultipartUploadInput, CompleteMultipartUploadOutput,
-    CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput,
+    CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput, ETag,
-    DeleteObjectTaggingInput, DeleteObjectTaggingOutput, ETag, GetBucketVersioningInput, GetBucketVersioningOutput,
+    GetBucketVersioningInput, GetBucketVersioningOutput, GetObjectInput, GetObjectOutput, HeadBucketInput, HeadBucketOutput,
-    GetObjectInput, GetObjectOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, HeadBucketOutput,
     HeadObjectInput, HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ObjectVersionId, PutObjectInput,
-    PutObjectOutput, PutObjectTaggingInput, PutObjectTaggingOutput, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat,
+    PutObjectOutput, StreamingBlob, Timestamp, TimestampFormat, UploadPartInput, UploadPartOutput,
-    UploadPartInput, UploadPartOutput,
 };
 use s3s::service::{S3Service, S3ServiceBuilder};
 use s3s::validation::{AwsNameValidation, NameValidation};
@@ -105,9 +103,6 @@ pub enum Operation {
     GetObject,
     HeadObject,
     DeleteObject,
-    GetObjectTagging,
-    PutObjectTagging,
-    DeleteObjectTagging,
     ListObjectVersions,
     CreateMultipartUpload,
     UploadPart,
@@ -154,35 +149,6 @@ impl ReplicationTimestampHeaders {
     }
 }
 
-/// Read-proxy related headers observed on a request, journaled so proxy
-/// tests can assert the exact wire contract: the anti-loop marker present,
-/// the replication-check exemption absent, and the client SSE-C key family
-/// forwarded verbatim. The SSE-C key value itself is never retained — only
-/// its presence.
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct ProxyHeaderSnapshot {
-    pub source_proxy_request: Option<String>,
-    pub replication_check: Option<String>,
-    pub ssec_algorithm: Option<String>,
-    pub ssec_key_present: bool,
-    pub ssec_key_md5: Option<String>,
-}
-
-impl ProxyHeaderSnapshot {
-    fn from_headers(headers: &HeaderMap) -> Self {
-        Self {
-            source_proxy_request: header_value(headers, &["x-rustfs-source-proxy-request", "x-minio-source-proxy-request"])
-                .map(bounded_journal_value),
-            replication_check: header_value(headers, &["x-rustfs-source-replication-check", "x-minio-source-replication-check"])
-                .map(bounded_journal_value),
-            ssec_algorithm: header_value(headers, &["x-amz-server-side-encryption-customer-algorithm"])
-                .map(bounded_journal_value),
-            ssec_key_present: headers.contains_key("x-amz-server-side-encryption-customer-key"),
-            ssec_key_md5: header_value(headers, &["x-amz-server-side-encryption-customer-key-md5"]).map(bounded_journal_value),
-        }
-    }
-}
-
 /// Credential-free request metadata retained for deterministic assertions.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct RequestRecord {
@@ -197,7 +163,6 @@ pub struct RequestRecord {
     pub content_length: Option<u64>,
     pub consumed_bytes: Option<usize>,
     pub replication_timestamps: ReplicationTimestampHeaders,
-    pub proxy_headers: ProxyHeaderSnapshot,
     pub fault: Option<FaultAction>,
 }
 
@@ -234,9 +199,6 @@ struct ObjectVersion {
     delete_marker: bool,
     content_type: Option<String>,
     metadata: Option<HashMap<String, String>>,
-    /// Object tags as ordered key/value pairs (PutObjectTagging replaces the
-    /// whole set, DeleteObjectTagging clears it).
-    tags: Vec<(String, String)>,
 }
 
 #[derive(Clone)]
@@ -607,7 +569,6 @@ impl S3Access for FaultAccess {
             .and_then(|value| value.to_str().ok())
             .and_then(|value| value.parse().ok());
         let replication_timestamps = ReplicationTimestampHeaders::from_headers(context.headers());
-        let proxy_headers = ProxyHeaderSnapshot::from_headers(context.headers());
         let fault = record_request(
             &self.control,
             operation,
@@ -615,7 +576,6 @@ impl S3Access for FaultAccess {
             parsed,
             content_length,
             replication_timestamps,
-            proxy_headers,
         );
         if let Some(RequestFault {
             action: FaultAction::Status(status),
@@ -655,9 +615,6 @@ fn operation_from_s3_name(name: &str) -> Operation {
         "GetObject" => Operation::GetObject,
         "HeadObject" => Operation::HeadObject,
         "DeleteObject" => Operation::DeleteObject,
-        "GetObjectTagging" => Operation::GetObjectTagging,
-        "PutObjectTagging" => Operation::PutObjectTagging,
-        "DeleteObjectTagging" => Operation::DeleteObjectTagging,
         "CreateMultipartUpload" => Operation::CreateMultipartUpload,
         "UploadPart" => Operation::UploadPart,
         "CompleteMultipartUpload" => Operation::CompleteMultipartUpload,
@@ -673,7 +630,6 @@ fn record_request(
     parsed: ParsedRequest,
     content_length: Option<u64>,
     replication_timestamps: ReplicationTimestampHeaders,
-    proxy_headers: ProxyHeaderSnapshot,
 ) -> Option<RequestFault> {
     let mut state = lock(control);
     let action = parsed
@@ -699,7 +655,6 @@ fn record_request(
         content_length,
         consumed_bytes: None,
         replication_timestamps,
-        proxy_headers,
         fault: action.clone(),
     });
     action.map(|action| RequestFault { sequence, action })
@@ -766,15 +721,6 @@ fn parse_request(method: &Method, uri: &Uri) -> ParsedRequest {
         (&Method::POST, true) if query.contains_key("uploads") => Operation::CreateMultipartUpload,
         (&Method::POST, true) if upload_id.is_some() => Operation::CompleteMultipartUpload,
         (&Method::DELETE, true) if upload_id.is_some() => Operation::AbortMultipartUpload,
-        (&Method::GET, true) if query.contains_key("tagging") && only_query_keys(&["tagging", "versionId"]) => {
-            Operation::GetObjectTagging
-        }
-        (&Method::PUT, true) if query.contains_key("tagging") && only_query_keys(&["tagging", "versionId"]) => {
-            Operation::PutObjectTagging
-        }
-        (&Method::DELETE, true) if query.contains_key("tagging") && only_query_keys(&["tagging", "versionId"]) => {
-            Operation::DeleteObjectTagging
-        }
         // A replication PUT addresses the source version via `?versionId=`.
         (&Method::PUT, true) if only_query_keys(&["versionId"]) => Operation::PutObject,
         (&Method::GET, true) if only_query_keys(&["versionId"]) => Operation::GetObject,
@@ -1189,33 +1135,6 @@ fn find_version(state: &StoreState, bucket: &str, key: &str, version_id: Option<
     Ok(version.clone())
 }
 
-/// Replace (or clear, with an empty vec) the tag set of the addressed
-/// version, returning its version id. Mirrors `find_version` addressing:
-/// explicit version id or the latest version, delete markers rejected.
-fn set_version_tags(
-    state: &mut StoreState,
-    bucket: &str,
-    key: &str,
-    version_id: Option<&str>,
-    tags: Vec<(String, String)>,
-) -> S3Result<String> {
-    // Resolve first (immutable) so the error paths match find_version.
-    let resolved = find_version(state, bucket, key, version_id)?.version_id;
-    let versions = state
-        .buckets
-        .get_mut(bucket)
-        .expect("bucket existence checked by find_version")
-        .objects
-        .get_mut(key)
-        .expect("key existence checked by find_version");
-    let version = versions
-        .iter_mut()
-        .find(|version| version.version_id == resolved)
-        .expect("version existence checked by find_version");
-    version.tags = tags;
-    Ok(resolved)
-}
-
 #[async_trait]
 impl S3 for FakeBackend {
     async fn head_bucket(&self, req: S3Request<HeadBucketInput>) -> S3Result<S3Response<HeadBucketOutput>> {
@@ -1329,7 +1248,6 @@ impl S3 for FakeBackend {
             delete_marker: false,
             content_type: input.content_type,
             metadata: input.metadata,
-            tags: Vec::new(),
         };
         upsert_version(&mut lock(&self.store), &input.bucket, input.key, version)?;
         Ok(apply_response_fault(
@@ -1387,72 +1305,6 @@ impl S3 for FakeBackend {
         ))
     }
 
-    async fn get_object_tagging(&self, req: S3Request<GetObjectTaggingInput>) -> S3Result<S3Response<GetObjectTaggingOutput>> {
-        let fault = request_fault(&req);
-        apply_non_body_fault(fault.as_ref(), &self.control).await?;
-        let input = req.input;
-        let version = {
-            let state = lock(&self.store);
-            find_version(&state, &input.bucket, &input.key, input.version_id.as_deref())?
-        };
-        let tag_set: TagSet = version
-            .tags
-            .into_iter()
-            .map(|(key, value)| Tag {
-                key: Some(key),
-                value: Some(value),
-            })
-            .collect();
-        Ok(apply_response_fault(
-            S3Response::new(GetObjectTaggingOutput {
-                tag_set,
-                version_id: Some(ObjectVersionId::from(version.version_id)),
-            }),
-            fault.as_ref(),
-        ))
-    }
-
-    async fn put_object_tagging(&self, req: S3Request<PutObjectTaggingInput>) -> S3Result<S3Response<PutObjectTaggingOutput>> {
-        let fault = request_fault(&req);
-        apply_non_body_fault(fault.as_ref(), &self.control).await?;
-        let input = req.input;
-        let tags = input
-            .tagging
-            .tag_set
-            .into_iter()
-            .map(|tag| (tag.key.unwrap_or_default(), tag.value.unwrap_or_default()))
-            .collect();
-        let version_id = {
-            let mut state = lock(&self.store);
-            set_version_tags(&mut state, &input.bucket, &input.key, input.version_id.as_deref(), tags)?
-        };
-        Ok(apply_response_fault(
-            S3Response::new(PutObjectTaggingOutput {
-                version_id: Some(ObjectVersionId::from(version_id)),
-            }),
-            fault.as_ref(),
-        ))
-    }
-
-    async fn delete_object_tagging(
-        &self,
-        req: S3Request<DeleteObjectTaggingInput>,
-    ) -> S3Result<S3Response<DeleteObjectTaggingOutput>> {
-        let fault = request_fault(&req);
-        apply_non_body_fault(fault.as_ref(), &self.control).await?;
-        let input = req.input;
-        let version_id = {
-            let mut state = lock(&self.store);
-            set_version_tags(&mut state, &input.bucket, &input.key, input.version_id.as_deref(), Vec::new())?
-        };
-        Ok(apply_response_fault(
-            S3Response::new(DeleteObjectTaggingOutput {
-                version_id: Some(ObjectVersionId::from(version_id)),
-            }),
-            fault.as_ref(),
-        ))
-    }
-
     async fn delete_object(&self, req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> {
         let fault = request_fault(&req);
         apply_non_body_fault(fault.as_ref(), &self.control).await?;
@@ -1529,7 +1381,6 @@ impl S3 for FakeBackend {
                 delete_marker: true,
                 content_type: None,
                 metadata: None,
-                tags: Vec::new(),
             },
         )?;
         Ok(apply_response_fault(
@@ -1732,7 +1583,6 @@ impl S3 for FakeBackend {
             delete_marker: false,
             content_type: upload.content_type,
             metadata: upload.metadata,
-            tags: Vec::new(),
         };
         let mut state = lock(&self.store);
         let current = state
@@ -3224,7 +3074,6 @@ mod tests {
                     },
                     Some(0),
                     ReplicationTimestampHeaders::default(),
-                    ProxyHeaderSnapshot::default(),
                 );
             }
             let records = lock(&control).requests.clone();
@@ -3247,7 +3096,6 @@ mod tests {
                 },
                 None,
                 ReplicationTimestampHeaders::default(),
-                ProxyHeaderSnapshot::default(),
             );
             {
                 let bounded_records = lock(&bounded_control);

crates/e2e_test/src/inline_fast_path_cluster_test.rs

@@ -67,6 +67,9 @@ type MetricValues = Arc<Mutex<BTreeMap<String, MetricPointVersions>>>;
 
 const KIB: usize = 1024;
 const READER_PATH_COUNTER: &str = "rustfs_io_get_object_reader_path_by_size_total";
+/// Physical bytes the erasure layer pulled from disk, emitted per shard read by
+/// `crates/ecstore/src/erasure/coding/decode.rs`.
+const SHARD_READ_BYTES_COUNTER: &str = "rustfs_io_get_object_shard_read_observed_bytes_total";
 const MSGPACK_JSON_DECODE_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_decode_total";
 const MSGPACK_JSON_FALLBACK_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_fallback_total";
 const MSGPACK_JSON_DECODE_ERROR_COUNTER: &str = "rustfs_system_network_internode_msgpack_json_decode_error_total";
@@ -146,6 +149,7 @@ struct OtlpMetricCollector {
     decode_values: MetricValues,
     fallback_values: MetricValues,
     decode_error_values: MetricValues,
+    shard_read_values: MetricValues,
     task: JoinHandle<()>,
 }
 
@@ -157,10 +161,12 @@ impl OtlpMetricCollector {
         let decode_values = Arc::new(Mutex::new(BTreeMap::new()));
         let fallback_values = Arc::new(Mutex::new(BTreeMap::new()));
         let decode_error_values = Arc::new(Mutex::new(BTreeMap::new()));
+        let shard_read_values = Arc::new(Mutex::new(BTreeMap::new()));
         let task_values = values.clone();
         let task_decode_values = decode_values.clone();
         let task_fallback_values = fallback_values.clone();
         let task_decode_error_values = decode_error_values.clone();
+        let task_shard_read_values = shard_read_values.clone();
         let task = tokio::spawn(async move {
             loop {
                 let Ok((stream, _)) = listener.accept().await else {
@@ -170,6 +176,7 @@ impl OtlpMetricCollector {
                 let decode_values = task_decode_values.clone();
                 let fallback_values = task_fallback_values.clone();
                 let decode_error_values = task_decode_error_values.clone();
+                let shard_read_values = task_shard_read_values.clone();
                 tokio::spawn(async move {
                     let _ = hyper::server::conn::http1::Builder::new()
                         .serve_connection(
@@ -181,6 +188,7 @@ impl OtlpMetricCollector {
                                     decode_values.clone(),
                                     fallback_values.clone(),
                                     decode_error_values.clone(),
+                                    shard_read_values.clone(),
                                 )
                             }),
                         )
@@ -194,10 +202,48 @@ impl OtlpMetricCollector {
             decode_values,
             fallback_values,
             decode_error_values,
+            shard_read_values,
             task,
         })
     }
 
+    /// Total physical bytes read from disk across every shard-read label set.
+    async fn shard_read_bytes_total(&self) -> u64 {
+        self.shard_read_values
+            .lock()
+            .await
+            .values()
+            .map(|versions| versions.values().map(|(_, value)| *value).sum::<u64>())
+            .sum()
+    }
+
+    /// Waits until the shard-read counter stops advancing so a measurement window
+    /// is not polluted by exports still in flight.
+    ///
+    /// Requires several consecutive equal samples spanning more than one export
+    /// interval (`RUSTFS_OBS_METER_INTERVAL=1`): a single unchanged sample only
+    /// proves the latest export has not landed yet, which silently reads as "no
+    /// disk reads happened" and makes any upper-bound assertion vacuous.
+    async fn wait_for_shard_read_bytes_to_settle(&self) -> TestResult<u64> {
+        const REQUIRED_STABLE_SAMPLES: usize = 5;
+        let mut last = self.shard_read_bytes_total().await;
+        let mut stable = 0;
+        for _ in 0..60 {
+            sleep(Duration::from_millis(500)).await;
+            let current = self.shard_read_bytes_total().await;
+            if current == last {
+                stable += 1;
+                if stable >= REQUIRED_STABLE_SAMPLES {
+                    return Ok(current);
+                }
+            } else {
+                stable = 0;
+                last = current;
+            }
+        }
+        Err("timed out waiting for shard-read byte counter to settle".into())
+    }
+
     async fn reader_path_total(&self, path: &str, object_class: &str, size_bucket: &str) -> u64 {
         self.reader_path_values(path, object_class, size_bucket).await.values().sum()
     }
@@ -321,6 +367,7 @@ async fn handle_metric_export(
     decode_values: MetricValues,
     fallback_values: MetricValues,
     decode_error_values: MetricValues,
+    shard_read_values: MetricValues,
 ) -> Result<Response<Full<Bytes>>, Infallible> {
     if request.uri().path() != "/v1/metrics" {
         return Ok(response(StatusCode::NOT_FOUND));
@@ -354,7 +401,9 @@ async fn handle_metric_export(
             let mut decode_values = decode_values.lock().await;
             let mut fallback_values = fallback_values.lock().await;
             let mut decode_error_values = decode_error_values.lock().await;
+            let mut shard_read_values = shard_read_values.lock().await;
             record_reader_path_metrics(&export, &mut values);
+            record_shard_read_bytes_metrics(&export, &mut shard_read_values);
             record_msgpack_decode_metrics(&export, &mut decode_values);
             record_msgpack_fallback_metrics(&export, &mut fallback_values);
             record_msgpack_decode_error_metrics(&export, &mut decode_error_values);
@@ -375,6 +424,50 @@ fn reader_path_metric_key(path: &str, object_class: &str, size_bucket: &str) ->
     format!("{path}\u{1f}{object_class}\u{1f}{size_bucket}")
 }
 
+/// Accumulates `SHARD_READ_BYTES_COUNTER` across all label sets. Only the total
+/// matters: it is the number of physical bytes the erasure layer actually pulled
+/// from disk, which is what separates a bounded per-part read from a decode of
+/// the whole object.
+fn record_shard_read_bytes_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap<String, MetricPointVersions>) {
+    for resource_metrics in &export.resource_metrics {
+        for scope_metrics in &resource_metrics.scope_metrics {
+            for metric in &scope_metrics.metrics {
+                if metric.name != SHARD_READ_BYTES_COUNTER {
+                    continue;
+                }
+                let Some(metric::Data::Sum(sum)) = &metric.data else {
+                    continue;
+                };
+                for point in &sum.data_points {
+                    let Some(number_data_point::Value::AsInt(value)) = point.value.as_ref() else {
+                        continue;
+                    };
+                    let value = u64::try_from(*value).unwrap_or_default();
+                    // Keyed by labels, not by position: point order within an export
+                    // is not guaranteed stable, so an index key would alias distinct
+                    // series across batches.
+                    let key = format!(
+                        "{}\u{1f}{}\u{1f}{}",
+                        attribute_string(&point.attributes, "path").unwrap_or_default(),
+                        attribute_string(&point.attributes, "role").unwrap_or_default(),
+                        attribute_string(&point.attributes, "outcome").unwrap_or_default(),
+                    );
+                    values
+                        .entry(key)
+                        .or_default()
+                        .entry(point.start_time_unix_nano)
+                        .and_modify(|current| {
+                            if point.time_unix_nano >= current.0 {
+                                *current = (point.time_unix_nano, value);
+                            }
+                        })
+                        .or_insert((point.time_unix_nano, value));
+                }
+            }
+        }
+    }
+}
+
 fn record_reader_path_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap<String, MetricPointVersions>) {
     for resource_metrics in &export.resource_metrics {
         for scope_metrics in &resource_metrics.scope_metrics {
@@ -1864,6 +1957,86 @@ async fn four_node_multipart_disk_compression_roundtrip() -> TestResult {
     Ok(())
 }
 
+/// A tail range over a compressed multipart object must read only the physical
+/// data it needs, not decode the object from byte zero.
+///
+/// The byte-exactness tests around this one stay green even if the seek path
+/// regresses into decoding from the start of the object: the bytes returned are
+/// still correct, only the read amplification explodes. This asserts the cost
+/// side, using `SHARD_READ_BYTES_COUNTER` — already emitted per shard read by the
+/// erasure layer, so no production code is instrumented for the test.
+///
+/// `get_compressed_offsets` skips whole preceding parts by their stored size and
+/// then seeks inside the covering part via its compression index, so a bounded
+/// read costs on the order of the covering part's block size against a ~5 MiB
+/// object.
+#[tokio::test]
+#[serial]
+async fn four_node_compressed_multipart_tail_range_reads_are_bounded() -> TestResult {
+    init_logging();
+
+    let collector = OtlpMetricCollector::start().await?;
+    let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
+    configure_reader_metric_cluster(&mut cluster, &collector);
+    cluster.set_env("RUSTFS_COMPRESSION_ENABLED", "true");
+    cluster.set_env("RUSTFS_COMPRESSION_MULTIPART_ENABLED", "true");
+    cluster.start().await?;
+
+    let bucket = "inline-multipart-compression-tail-range";
+    cluster.create_test_bucket(bucket).await?;
+    let client = cluster.create_s3_client(0)?;
+    let key = "multipart/tail-range.txt";
+    let (body, _second_part, etag) = put_two_part_multipart(&client, bucket, key).await?;
+
+    // Establish that the object really took the compressed read path; otherwise a
+    // small delta below would only prove compression never happened.
+    assert_reader_path(
+        &collector,
+        &client,
+        ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, etag.as_deref(), None), LEGACY_DUPLEX, COMPRESSED),
+    )
+    .await?;
+
+    let baseline = collector.wait_for_shard_read_bytes_to_settle().await?;
+
+    let tail_len = 4 * KIB;
+    let start = body.len() - tail_len;
+    let end = body.len() - 1;
+    let range = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .range(format!("bytes={start}-{end}"))
+        .send()
+        .await?;
+    let tail = range.body.collect().await?.into_bytes();
+    assert_eq!(tail.as_ref(), &body[start..], "tail range returned wrong bytes");
+
+    let after = collector.wait_for_shard_read_bytes_to_settle().await?;
+    let read_bytes = after.saturating_sub(baseline);
+
+    // A zero delta means the window caught nothing — an unexported counter, or a
+    // read served without touching the erasure layer — which would make the upper
+    // bound vacuously true. Fail instead of passing blind.
+    assert!(
+        read_bytes > 0,
+        "no shard reads observed for the tail range; the budget assertion below would be vacuous"
+    );
+
+    // Part 1 alone is MPU_PART_1_SIZE, so a whole-object decode cannot come in
+    // under it. Half the logical size leaves generous headroom for erasure padding
+    // and unrelated background reads while still failing loudly on a full decode.
+    let budget = (body.len() / 2) as u64;
+    assert!(
+        read_bytes < budget,
+        "tail range read {read_bytes} physical bytes for a {tail_len}-byte range (budget {budget}, object {} bytes): \
+         the read is not bounded to the covering part",
+        body.len()
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 #[serial]
 async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> TestResult {

crates/e2e_test/src/replication_extension_test.rs

@@ -8417,304 +8417,3 @@ async fn test_scanner_never_compensates_when_existing_object_replication_disable
 
     Ok(())
 }
-
-/// Shared setup for the P1-5 read-proxy scenarios (backlog#1675): a RustFS
-/// source with an enabled replication rule pointing at the fake target, and
-/// an object seeded DIRECTLY on the target — it exists remotely but not
-/// locally, exactly the active-active replication-lag window the read proxy
-/// serves.
-async fn start_read_proxy_lab(
-    source_bucket: &str,
-    target_bucket: &str,
-) -> Result<(FakeS3Target, RustFSTestEnvironment, Client, Client), Box<dyn Error + Send + Sync>> {
-    let target = FakeS3Target::start().await?;
-    target.create_bucket(target_bucket);
-    target.assign_own_version_ids(true);
-
-    let mut source_env = RustFSTestEnvironment::new().await?;
-    let mut process_env = replication_fast_env();
-    process_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
-    process_env.extend_from_slice(&[
-        ("NO_PROXY", "127.0.0.1,localhost"),
-        ("HTTP_PROXY", ""),
-        ("HTTPS_PROXY", ""),
-        ("RUST_LOG", "error"),
-    ]);
-    source_env.start_rustfs_server_with_env(vec![], &process_env).await?;
-
-    let source_client = source_env.create_s3_client();
-    source_client.create_bucket().bucket(source_bucket).send().await?;
-    enable_bucket_versioning(&source_env, source_bucket).await?;
-    let target_arn = set_replication_target_with_options(
-        &source_env,
-        source_bucket,
-        ReplicationTargetOptions {
-            endpoint: &target.address(),
-            access_key: FAKE_ACCESS_KEY,
-            secret_key: FAKE_SECRET_KEY,
-            target_bucket,
-            secure: false,
-            skip_tls_verify: false,
-            ca_cert_pem: None,
-        },
-    )
-    .await?;
-    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
-
-    let target_client = Client::from_conf(crate::common::build_test_s3_config(
-        target.endpoint(),
-        FAKE_ACCESS_KEY,
-        FAKE_SECRET_KEY,
-        None,
-        "read-proxy-e2e",
-    ));
-
-    Ok((target, source_env, source_client, target_client))
-}
-
-/// P1-5 (backlog#1675): during the active-active replication lag window a
-/// GET/HEAD for an object the local site does not have yet is proxied to the
-/// replication target. Pins the wire contract: the anti-loop
-/// `source-proxy-request` marker is sent, the replication worker's
-/// `source-replication-check` SSE-C exemption is NEVER sent, client SSE-C
-/// headers are forwarded verbatim, and an inbound request that was itself
-/// proxied is answered locally (404) without touching the target.
-#[tokio::test]
-#[serial]
-async fn test_get_and_head_proxy_unreplicated_object_to_replication_target() -> TestResult {
-    init_logging();
-
-    let source_bucket = "proxy-read-src";
-    let target_bucket = "proxy-read-dst";
-    let (target, source_env, source_client, target_client) = start_read_proxy_lab(source_bucket, target_bucket).await?;
-
-    let payload = b"proxy payload".to_vec();
-    target_client
-        .put_object()
-        .bucket(target_bucket)
-        .key("proxy-only")
-        .body(ByteStream::from(payload.clone()))
-        .send()
-        .await?;
-    target.take_requests();
-
-    // a. GET of the locally-missing object is served through the proxy.
-    let got = source_client
-        .get_object()
-        .bucket(source_bucket)
-        .key("proxy-only")
-        .send()
-        .await
-        .map_err(|err| format!("proxied GET failed: {}", err.into_service_error()))?;
-    assert_eq!(got.content_length, Some(payload.len() as i64));
-    let body = got.body.collect().await?.into_bytes();
-    assert_eq!(body.as_ref(), payload.as_slice(), "proxied GET must stream the target's body");
-
-    let get_record = target
-        .requests()
-        .into_iter()
-        .find(|record| record.operation == FakeTargetOperation::GetObject && record.key.as_deref() == Some("proxy-only"))
-        .ok_or("fake target never received the proxied GET")?;
-    assert_eq!(
-        get_record.proxy_headers.source_proxy_request.as_deref(),
-        Some("true"),
-        "proxied GET must carry the anti-loop source-proxy-request marker"
-    );
-    assert!(
-        get_record.proxy_headers.replication_check.is_none(),
-        "proxied GET must never carry the replication worker's source-replication-check exemption"
-    );
-    assert!(
-        get_record.proxy_headers.ssec_algorithm.is_none() && !get_record.proxy_headers.ssec_key_present,
-        "no client SSE-C headers were sent, so none may be forwarded"
-    );
-
-    // a2. Client SSE-C headers travel verbatim to the target (the target owns
-    // the real SSE-C decryption; the plaintext fake simply ignores them).
-    target.take_requests();
-    let ssec_key = "01234567890123456789012345678901";
-    let ssec_key_b64 = BASE64_STANDARD.encode(ssec_key);
-    let ssec_key_md5 = sse_customer_key_md5_base64(ssec_key);
-    let _ = source_client
-        .get_object()
-        .bucket(source_bucket)
-        .key("proxy-only")
-        .sse_customer_algorithm("AES256")
-        .sse_customer_key(&ssec_key_b64)
-        .sse_customer_key_md5(&ssec_key_md5)
-        .send()
-        .await
-        .map_err(|err| format!("proxied SSE-C GET failed: {}", err.into_service_error()))?;
-    let ssec_record = target
-        .requests()
-        .into_iter()
-        .find(|record| record.operation == FakeTargetOperation::GetObject && record.key.as_deref() == Some("proxy-only"))
-        .ok_or("fake target never received the proxied SSE-C GET")?;
-    assert_eq!(ssec_record.proxy_headers.ssec_algorithm.as_deref(), Some("AES256"));
-    assert!(ssec_record.proxy_headers.ssec_key_present, "SSE-C key header must be forwarded verbatim");
-    assert_eq!(ssec_record.proxy_headers.ssec_key_md5.as_deref(), Some(ssec_key_md5.as_str()));
-    assert!(ssec_record.proxy_headers.replication_check.is_none());
-
-    // b. HEAD of the locally-missing object is served through the proxy.
-    target.take_requests();
-    let head = source_client
-        .head_object()
-        .bucket(source_bucket)
-        .key("proxy-only")
-        .send()
-        .await
-        .map_err(|err| format!("proxied HEAD failed: {}", err.into_service_error()))?;
-    assert_eq!(head.content_length, Some(payload.len() as i64));
-    let head_record = target
-        .requests()
-        .into_iter()
-        .find(|record| record.operation == FakeTargetOperation::HeadObject && record.key.as_deref() == Some("proxy-only"))
-        .ok_or("fake target never received the proxied HEAD")?;
-    assert_eq!(head_record.proxy_headers.source_proxy_request.as_deref(), Some("true"));
-    assert!(head_record.proxy_headers.replication_check.is_none());
-
-    // c. Anti-loop: an inbound request that already carries the proxy marker
-    // is answered locally with 404 and never forwarded to the target.
-    target.take_requests();
-    let err = source_client
-        .get_object()
-        .bucket(source_bucket)
-        .key("proxy-only")
-        .customize()
-        .mutate_request(|req| {
-            req.headers_mut().insert("x-minio-source-proxy-request", "true");
-        })
-        .send()
-        .await
-        .expect_err("anti-loop GET must fail locally instead of proxying");
-    let service_err = err.into_service_error();
-    assert!(service_err.is_no_such_key(), "anti-loop GET must 404, got: {service_err}");
-    assert!(
-        !target
-            .requests()
-            .iter()
-            .any(|record| record.operation == FakeTargetOperation::GetObject),
-        "anti-loop GET must not reach the replication target; journal: {:?}",
-        target.requests()
-    );
-
-    // c2. MinIO ProxyHeaderSet parity: the header's mere PRESENCE disables
-    // proxying — "false" is exactly what a peer's replication worker sends on
-    // its convergence HEADs, and proxying that miss back would fake
-    // convergence.
-    target.take_requests();
-    let err = source_client
-        .get_object()
-        .bucket(source_bucket)
-        .key("proxy-only")
-        .customize()
-        .mutate_request(|req| {
-            req.headers_mut().insert("x-minio-source-proxy-request", "false");
-        })
-        .send()
-        .await
-        .expect_err("proxy-header-set GET must fail locally instead of proxying");
-    let service_err = err.into_service_error();
-    assert!(service_err.is_no_such_key(), "proxy-header-set GET must 404, got: {service_err}");
-    assert!(
-        !target
-            .requests()
-            .iter()
-            .any(|record| record.operation == FakeTargetOperation::GetObject),
-        "proxy-header-set GET must not reach the replication target; journal: {:?}",
-        target.requests()
-    );
-
-    // d. The replication worker's own convergence HEAD against the target
-    // must carry `source-proxy-request: false` (never proxied back) and the
-    // replication-check exemption. Trigger real replication and inspect the
-    // fake journal.
-    target.take_requests();
-    source_client
-        .put_object()
-        .bucket(source_bucket)
-        .key("worker-replicated")
-        .body(ByteStream::from_static(b"worker payload"))
-        .send()
-        .await?;
-    wait_for_target_request_version_id(&target, FakeTargetOperation::PutObject, "worker-replicated").await?;
-    let worker_head = target
-        .requests()
-        .into_iter()
-        .find(|record| record.operation == FakeTargetOperation::HeadObject && record.key.as_deref() == Some("worker-replicated"))
-        .ok_or_else(|| format!("replication worker never HEAD-ed the target; journal: {:?}", target.requests()))?;
-    assert_eq!(
-        worker_head.proxy_headers.source_proxy_request.as_deref(),
-        Some("false"),
-        "worker convergence HEAD must send source-proxy-request: false so the target answers locally"
-    );
-    assert_eq!(
-        worker_head.proxy_headers.replication_check.as_deref(),
-        Some("true"),
-        "worker convergence HEAD keeps the replication-check exemption"
-    );
-
-    drop(source_env);
-    target.shutdown().await;
-    Ok(())
-}
-
-/// P1-5 (backlog#1675): GetObjectTagging for an object missing locally is
-/// proxied to the replication target with the anti-loop marker, mirroring
-/// MinIO `proxyGetTaggingToRepTarget`.
-#[tokio::test]
-#[serial]
-async fn test_get_object_tagging_proxies_unreplicated_object_to_replication_target() -> TestResult {
-    init_logging();
-
-    let source_bucket = "proxy-tag-src";
-    let target_bucket = "proxy-tag-dst";
-    let (target, source_env, source_client, target_client) = start_read_proxy_lab(source_bucket, target_bucket).await?;
-
-    target_client
-        .put_object()
-        .bucket(target_bucket)
-        .key("proxy-tagged")
-        .body(ByteStream::from_static(b"tagged payload"))
-        .send()
-        .await?;
-    target_client
-        .put_object_tagging()
-        .bucket(target_bucket)
-        .key("proxy-tagged")
-        .tagging(
-            aws_sdk_s3::types::Tagging::builder()
-                .tag_set(aws_sdk_s3::types::Tag::builder().key("team").value("storage").build()?)
-                .build()?,
-        )
-        .send()
-        .await?;
-    target.take_requests();
-
-    let tags = source_client
-        .get_object_tagging()
-        .bucket(source_bucket)
-        .key("proxy-tagged")
-        .send()
-        .await
-        .map_err(|err| format!("proxied GetObjectTagging failed: {}", err.into_service_error()))?;
-    assert_eq!(tags.tag_set.len(), 1, "proxied tagging read must return the target's tags");
-    assert_eq!(tags.tag_set[0].key.as_str(), "team");
-    assert_eq!(tags.tag_set[0].value.as_str(), "storage");
-
-    let record = target
-        .requests()
-        .into_iter()
-        .find(|record| record.operation == FakeTargetOperation::GetObjectTagging && record.key.as_deref() == Some("proxy-tagged"))
-        .ok_or("fake target never received the proxied GetObjectTagging")?;
-    assert_eq!(
-        record.proxy_headers.source_proxy_request.as_deref(),
-        Some("true"),
-        "proxied tagging read must carry the anti-loop marker"
-    );
-    assert!(record.proxy_headers.replication_check.is_none());
-
-    drop(source_env);
-    target.shutdown().await;
-    Ok(())
-}

crates/ecstore/src/api/mod.rs

@@ -198,13 +198,12 @@ pub mod bucket {
             ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
             VersionPurgeStatusType, XferStats, commit_force_delete_intent, complete_force_delete_intent,
             delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool,
-            get_global_replication_stats, get_proxy_targets, init_background_replication,
+            get_global_replication_stats, init_background_replication, invalid_replication_config_status_field,
-            invalid_replication_config_status_field, persist_force_delete_intent, read_durable_mrf_backlog,
+            persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta,
-            replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns,
+            replication_statuses_map, replication_target_arns, resync_start_conflict_id, should_remove_replication_target,
-            resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication,
+            should_schedule_delete_replication, should_use_existing_delete_replication_info,
-            should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
+            should_use_existing_delete_replication_source, unsupported_replication_config_field,
-            unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
+            validate_replication_config_structure, validate_replication_config_target_arns, version_purge_status_to_filemeta,
-            version_purge_status_to_filemeta,
         };
     }
 

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -27,15 +27,10 @@ use aws_sdk_s3::config::SharedHttpClient;
 use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::error::SdkError;
 use aws_sdk_s3::operation::complete_multipart_upload::CompleteMultipartUploadOutput;
-use aws_sdk_s3::operation::delete_object_tagging::{DeleteObjectTaggingError, DeleteObjectTaggingOutput};
-use aws_sdk_s3::operation::get_object::{GetObjectError, GetObjectOutput};
-use aws_sdk_s3::operation::get_object_tagging::{GetObjectTaggingError, GetObjectTaggingOutput};
 use aws_sdk_s3::operation::head_bucket::HeadBucketError;
 use aws_sdk_s3::operation::head_object::HeadObjectError;
-use aws_sdk_s3::operation::put_object_tagging::{PutObjectTaggingError, PutObjectTaggingOutput};
 use aws_sdk_s3::operation::upload_part::UploadPartOutput;
 use aws_sdk_s3::primitives::ByteStream;
-use aws_sdk_s3::types::Tagging as SdkTagging;
 use aws_sdk_s3::types::{
     ChecksumMode, CompletedMultipartUpload, CompletedPart, ObjectLockLegalHoldStatus, ObjectLockRetentionMode,
 };
@@ -62,8 +57,8 @@ use rustfs_utils::http::{
     is_rustfs_header, is_standard_header, is_storageclass_header,
 };
 use rustfs_utils::http::{
-    SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_PROXY_REQUEST,
+    SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK,
-    SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
+    SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
     SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
     insert_header,
 };
@@ -1451,43 +1446,6 @@ fn resolve_put_api_version_id(source_version_id: &str) -> Option<&str> {
     }
 }
 
-/// Resolve the S3 `versionId` for a proxied read against a remote target.
-/// RustFS represents the null version internally as the nil UUID while the S3
-/// API addresses it as the literal "null" (same mapping as
-/// [`resolve_put_api_version_id`]); empty means "no version requested".
-fn resolve_read_api_version_id(version_id: Option<String>) -> Option<String> {
-    let version_id = version_id?;
-    let trimmed = version_id.trim();
-    if trimmed.is_empty() {
-        None
-    } else if Uuid::parse_str(trimmed).is_ok_and(|uuid| uuid.is_nil()) {
-        Some(rustfs_filemeta::NULL_VERSION_ID.to_string())
-    } else {
-        Some(trimmed.to_string())
-    }
-}
-
-/// Outbound header set for a proxied read: the caller-provided passthrough
-/// headers (client SSE-C key family, conditional headers) plus the anti-loop
-/// `source-proxy-request` marker in both the x-rustfs- and x-minio- prefixes
-/// (a MinIO target only understands the latter). Never adds
-/// `source-replication-check`: that exemption channel belongs exclusively to
-/// the replication worker's HEAD.
-fn proxy_outbound_headers(mut extra_headers: HeaderMap) -> HeaderMap {
-    insert_header(&mut extra_headers, SUFFIX_SOURCE_PROXY_REQUEST, "true");
-    extra_headers
-}
-
-/// Copy `headers` onto an SDK request inside `customize().map_request` (runs
-/// before signing, so the headers join the SigV4 canonical request).
-fn apply_extra_headers(mut req: HttpRequest, headers: &HeaderMap) -> Result<HttpRequest, std::convert::Infallible> {
-    for (k, v) in headers.iter() {
-        req.headers_mut()
-            .insert(k.as_str().to_string(), v.to_str().unwrap_or("").to_string());
-    }
-    Ok(req)
-}
-
 /// Append `versionId=<id>` to an already-built request URI. aws-sdk-s3's
 /// `PutObjectInput` / `CreateMultipartUploadInput` expose no version id
 /// member, so the query is spliced in via `map_request`, which runs at
@@ -1893,13 +1851,6 @@ impl TargetClient {
         // worker cannot hold; otherwise SSE-C replicas never converge on HEAD.
         let mut headers = HeaderMap::new();
         insert_header(&mut headers, SUFFIX_SOURCE_REPLICATION_CHECK, "true");
-        // `source-proxy-request: false` (MinIO `ProxyHeaderSet` semantics):
-        // the header's mere presence tells the receiver to answer LOCALLY
-        // instead of proxying the miss back to us. Without it, a not-found on
-        // the target gets read-proxied back to this source, echoes the source
-        // object with an identical ETag, and the worker concludes the object
-        // already converged — so it never actually replicates it.
-        insert_header(&mut headers, SUFFIX_SOURCE_PROXY_REQUEST, "false");
         match self
             .client
             .head_object()
@@ -1924,129 +1875,6 @@ impl TargetClient {
         }
     }
 
-    /// HEAD used by the read-proxy path (GET/HEAD of an object not yet
-    /// replicated locally, MinIO `proxyHeadToRepTarget`).
-    ///
-    /// Deliberately different from [`TargetClient::head_object`]: it must NOT
-    /// send `source-replication-check` — that header is the replication
-    /// worker's SSE-C metadata exemption channel. A proxied client request
-    /// instead forwards the client's own SSE-C headers (`extra_headers`) so
-    /// the target performs the real SSE-C validation/decryption. The
-    /// `source-proxy-request` marker is always added so the target does not
-    /// proxy the request onward (anti-loop).
-    pub async fn head_object_for_proxy(
-        &self,
-        bucket: &str,
-        object: &str,
-        version_id: Option<String>,
-        range: Option<String>,
-        part_number: Option<i32>,
-        extra_headers: HeaderMap,
-    ) -> Result<HeadObjectOutput, SdkError<HeadObjectError>> {
-        let headers = proxy_outbound_headers(extra_headers);
-        self.client
-            .head_object()
-            .bucket(bucket)
-            .key(object)
-            .set_version_id(resolve_read_api_version_id(version_id))
-            .set_range(range)
-            .set_part_number(part_number)
-            .customize()
-            .map_request(move |req| apply_extra_headers(req, &headers))
-            .send()
-            .await
-    }
-
-    /// GET used by the read-proxy path (MinIO `proxyGetToReplicationTarget`).
-    /// Returns the streaming SDK output; callers must forward the body without
-    /// buffering it. Same header contract as [`Self::head_object_for_proxy`]:
-    /// anti-loop marker on, replication-check never sent, client SSE-C /
-    /// conditional headers forwarded verbatim via `extra_headers`.
-    pub async fn get_object(
-        &self,
-        bucket: &str,
-        object: &str,
-        version_id: Option<String>,
-        range: Option<String>,
-        part_number: Option<i32>,
-        extra_headers: HeaderMap,
-    ) -> Result<GetObjectOutput, SdkError<GetObjectError>> {
-        let headers = proxy_outbound_headers(extra_headers);
-        self.client
-            .get_object()
-            .bucket(bucket)
-            .key(object)
-            .set_version_id(resolve_read_api_version_id(version_id))
-            .set_range(range)
-            .set_part_number(part_number)
-            .customize()
-            .map_request(move |req| apply_extra_headers(req, &headers))
-            .send()
-            .await
-    }
-
-    /// GetObjectTagging for the tagging read-proxy path
-    /// (MinIO `proxyGetTaggingToRepTarget`). Anti-loop marker always added.
-    pub async fn get_object_tagging(
-        &self,
-        bucket: &str,
-        object: &str,
-        version_id: Option<String>,
-    ) -> Result<GetObjectTaggingOutput, SdkError<GetObjectTaggingError>> {
-        let headers = proxy_outbound_headers(HeaderMap::new());
-        self.client
-            .get_object_tagging()
-            .bucket(bucket)
-            .key(object)
-            .set_version_id(resolve_read_api_version_id(version_id))
-            .customize()
-            .map_request(move |req| apply_extra_headers(req, &headers))
-            .send()
-            .await
-    }
-
-    /// PutObjectTagging for the tagging proxy path
-    /// (MinIO `proxyTaggingToRepTarget`). Anti-loop marker always added.
-    pub async fn put_object_tagging(
-        &self,
-        bucket: &str,
-        object: &str,
-        version_id: Option<String>,
-        tagging: SdkTagging,
-    ) -> Result<PutObjectTaggingOutput, SdkError<PutObjectTaggingError>> {
-        let headers = proxy_outbound_headers(HeaderMap::new());
-        self.client
-            .put_object_tagging()
-            .bucket(bucket)
-            .key(object)
-            .set_version_id(resolve_read_api_version_id(version_id))
-            .tagging(tagging)
-            .customize()
-            .map_request(move |req| apply_extra_headers(req, &headers))
-            .send()
-            .await
-    }
-
-    /// DeleteObjectTagging for the tagging proxy path
-    /// (MinIO `proxyTaggingToRepTarget`). Anti-loop marker always added.
-    pub async fn delete_object_tagging(
-        &self,
-        bucket: &str,
-        object: &str,
-        version_id: Option<String>,
-    ) -> Result<DeleteObjectTaggingOutput, SdkError<DeleteObjectTaggingError>> {
-        let headers = proxy_outbound_headers(HeaderMap::new());
-        self.client
-            .delete_object_tagging()
-            .bucket(bucket)
-            .key(object)
-            .set_version_id(resolve_read_api_version_id(version_id))
-            .customize()
-            .map_request(move |req| apply_extra_headers(req, &headers))
-            .send()
-            .await
-    }
-
     /// On success returns the version id the target assigned (from
     /// `x-amz-version-id`), letting callers audit the version-identity
     /// contract — a target that adopts the source version echoes it back.

crates/ecstore/src/bucket/replication/README.md

@@ -14,7 +14,6 @@ paths.
 | `datatypes.rs` | ECStore compatibility re-export for resync status enums. | Re-exports `rustfs-replication` contracts while downstream facade consumers migrate. |
 | `replication_object_decision_boundary.rs` | Object replication option DTOs, resync target projection, delete replication decisions, and multipart planning helpers. | Keeps ECStore runtime modules from importing object decision contracts directly from `rustfs-replication`. |
 | `replication_pool.rs` | Replication queue, worker pool, MRF persistence, bucket stats, and delete/object scheduling. | Depends on bucket target sys, bucket metadata sys, metadata paths, queue contracts through the queue boundary, file metadata replication contracts through local boundaries, config storage, storage contracts through the replication storage boundary, runtime sources, and notification state. |
-| `replication_proxy.rs` | Proxy-target selection for GET/HEAD/Tagging reads of objects not yet replicated locally (MinIO `getProxyTargets` parity: anti-loop, version-suspended, and no-config empty branches). | Uses replication config lookup, rule matching, and target clients through local boundaries. |
 | `replication_queue_boundary.rs` | Queue/admission DTOs, heal queue DTOs, worker sizing, and backpressure helpers. | Keeps ECStore runtime modules from importing queue/backpressure contracts directly from `rustfs-replication`. |
 | `replication_resync_boundary.rs` | Resync DTOs, status classifiers, persisted resync/MRF codec wrappers, and ECStore error mapping. | Keeps ECStore runtime modules from importing resync contract helpers directly from `rustfs-replication`. |
 | `replication_resyncer.rs` | Object replication, delete replication, resync execution, target calls, and multipart target upload paths. | Depends on target calls and target config types through the replication target boundary, metadata paths and metadata systems through the replication metadata boundary, file metadata replication contracts through the filemeta boundary, object decisions and multipart planning through the object decision boundary, resync contracts through the resync boundary, queue DTOs through the queue boundary, error contracts through the error boundary, versioning systems, storage contracts through the replication storage boundary, config-derived storage class labels through the config store, runtime sources, notification events and local event host selection through the event sink, bandwidth reader wrapping, and SetDisks lock timing. |

crates/ecstore/src/bucket/replication/mod.rs

@@ -29,7 +29,6 @@ mod replication_object_bridge;
 mod replication_object_config;
 mod replication_object_decision_boundary;
 pub(crate) mod replication_pool;
-mod replication_proxy;
 mod replication_queue_boundary;
 mod replication_resync_boundary;
 mod replication_resyncer;
@@ -75,7 +74,6 @@ pub use replication_pool::{
     get_global_replication_pool, get_global_replication_stats, init_background_replication, persist_force_delete_intent,
     read_durable_mrf_backlog, resync_start_conflict_id,
 };
-pub use replication_proxy::get_proxy_targets;
 pub use replication_queue_boundary::{
     DeletedObjectReplicationInfo, ReplicationBatchAdmission, ReplicationHealQueueResult, ReplicationOperation,
     ReplicationPriority, ReplicationQueueAdmission,

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -667,368 +667,6 @@ async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
     Err(EcstoreError::PreconditionFailed)
 }
 
-/// Acquires the MRF recovery leader lock for the startup replay.
-/// Returns `None` (after logging) when the lock cannot be created or another
-/// node is already processing the backlog.
-async fn acquire_mrf_recovery_guard<S: ReplicationStorage>(storage: &Arc<S>) -> Option<rustfs_lock::NamespaceLockGuard> {
-    let recovery_lock = match storage
-        .new_ns_lock(
-            ReplicationMetadataStore::rustfs_meta_bucket(),
-            ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
-        )
-        .await
-    {
-        Ok(lock) => lock,
-        Err(error) => {
-            warn!(
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION,
-                error = %error,
-                "Failed to create the MRF recovery leader lock"
-            );
-            return None;
-        }
-    };
-    match recovery_lock
-        .get_write_lock_quiet(ReplicationLockTiming::acquire_timeout())
-        .await
-    {
-        Ok(guard) => Some(guard),
-        Err(_) => {
-            debug!(
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION,
-                "Another node is already processing the MRF recovery backlog"
-            );
-            None
-        }
-    }
-}
-
-/// Reads and decodes the on-disk MRF recovery file.
-/// Returns `None` when there is nothing to replay: missing file (publishes an
-/// empty available summary), read failure, or corrupt data (quarantined).
-async fn load_mrf_recovery_entries<S: ReplicationStorage>(storage: &Arc<S>) -> Option<Vec<MrfReplicateEntry>> {
-    let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
-        Ok(d) => d,
-        Err(EcstoreError::ConfigNotFound) => {
-            set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
-                available: true,
-                buckets: Vec::new(),
-            });
-            return None;
-        }
-        Err(e) => {
-            warn!(
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION,
-                error = %e,
-                "Failed to load MRF recovery file"
-            );
-            return None;
-        }
-    };
-
-    match decode_mrf_file(&data) {
-        Ok(v) => Some(v),
-        Err(e) => {
-            warn!(
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION,
-                error = %e,
-                "Failed to decode MRF recovery file — preserving corrupt data"
-            );
-            quarantine_mrf_file(storage, &data).await;
-            None
-        }
-    }
-}
-
-/// Replays one MRF recovery entry by operation kind.
-/// Returns `None` when the entry is skipped entirely (no admission outcome);
-/// entries that must be retried later are pushed onto `retry_entries`.
-async fn replay_mrf_entry<S: ReplicationStorage>(
-    entry: &MrfReplicateEntry,
-    storage: &Arc<S>,
-    retry_entries: &mut Vec<MrfReplicateEntry>,
-) -> Option<ReplicationQueueAdmission> {
-    match entry.op {
-        MrfOpKind::Delete => replay_mrf_delete_entry(entry, storage, retry_entries).await,
-        MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
-            replay_mrf_object_entry(entry, storage, retry_entries).await
-        }
-        MrfOpKind::Metadata => replay_mrf_metadata_entry(entry, storage, retry_entries).await,
-    }
-}
-
-/// Replays a delete-kind MRF entry: force-delete intents replay directly,
-/// stale force-delete generations are skipped, and plain deletes are
-/// reconstructed as heal deletes.
-async fn replay_mrf_delete_entry<S: ReplicationStorage>(
-    entry: &MrfReplicateEntry,
-    storage: &Arc<S>,
-    retry_entries: &mut Vec<MrfReplicateEntry>,
-) -> Option<ReplicationQueueAdmission> {
-    if should_replay_force_delete_intent(entry) {
-        let operation_id = entry.force_delete_id?;
-        let delete = force_delete_heal_replication_info(entry, operation_id);
-        if replicate_delete_with_outcome(delete, storage.clone()).await {
-            Some(ReplicationQueueAdmission::Queued)
-        } else {
-            Some(ReplicationQueueAdmission::Missed)
-        }
-    } else if entry.force_delete_id.is_some() {
-        Some(ReplicationQueueAdmission::Skipped)
-    } else {
-        replay_mrf_reconstructed_delete(entry, storage, retry_entries).await
-    }
-}
-
-/// Pure DTO construction: heal replication info for a replayed force-delete intent.
-fn force_delete_heal_replication_info(entry: &MrfReplicateEntry, operation_id: uuid::Uuid) -> DeletedObjectReplicationInfo {
-    DeletedObjectReplicationInfo {
-        delete_object: ReplicationDeletedObject {
-            object_name: entry.object.clone(),
-            force_delete: true,
-            force_delete_id: Some(operation_id),
-            force_delete_target_arns: entry.target_arns.clone(),
-            force_delete_generation: entry.force_delete_generation,
-            ..Default::default()
-        },
-        bucket: entry.bucket.clone(),
-        op_type: ReplicationType::Heal,
-        event_type: REPLICATE_HEAL_DELETE.to_string(),
-        ..Default::default()
-    }
-}
-
-/// Reconstruct a heal delete and re-queue it.  We do NOT call
-/// get_object_info here because the delete-marker or version may
-/// already be absent from the local store — that is expected.
-async fn replay_mrf_reconstructed_delete<S: ReplicationStorage>(
-    entry: &MrfReplicateEntry,
-    storage: &Arc<S>,
-    retry_entries: &mut Vec<MrfReplicateEntry>,
-) -> Option<ReplicationQueueAdmission> {
-    let versioned = ReplicationVersioningStore::prefix_enabled(&entry.bucket, &entry.object).await;
-    let oi = ObjectInfo {
-        bucket: entry.bucket.clone(),
-        name: entry.object.clone(),
-        version_id: entry.version_id,
-        delete_marker: entry.delete_marker,
-        ..Default::default()
-    };
-    let dsc = resolve_mrf_delete_replicate_decision(entry, &oi, versioned, retry_entries).await?;
-    let dv = reconstructed_heal_delete_info(entry, &oi, &dsc);
-    if replicate_delete_with_outcome(dv, storage.clone()).await {
-        Some(ReplicationQueueAdmission::Queued)
-    } else {
-        Some(ReplicationQueueAdmission::Missed)
-    }
-}
-
-/// The MRF entry does not persist the replication decision and the
-/// source object is gone, so re-derive the decision from the live
-/// bucket config (mirroring get_heal_replicate_object_info) and set
-/// it on the reconstructed delete. Without this the decision string
-/// is empty and the delete replicates to zero targets — a silent
-/// no-op that leaves replicas diverged (backlog#858 / #799 B9).
-async fn resolve_mrf_delete_replicate_decision(
-    entry: &MrfReplicateEntry,
-    oi: &ObjectInfo,
-    versioned: bool,
-    retry_entries: &mut Vec<MrfReplicateEntry>,
-) -> Option<ReplicateDecision> {
-    if entry.target_arns.is_empty() {
-        match ReplicationMetadataStore::optional_replication_config(&entry.bucket).await {
-            Ok(None) => None,
-            Err(_) => {
-                retry_entries.push(entry.clone());
-                None
-            }
-            Ok(Some(_)) => match check_replicate_delete_strict(
-                &entry.bucket,
-                &ObjectToDelete {
-                    object_name: entry.object.clone(),
-                    version_id: entry.version_id,
-                    ..Default::default()
-                },
-                oi,
-                &ObjectOptions {
-                    versioned,
-                    ..Default::default()
-                },
-                None,
-            )
-            .await
-            {
-                Ok(dsc) => Some(dsc),
-                Err(_) => {
-                    retry_entries.push(entry.clone());
-                    None
-                }
-            },
-        }
-    } else {
-        Some(replicate_decision_for_admitted_targets(&entry.target_arns))
-    }
-}
-
-/// Pure DTO construction: reconstructed heal delete carrying the re-derived
-/// replication decision.
-fn reconstructed_heal_delete_info(
-    entry: &MrfReplicateEntry,
-    oi: &ObjectInfo,
-    dsc: &ReplicateDecision,
-) -> DeletedObjectReplicationInfo {
-    let mut rstate = oi.replication_state();
-    rstate.replicate_decision_str = dsc.to_string();
-
-    let delete_marker_mtime = entry
-        .delete_marker_mtime
-        .and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(i128::from(nanos)).ok());
-
-    DeletedObjectReplicationInfo {
-        delete_object: ReplicationDeletedObject {
-            object_name: entry.object.clone(),
-            version_id: entry.version_id,
-            delete_marker_version_id: entry.delete_marker_version_id,
-            delete_marker: entry.delete_marker,
-            delete_marker_mtime,
-            force_delete: entry.force_delete,
-            replication_state: Some(rstate),
-            ..Default::default()
-        },
-        bucket: entry.bucket.clone(),
-        op_type: ReplicationType::Heal,
-        event_type: REPLICATE_HEAL_DELETE.to_string(),
-        ..Default::default()
-    }
-}
-
-/// Replays an Object/Heal/ExistingObject MRF entry against the live source object.
-async fn replay_mrf_object_entry<S: ReplicationStorage>(
-    entry: &MrfReplicateEntry,
-    storage: &Arc<S>,
-    retry_entries: &mut Vec<MrfReplicateEntry>,
-) -> Option<ReplicationQueueAdmission> {
-    let opts = ObjectOptions {
-        version_id: entry.version_id.map(|u| u.to_string()),
-        ..Default::default()
-    };
-    let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
-        Ok(oi) => oi,
-        Err(e) => {
-            debug!(
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION,
-                bucket = %entry.bucket,
-                object = %entry.object,
-                error = %e,
-                "MRF recovery: source object lookup failed"
-            );
-            if should_retry_mrf_source_lookup(&e) {
-                retry_entries.push(entry.clone());
-            }
-            return None;
-        }
-    };
-    if entry.target_arns.is_empty() {
-        // Legacy entries predate target admission persistence. They cannot
-        // be safely attributed, so retain the old live-config fallback.
-        Some(queue_replication_heal(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
-    } else {
-        let roi = admitted_mrf_replicate_object(oi, entry, entry.op.replication_type());
-        if replicate_object_with_outcome(roi, storage.clone()).await.1 {
-            Some(ReplicationQueueAdmission::Queued)
-        } else {
-            Some(ReplicationQueueAdmission::Missed)
-        }
-    }
-}
-
-/// Replays a metadata-kind MRF entry against the live source object.
-async fn replay_mrf_metadata_entry<S: ReplicationStorage>(
-    entry: &MrfReplicateEntry,
-    storage: &Arc<S>,
-    retry_entries: &mut Vec<MrfReplicateEntry>,
-) -> Option<ReplicationQueueAdmission> {
-    let opts = ObjectOptions {
-        version_id: entry.version_id.map(|u| u.to_string()),
-        ..Default::default()
-    };
-    let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
-        Ok(oi) => oi,
-        Err(e) => {
-            debug!(
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION,
-                bucket = %entry.bucket,
-                object = %entry.object,
-                error = %e,
-                "MRF metadata recovery: source object lookup failed"
-            );
-            if should_retry_mrf_source_lookup(&e) {
-                retry_entries.push(entry.clone());
-            }
-            return None;
-        }
-    };
-    if entry.target_arns.is_empty() {
-        Some(queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
-    } else {
-        let roi = admitted_mrf_replicate_object(oi, entry, ReplicationType::Metadata);
-        if replicate_object_with_outcome(roi, storage.clone()).await.1 {
-            Some(ReplicationQueueAdmission::Queued)
-        } else {
-            Some(ReplicationQueueAdmission::Missed)
-        }
-    }
-}
-
-/// Pure DTO construction: replicate-object info for an entry with persisted
-/// admitted targets, carrying over the entry's retry count.
-fn admitted_mrf_replicate_object(oi: ObjectInfo, entry: &MrfReplicateEntry, op_type: ReplicationType) -> ReplicateObjectInfo {
-    let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
-    let mut roi = replicate_object_info_from_object_info(oi, dsc, op_type);
-    roi.retry_count = entry.retry_count.max(0) as u32;
-    roi
-}
-
-/// Acknowledges the replayed MRF prefix and returns the retained backlog.
-/// On acknowledgement failure the backlog is preserved for the next startup and
-/// re-read (falling back to the replayed snapshot) so the published summary stays accurate.
-async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
-    storage: &Arc<S>,
-    recovery_guard: &rustfs_lock::NamespaceLockGuard,
-    entries: &[MrfReplicateEntry],
-    retry_entries: &[MrfReplicateEntry],
-) -> Vec<MrfReplicateEntry> {
-    match acknowledge_mrf_recovery(storage.clone(), recovery_guard, entries, retry_entries).await {
-        Ok(retained) => retained,
-        Err(error) => {
-            warn!(
-                component = LOG_COMPONENT_ECSTORE,
-                subsystem = LOG_SUBSYSTEM_REPLICATION,
-                error = %error,
-                "Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
-            );
-            match read_mrf_entries(storage.clone()).await {
-                Ok(current) => current,
-                Err(read_error) => {
-                    warn!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        error = %read_error,
-                        "Failed to refresh the MRF backlog after acknowledgement failure"
-                    );
-                    entries.to_vec()
-                }
-            }
-        }
-    }
-}
-
 #[derive(Debug, thiserror::Error)]
 #[error("replication resync {active_resync_id} is already active for {bucket}/{arn}")]
 struct ResyncActiveConflictError {
@@ -1583,12 +1221,71 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         let storage = self.storage.clone();
 
         let handle = tokio::spawn(async move {
-            let Some(recovery_guard) = acquire_mrf_recovery_guard(&storage).await else {
+            let recovery_lock = match storage
-                return;
+                .new_ns_lock(
+                    ReplicationMetadataStore::rustfs_meta_bucket(),
+                    ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
+                )
+                .await
+            {
+                Ok(lock) => lock,
+                Err(error) => {
+                    warn!(
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION,
+                        error = %error,
+                        "Failed to create the MRF recovery leader lock"
+                    );
+                    return;
+                }
+            };
+            let recovery_guard = match recovery_lock
+                .get_write_lock_quiet(ReplicationLockTiming::acquire_timeout())
+                .await
+            {
+                Ok(guard) => guard,
+                Err(_) => {
+                    debug!(
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION,
+                        "Another node is already processing the MRF recovery backlog"
+                    );
+                    return;
+                }
             };
 
-            let Some(entries) = load_mrf_recovery_entries(&storage).await else {
+            let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
-                return;
+                Ok(d) => d,
+                Err(EcstoreError::ConfigNotFound) => {
+                    set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
+                        available: true,
+                        buckets: Vec::new(),
+                    });
+                    return;
+                }
+                Err(e) => {
+                    warn!(
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION,
+                        error = %e,
+                        "Failed to load MRF recovery file"
+                    );
+                    return;
+                }
+            };
+
+            let entries = match decode_mrf_file(&data) {
+                Ok(v) => v,
+                Err(e) => {
+                    warn!(
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION,
+                        error = %e,
+                        "Failed to decode MRF recovery file — preserving corrupt data"
+                    );
+                    quarantine_mrf_file(&storage, &data).await;
+                    return;
+                }
             };
             set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
 
@@ -1597,8 +1294,187 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             let mut retry_entries = Vec::new();
 
             for entry in entries.iter() {
-                let Some(admission) = replay_mrf_entry(entry, &storage, &mut retry_entries).await else {
+                let admission = match entry.op {
-                    continue;
+                    MrfOpKind::Delete => {
+                        if should_replay_force_delete_intent(entry) {
+                            let Some(operation_id) = entry.force_delete_id else {
+                                continue;
+                            };
+                            let delete = DeletedObjectReplicationInfo {
+                                delete_object: ReplicationDeletedObject {
+                                    object_name: entry.object.clone(),
+                                    force_delete: true,
+                                    force_delete_id: Some(operation_id),
+                                    force_delete_target_arns: entry.target_arns.clone(),
+                                    force_delete_generation: entry.force_delete_generation,
+                                    ..Default::default()
+                                },
+                                bucket: entry.bucket.clone(),
+                                op_type: ReplicationType::Heal,
+                                event_type: REPLICATE_HEAL_DELETE.to_string(),
+                                ..Default::default()
+                            };
+                            if replicate_delete_with_outcome(delete, storage.clone()).await {
+                                ReplicationQueueAdmission::Queued
+                            } else {
+                                ReplicationQueueAdmission::Missed
+                            }
+                        } else if entry.force_delete_id.is_some() {
+                            ReplicationQueueAdmission::Skipped
+                        } else {
+                            // Reconstruct a heal delete and re-queue it.  We do NOT call
+                            // get_object_info here because the delete-marker or version may
+                            // already be absent from the local store — that is expected.
+                            //
+                            // The MRF entry does not persist the replication decision and the
+                            // source object is gone, so re-derive the decision from the live
+                            // bucket config (mirroring get_heal_replicate_object_info) and set
+                            // it on the reconstructed delete. Without this the decision string
+                            // is empty and the delete replicates to zero targets — a silent
+                            // no-op that leaves replicas diverged (backlog#858 / #799 B9).
+                            let versioned = ReplicationVersioningStore::prefix_enabled(&entry.bucket, &entry.object).await;
+                            let oi = ObjectInfo {
+                                bucket: entry.bucket.clone(),
+                                name: entry.object.clone(),
+                                version_id: entry.version_id,
+                                delete_marker: entry.delete_marker,
+                                ..Default::default()
+                            };
+                            let dsc = if entry.target_arns.is_empty() {
+                                match ReplicationMetadataStore::optional_replication_config(&entry.bucket).await {
+                                    Ok(None) => continue,
+                                    Err(_) => {
+                                        retry_entries.push(entry.clone());
+                                        continue;
+                                    }
+                                    Ok(Some(_)) => match check_replicate_delete_strict(
+                                        &entry.bucket,
+                                        &ObjectToDelete {
+                                            object_name: entry.object.clone(),
+                                            version_id: entry.version_id,
+                                            ..Default::default()
+                                        },
+                                        &oi,
+                                        &ObjectOptions {
+                                            versioned,
+                                            ..Default::default()
+                                        },
+                                        None,
+                                    )
+                                    .await
+                                    {
+                                        Ok(dsc) => dsc,
+                                        Err(_) => {
+                                            retry_entries.push(entry.clone());
+                                            continue;
+                                        }
+                                    },
+                                }
+                            } else {
+                                replicate_decision_for_admitted_targets(&entry.target_arns)
+                            };
+                            let mut rstate = oi.replication_state();
+                            rstate.replicate_decision_str = dsc.to_string();
+
+                            let delete_marker_mtime = entry
+                                .delete_marker_mtime
+                                .and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(i128::from(nanos)).ok());
+
+                            let dv = DeletedObjectReplicationInfo {
+                                delete_object: ReplicationDeletedObject {
+                                    object_name: entry.object.clone(),
+                                    version_id: entry.version_id,
+                                    delete_marker_version_id: entry.delete_marker_version_id,
+                                    delete_marker: entry.delete_marker,
+                                    delete_marker_mtime,
+                                    force_delete: entry.force_delete,
+                                    replication_state: Some(rstate),
+                                    ..Default::default()
+                                },
+                                bucket: entry.bucket.clone(),
+                                op_type: ReplicationType::Heal,
+                                event_type: REPLICATE_HEAL_DELETE.to_string(),
+                                ..Default::default()
+                            };
+                            if replicate_delete_with_outcome(dv, storage.clone()).await {
+                                ReplicationQueueAdmission::Queued
+                            } else {
+                                ReplicationQueueAdmission::Missed
+                            }
+                        }
+                    }
+                    MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
+                        let opts = ObjectOptions {
+                            version_id: entry.version_id.map(|u| u.to_string()),
+                            ..Default::default()
+                        };
+                        let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
+                            Ok(oi) => oi,
+                            Err(e) => {
+                                debug!(
+                                    component = LOG_COMPONENT_ECSTORE,
+                                    subsystem = LOG_SUBSYSTEM_REPLICATION,
+                                    bucket = %entry.bucket,
+                                    object = %entry.object,
+                                    error = %e,
+                                    "MRF recovery: source object lookup failed"
+                                );
+                                if should_retry_mrf_source_lookup(&e) {
+                                    retry_entries.push(entry.clone());
+                                }
+                                continue;
+                            }
+                        };
+                        if entry.target_arns.is_empty() {
+                            // Legacy entries predate target admission persistence. They cannot
+                            // be safely attributed, so retain the old live-config fallback.
+                            queue_replication_heal(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
+                        } else {
+                            let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
+                            let mut roi = replicate_object_info_from_object_info(oi, dsc, entry.op.replication_type());
+                            roi.retry_count = entry.retry_count.max(0) as u32;
+                            if replicate_object_with_outcome(roi, storage.clone()).await.1 {
+                                ReplicationQueueAdmission::Queued
+                            } else {
+                                ReplicationQueueAdmission::Missed
+                            }
+                        }
+                    }
+                    MrfOpKind::Metadata => {
+                        let opts = ObjectOptions {
+                            version_id: entry.version_id.map(|u| u.to_string()),
+                            ..Default::default()
+                        };
+                        let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
+                            Ok(oi) => oi,
+                            Err(e) => {
+                                debug!(
+                                    component = LOG_COMPONENT_ECSTORE,
+                                    subsystem = LOG_SUBSYSTEM_REPLICATION,
+                                    bucket = %entry.bucket,
+                                    object = %entry.object,
+                                    error = %e,
+                                    "MRF metadata recovery: source object lookup failed"
+                                );
+                                if should_retry_mrf_source_lookup(&e) {
+                                    retry_entries.push(entry.clone());
+                                }
+                                continue;
+                            }
+                        };
+                        if entry.target_arns.is_empty() {
+                            queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
+                        } else {
+                            let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
+                            let mut roi = replicate_object_info_from_object_info(oi, dsc, ReplicationType::Metadata);
+                            roi.retry_count = entry.retry_count.max(0) as u32;
+                            if replicate_object_with_outcome(roi, storage.clone()).await.1 {
+                                ReplicationQueueAdmission::Queued
+                            } else {
+                                ReplicationQueueAdmission::Missed
+                            }
+                        }
+                    }
                 };
 
                 if admission == ReplicationQueueAdmission::Missed {
@@ -1608,7 +1484,29 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                 }
             }
 
-            let retained = resolve_retained_mrf_entries(&storage, &recovery_guard, &entries, &retry_entries).await;
+            let retained = match acknowledge_mrf_recovery(storage.clone(), &recovery_guard, &entries, &retry_entries).await {
+                Ok(retained) => retained,
+                Err(error) => {
+                    warn!(
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION,
+                        error = %error,
+                        "Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
+                    );
+                    match read_mrf_entries(storage.clone()).await {
+                        Ok(current) => current,
+                        Err(read_error) => {
+                            warn!(
+                                component = LOG_COMPONENT_ECSTORE,
+                                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                                error = %read_error,
+                                "Failed to refresh the MRF backlog after acknowledgement failure"
+                            );
+                            entries.clone()
+                        }
+                    }
+                }
+            };
             let retained_count = retained.len();
             set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&retained));
 

crates/ecstore/src/bucket/replication/replication_proxy.rs

@@ -1,150 +0,0 @@
-// 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.
-
-//! Proxy-target selection for reads of objects not yet replicated locally
-//! (MinIO `getProxyTargets`, bucket-replication.go).
-//!
-//! During the active-active replication lag window a GET/HEAD/Tagging request
-//! for an object the local site does not have yet may be served by proxying to
-//! a replication target. This module only *selects* the candidate targets; the
-//! request-path callers perform the remote calls and response translation.
-
-use std::sync::Arc;
-
-use tracing::debug;
-
-use super::replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt as _};
-use super::replication_object_config::get_replication_config;
-use super::replication_storage_boundary::ObjectOptions;
-use super::replication_target_boundary::{ReplicationTargetStore, TargetClient};
-
-/// Returns the replication-target clients eligible to serve a proxied read of
-/// `bucket/object`, in rule order. Mirrors MinIO's `getProxyTargets`:
-///
-/// - the `source-proxy-request` header family was present at all
-///   (`opts.proxy_request` / `opts.proxy_header_set`, MinIO `ProxyRequest` /
-///   `ProxyHeaderSet`) -> empty. "true" is the anti-loop marker of an
-///   already-proxied client read; "false" is what a peer's replication
-///   worker sends on convergence HEADs so the receiver answers locally —
-///   proxying that miss back would echo the source object and fake
-///   convergence, permanently skipping replication;
-/// - the bucket's versioning is suspended for the object -> empty;
-/// - no replication configuration / no matching rule -> empty;
-/// - otherwise every distinct target ARN whose rules match the object,
-///   resolved through the bucket target system, skipping targets that opted
-///   out of proxying (`disable_proxy`).
-pub async fn get_proxy_targets(bucket: &str, object: &str, opts: &ObjectOptions) -> Vec<Arc<TargetClient>> {
-    if opts.proxy_request || opts.proxy_header_set {
-        return Vec::new();
-    }
-    if opts.version_suspended {
-        return Vec::new();
-    }
-
-    let cfg = match get_replication_config(bucket).await {
-        Ok(Some(cfg)) => cfg,
-        Ok(None) => return Vec::new(),
-        Err(err) => {
-            debug!(bucket, object, error = %err, "read proxy: failed to load replication config; not proxying");
-            return Vec::new();
-        }
-    };
-
-    let arns = cfg.filter_target_arns(&ObjectOpts {
-        name: object.to_string(),
-        ..Default::default()
-    });
-
-    let mut targets = Vec::with_capacity(arns.len());
-    for arn in arns {
-        let Some(client) = ReplicationTargetStore::remote_target_client(bucket, &arn).await else {
-            debug!(bucket, object, arn, "read proxy: no client for replication target ARN");
-            continue;
-        };
-        if client.disable_proxy {
-            continue;
-        }
-        targets.push(client);
-    }
-
-    targets
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    fn opts() -> ObjectOptions {
-        ObjectOptions::default()
-    }
-
-    /// Anti-loop: a request that was already proxied by a peer must never be
-    /// proxied onward, regardless of replication configuration.
-    #[tokio::test]
-    async fn proxy_request_yields_no_targets() {
-        let targets = get_proxy_targets(
-            "bucket",
-            "object",
-            &ObjectOptions {
-                proxy_request: true,
-                ..opts()
-            },
-        )
-        .await;
-        assert!(targets.is_empty());
-    }
-
-    /// MinIO `ProxyHeaderSet` parity: the header family being present at all
-    /// disables proxying, even with the value "false" — that is what a
-    /// peer's replication worker sends on convergence HEADs.
-    #[tokio::test]
-    async fn proxy_header_set_yields_no_targets() {
-        let targets = get_proxy_targets(
-            "bucket",
-            "object",
-            &ObjectOptions {
-                proxy_header_set: true,
-                proxy_request: false,
-                ..opts()
-            },
-        )
-        .await;
-        assert!(targets.is_empty());
-    }
-
-    /// Suspended versioning disables proxying (MinIO parity): the local null
-    /// version is authoritative and a remote read could resurrect data.
-    #[tokio::test]
-    async fn version_suspended_yields_no_targets() {
-        let targets = get_proxy_targets(
-            "bucket",
-            "object",
-            &ObjectOptions {
-                version_suspended: true,
-                ..opts()
-            },
-        )
-        .await;
-        assert!(targets.is_empty());
-    }
-
-    /// A bucket without replication configuration has nothing to proxy to.
-    /// (No metadata system is running in unit tests, so the config lookup
-    /// resolves to "no configuration" — the same empty-result contract.)
-    #[tokio::test]
-    async fn missing_replication_config_yields_no_targets() {
-        let targets = get_proxy_targets("bucket-without-replication", "object", &opts()).await;
-        assert!(targets.is_empty());
-    }
-}

crates/ecstore/src/bucket/replication/replication_resync_boundary.rs

@@ -15,14 +15,10 @@
 use super::replication_error_boundary::{Error, Result};
 use super::replication_filemeta_boundary::MrfReplicateEntry;
 
-/// Kept test-only: the runtime consumer was the worker HEAD's fake proxy
-/// counting (removed in backlog#1675 P1-5); the resyncer tests still pin the
-/// classifier's semantics for the real client read-proxy failure accounting.
-#[cfg(test)]
-pub(crate) use rustfs_replication::should_count_head_proxy_failure;
 pub use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus};
 pub(crate) use rustfs_replication::{
     is_version_id_mismatch, resync_state_accepts_update, sanitize_resync_error_detail, should_auto_resume_resync,
+    should_count_head_proxy_failure,
 };
 
 #[allow(

crates/ecstore/src/bucket/replication/replication_state.rs

@@ -1161,31 +1161,6 @@ mod tests {
         assert!(all.contains_key("proxy-only-bucket"));
     }
 
-    /// Pins the read-proxy metric contract (backlog#1675 P1-5): the API
-    /// strings the GET/HEAD/Tagging proxy paths record map onto the
-    /// get/head/tagging totals, and only unexpected failures raise the
-    /// failed counters.
-    #[tokio::test]
-    async fn test_proxy_stats_map_read_proxy_apis_to_totals() {
-        let stats = ReplicationStats::new();
-        stats.inc_proxy("proxy-bucket", "GetObject", false).await;
-        stats.inc_proxy("proxy-bucket", "GetObject", true).await;
-        stats.inc_proxy("proxy-bucket", "HeadObject", false).await;
-        stats.inc_proxy("proxy-bucket", "GetObjectTagging", false).await;
-        stats.inc_proxy("proxy-bucket", "PutObjectTagging", false).await;
-        stats.inc_proxy("proxy-bucket", "DeleteObjectTagging", true).await;
-
-        let metric = stats.get_proxy_stats("proxy-bucket").await;
-        assert_eq!(metric.get_total, 2);
-        assert_eq!(metric.get_failed, 1);
-        assert_eq!(metric.head_total, 1);
-        assert_eq!(metric.head_failed, 0);
-        assert_eq!(metric.get_tag_total, 1);
-        assert_eq!(metric.put_tag_total, 1);
-        assert_eq!(metric.delete_tag_total, 1);
-        assert_eq!(metric.delete_tag_failed, 1);
-    }
-
     #[tokio::test]
     async fn test_calculate_bucket_replication_stats_merges_resync_metrics() {
         let stats = ReplicationStats::new();

crates/ecstore/src/object_api/types.rs

@@ -277,20 +277,6 @@ pub struct ObjectOptions {
     /// fence avoids recursively acquiring the read lock behind a queued writer.
     pub bucket_lifecycle_lock_fence: Option<NamespaceLockFence>,
     pub replication_request: bool,
-    /// True when the inbound request carried the
-    /// `{x-rustfs-,x-minio-}source-proxy-request` header family with the
-    /// value "true": the request was already proxied by a replication peer,
-    /// so this server must not proxy a local miss onward (anti-loop,
-    /// MinIO-compatible). The header only disables proxying — it grants no
-    /// capability — so no authorization gate is required to honor it.
-    pub proxy_request: bool,
-    /// True when the `source-proxy-request` header family was present at
-    /// all, regardless of value (MinIO's `ProxyHeaderSet`). A replication
-    /// peer sends `source-proxy-request: false` on its worker convergence
-    /// HEADs precisely so the receiver answers locally instead of proxying
-    /// back — otherwise a proxied 404->200 echo makes the worker believe the
-    /// object already converged and it never replicates it.
-    pub proxy_header_set: bool,
     /// Source-cluster LWW timestamps carried by an authorized replication
     /// request; None when the source never modified the category. Only the
     /// replication-authorized options builders may set these.

crates/kms/src/api_types.rs

@@ -293,15 +293,6 @@ enum StrictVaultAuthMethod {
         #[serde(default)]
         refresh_safety_window_secs: Option<u64>,
     },
-    Kubernetes {
-        role: String,
-        #[serde(default)]
-        mount: Option<String>,
-        #[serde(default)]
-        jwt_path: Option<std::path::PathBuf>,
-        #[serde(default)]
-        refresh_safety_window_secs: Option<u64>,
-    },
     TokenFile {
         path: std::path::PathBuf,
         #[serde(default)]
@@ -328,17 +319,6 @@ impl From<StrictVaultAuthMethod> for VaultAuthMethod {
                 mount: mount.unwrap_or_else(|| crate::config::DEFAULT_VAULT_APPROLE_MOUNT.to_string()),
                 refresh_safety_window_secs,
             },
-            StrictVaultAuthMethod::Kubernetes {
-                role,
-                mount,
-                jwt_path,
-                refresh_safety_window_secs,
-            } => Self::Kubernetes {
-                role,
-                mount: mount.unwrap_or_else(|| crate::config::DEFAULT_VAULT_KUBERNETES_MOUNT.to_string()),
-                jwt_path: jwt_path.unwrap_or_else(|| std::path::PathBuf::from(crate::config::DEFAULT_VAULT_KUBERNETES_JWT_PATH)),
-                refresh_safety_window_secs,
-            },
             StrictVaultAuthMethod::TokenFile {
                 path,
                 poll_interval_secs,
@@ -519,7 +499,6 @@ impl From<&KmsConfig> for KmsConfigSummary {
                 auth_method_type: match &vault_config.auth_method {
                     VaultAuthMethod::Token { .. } => "token".to_string(),
                     VaultAuthMethod::AppRole { .. } => "approle".to_string(),
-                    VaultAuthMethod::Kubernetes { .. } => "kubernetes".to_string(),
                     VaultAuthMethod::TokenFile { .. } => "token_file".to_string(),
                 },
                 has_stored_credentials: true,
@@ -534,7 +513,6 @@ impl From<&KmsConfig> for KmsConfigSummary {
                 auth_method_type: match &vault_config.auth_method {
                     VaultAuthMethod::Token { .. } => "token".to_string(),
                     VaultAuthMethod::AppRole { .. } => "approle".to_string(),
-                    VaultAuthMethod::Kubernetes { .. } => "kubernetes".to_string(),
                     VaultAuthMethod::TokenFile { .. } => "token_file".to_string(),
                 },
                 has_stored_credentials: true,
@@ -923,42 +901,6 @@ mod tests {
         assert!(request.to_kms_config().validate().is_ok());
     }
 
-    /// The admin API reaches Kubernetes auth with the role alone; the mount and
-    /// the projected token path fall back to the cluster defaults, so a Tenant
-    /// manifest carries no credential and no cluster-specific paths.
-    #[test]
-    fn test_deserialize_vault_configure_request_accepts_kubernetes_auth() {
-        let raw = serde_json::json!({
-            "backend_type": "vault-transit",
-            "address": "https://vault.example.com:8200",
-            "mount_path": "rustfs",
-            "auth_method": { "Kubernetes": { "role": "rustfs" } }
-        });
-
-        let request: ConfigureKmsRequest = serde_json::from_value(raw).expect("kubernetes auth should deserialize");
-        let config = request.to_kms_config();
-        config.validate().expect("kubernetes auth must validate");
-
-        let vault = config.vault_transit_config().expect("vault transit backend config");
-        let VaultAuthMethod::Kubernetes {
-            role, mount, jwt_path, ..
-        } = &vault.auth_method
-        else {
-            panic!("expected Kubernetes auth, got {:?}", vault.auth_method);
-        };
-        assert_eq!(role, "rustfs");
-        assert_eq!(mount, crate::config::DEFAULT_VAULT_KUBERNETES_MOUNT);
-        assert_eq!(jwt_path, std::path::Path::new(crate::config::DEFAULT_VAULT_KUBERNETES_JWT_PATH));
-
-        let unknown_field = serde_json::json!({
-            "backend_type": "vault-transit",
-            "address": "https://vault.example.com:8200",
-            "auth_method": { "Kubernetes": { "role": "rustfs", "service_account": "rustfs" } }
-        });
-        serde_json::from_value::<ConfigureKmsRequest>(unknown_field)
-            .expect_err("an unknown auth field must be rejected rather than silently dropped");
-    }
-
     #[test]
     fn test_deserialize_aws_configure_request_accepts_type_aliases() {
         for backend_type in ["AWS", "AwsKms", "aws", "aws-kms", "aws_kms"] {

crates/kms/src/backends/vault.rs

@@ -550,7 +550,6 @@ impl VaultKmsClient {
             address: config.address.clone(),
             namespace: config.namespace.clone(),
             attempt_timeout: kms_config.effective_timeout(),
-            skip_tls_verify: config.tls.as_ref().is_some_and(|tls| tls.skip_verify),
         };
         let source = token_source_for(&config.auth_method, &settings)?;
         let policy = VaultCredentialPolicy::from_kms_config(

crates/kms/src/backends/vault_credentials.rs

@@ -326,97 +326,6 @@ impl fmt::Debug for AppRoleLogin {
     }
 }
 
-/// Token source for [`VaultAuthMethod::Kubernetes`]: exchanges the pod's
-/// projected ServiceAccount token for a lease-bound Vault token.
-///
-/// The JWT is re-read on every login because the kubelet rotates a projected
-/// token well inside the pod's lifetime; caching it would strand the source on
-/// an expired assertion once the current Vault token can no longer be renewed.
-///
-/// Unlike [`TokenFileSource`], the file mode is not checked: the kubelet owns
-/// the projected token and mounts it world-readable by default, so rejecting
-/// group/other bits would refuse every standard pod rather than catch a
-/// deployment error.
-pub(crate) struct KubernetesLogin {
-    /// Unauthenticated client used only for the login exchange.
-    login_client: VaultClient,
-    mount: String,
-    role: String,
-    jwt_path: PathBuf,
-}
-
-impl KubernetesLogin {
-    pub(crate) fn new(settings: &VaultConnectionSettings, mount: String, role: String, jwt_path: PathBuf) -> Result<Self> {
-        Ok(Self {
-            login_client: settings.build_login_client()?,
-            mount,
-            role,
-            jwt_path,
-        })
-    }
-
-    /// Read the ServiceAccount token for one login attempt.
-    ///
-    /// Mirrors [`AppRoleLogin::resolve_secret_id`]: a read failure is fatal for
-    /// the attempt but the refresh loop keeps retrying, so a token the kubelet
-    /// has not projected yet heals the source without a restart.
-    async fn resolve_jwt(&self) -> AttemptResult<SecretString> {
-        let mut raw = tokio::fs::read_to_string(&self.jwt_path)
-            .await
-            .map_err(|error| AttemptError {
-                class: ErrorClass::Fatal,
-                error: KmsError::configuration_error(format!(
-                    "Failed to read Kubernetes ServiceAccount token {}: {error}",
-                    self.jwt_path.display()
-                )),
-            })?;
-        let trimmed = raw.trim();
-        if trimmed.is_empty() {
-            raw.zeroize();
-            return Err(AttemptError {
-                class: ErrorClass::Fatal,
-                error: KmsError::configuration_error(format!(
-                    "Kubernetes ServiceAccount token {} is empty",
-                    self.jwt_path.display()
-                )),
-            });
-        }
-        let jwt = SecretString::new(trimmed.to_string());
-        raw.zeroize();
-        Ok(jwt)
-    }
-}
-
-#[async_trait]
-impl TokenSource for KubernetesLogin {
-    async fn acquire(&self) -> AttemptResult<TokenLease> {
-        let jwt = self.resolve_jwt().await?;
-        let auth = vaultrs::auth::kubernetes::login(&self.login_client, &self.mount, &self.role, jwt.expose())
-            .await
-            .map_err(|error| attempt_error("Kubernetes login", error))?;
-        Ok(TokenLease::from_auth(auth))
-    }
-
-    async fn renew(&self, client: &VaultClient) -> AttemptResult<TokenLease> {
-        let auth = vaultrs::token::renew_self(client, None)
-            .await
-            .map_err(|error| attempt_error("token renewal", error))?;
-        Ok(TokenLease::from_auth(auth))
-    }
-}
-
-impl fmt::Debug for KubernetesLogin {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        // The login client embeds Vault client settings and must stay out of
-        // Debug output; the role name is not a secret, and the JWT is never held.
-        f.debug_struct("KubernetesLogin")
-            .field("mount", &self.mount)
-            .field("role", &self.role)
-            .field("jwt_path", &self.jwt_path)
-            .finish_non_exhaustive()
-    }
-}
-
 /// Token source for [`VaultAuthMethod::TokenFile`]: reads an agent-managed
 /// token file (for example a Vault Agent auto-auth sink).
 ///
@@ -555,9 +464,6 @@ pub(crate) fn token_source_for(
             secret_id.clone(),
             secret_id_file.clone(),
         )?)),
-        VaultAuthMethod::Kubernetes {
-            role, mount, jwt_path, ..
-        } => Ok(Box::new(KubernetesLogin::new(settings, mount.clone(), role.clone(), jwt_path.clone())?)),
         VaultAuthMethod::TokenFile {
             path,
             poll_interval_secs,
@@ -580,9 +486,6 @@ pub(crate) struct VaultConnectionSettings {
     pub(crate) namespace: Option<String>,
     /// Per-attempt HTTP timeout applied to the underlying reqwest client.
     pub(crate) attempt_timeout: Duration,
-    /// Whether to accept an unverified Vault server certificate. Gated on
-    /// `allow_insecure_dev_defaults` by `KmsConfig::validate`.
-    pub(crate) skip_tls_verify: bool,
 }
 
 impl VaultConnectionSettings {
@@ -596,11 +499,6 @@ impl VaultConnectionSettings {
         // operation-level retry policy.
         settings_builder.timeout(Some(self.attempt_timeout));
         settings_builder.token(token);
-        // Always set explicitly: left unset, vaultrs derives this from its own
-        // VAULT_SKIP_VERIFY variable, so a stray value in the environment would
-        // disable certificate verification behind the KMS configuration and its
-        // insecure-defaults gate.
-        settings_builder.verify(!self.skip_tls_verify);
 
         if let Some(namespace) = &self.namespace {
             settings_builder.namespace(Some(namespace.clone()));
@@ -653,10 +551,6 @@ impl VaultCredentialPolicy {
                 refresh_safety_window_secs: Some(secs),
                 ..
             }
-            | VaultAuthMethod::Kubernetes {
-                refresh_safety_window_secs: Some(secs),
-                ..
-            }
             | VaultAuthMethod::TokenFile {
                 refresh_safety_window_secs: Some(secs),
                 ..
@@ -690,25 +584,15 @@ pub(crate) struct VaultClientHandle {
 
 impl VaultClientHandle {
     /// Absolute expiry of this generation's token.
-    ///
-    /// `lease.ttl` is built from the `lease_duration` the Vault server sent, so
-    /// a value too large to add to `issued_at` would panic on the bare `+`. A
-    /// TTL that cannot be represented is indistinguishable from no expiry, so it
-    /// collapses to `None` — the same answer already given for the zero-lease
-    /// tokens Vault issues, which keeps the token in use and still fully
-    /// validated by Vault on every call.
     fn expires_at(&self) -> Option<Instant> {
-        self.lease.and_then(|lease| self.issued_at.checked_add(lease.ttl))
+        self.lease.map(|lease| self.issued_at + lease.ttl)
     }
 
     /// When the renewal task should refresh this generation: half the TTL,
     /// leaving the second half as budget for retries before the fail-closed
     /// window is reached.
-    ///
-    /// Unrepresentable TTLs collapse to `None` as in [`Self::expires_at`],
-    /// leaving a token that never expires with nothing to renew.
     fn renew_at(&self) -> Option<Instant> {
-        self.lease.and_then(|lease| self.issued_at.checked_add(lease.ttl / 2))
+        self.lease.map(|lease| self.issued_at + lease.ttl / 2)
     }
 }
 
@@ -778,7 +662,7 @@ impl VaultCredentialProvider {
         let handle = self.current.load_full();
         if let Some(expires_at) = handle.expires_at() {
             let now = Instant::now();
-            if self.inside_safety_window(now, expires_at) {
+            if now + self.policy.safety_window >= expires_at {
                 return Err(KmsError::credentials_unavailable(format!(
                     "Vault token (generation {}) is within {:?} of expiry and has not been refreshed; refusing to use it",
                     handle.generation, self.policy.safety_window
@@ -788,18 +672,6 @@ impl VaultCredentialProvider {
         Ok(handle)
     }
 
-    /// Whether the token expiring at `expires_at` is close enough to refuse.
-    ///
-    /// `safety_window` reaches here from persisted configuration, so it is not
-    /// guaranteed to have passed this version's validation: a window too large
-    /// to add to the current instant would panic on the bare `+`. Such a window
-    /// means every token is always inside it, so saturating to "refuse" is both
-    /// the fail-closed answer and the one the arithmetic was reaching for.
-    fn inside_safety_window(&self, now: Instant, expires_at: Instant) -> bool {
-        now.checked_add(self.policy.safety_window)
-            .is_none_or(|deadline| deadline >= expires_at)
-    }
-
     /// Publish the credential gauges for the generation currently installed.
     ///
     /// The fail-closed gauge re-evaluates the very gate
@@ -811,7 +683,7 @@ impl VaultCredentialProvider {
         let fail_closed = match handle.expires_at() {
             Some(expires_at) => {
                 metrics::gauge!(METRIC_TOKEN_TTL_SECONDS).set(expires_at.saturating_duration_since(now).as_secs_f64());
-                self.inside_safety_window(now, expires_at)
+                now + self.policy.safety_window >= expires_at
             }
             // A generation without an expiry has no remaining TTL to report
             // and can never lapse, so it can never fail closed either.
@@ -988,7 +860,7 @@ impl Drop for CredentialTaskHandle {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::config::{DEFAULT_VAULT_KUBERNETES_MOUNT, REDACTED_SECRET};
+    use crate::config::REDACTED_SECRET;
     use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 
     const TEST_TOKEN: &str = "vault-token-debug-leak-canary";
@@ -999,7 +871,6 @@ mod tests {
             address: "http://127.0.0.1:8200".to_string(),
             namespace: Some("team-namespace".to_string()),
             attempt_timeout: Duration::from_secs(30),
-            skip_tls_verify: false,
         }
     }
 
@@ -1186,143 +1057,6 @@ mod tests {
         assert!(format!("{source:?}").contains("AppRoleLogin"));
     }
 
-    #[tokio::test]
-    async fn test_kubernetes_auth_method_maps_to_login_source() {
-        let settings = test_settings();
-        let source = token_source_for(&VaultAuthMethod::kubernetes("rustfs".to_string()), &settings)
-            .expect("kubernetes auth must map to a login source");
-
-        assert!(format!("{source:?}").contains("KubernetesLogin"));
-    }
-
-    /// `refresh_safety_window_secs` is operator-supplied and reaches the request
-    /// path from persisted configuration, so the fail-closed comparison must
-    /// survive a window too large to add to the current instant. Before the
-    /// checked arithmetic this panicked with "overflow when adding duration to
-    /// instant" on the first request after a lease-bearing login.
-    #[tokio::test]
-    async fn test_current_refuses_rather_than_panics_on_an_unrepresentable_safety_window() {
-        let (provider, _state) = scripted_provider(
-            Duration::from_secs(60),
-            true,
-            test_policy(Duration::from_secs(u64::MAX), Duration::from_secs(5)),
-        )
-        .await;
-
-        let error = provider
-            .current()
-            .expect_err("a window wider than any lease must refuse the token");
-        assert!(
-            matches!(error, KmsError::CredentialsUnavailable { .. }),
-            "expected CredentialsUnavailable, got {error:?}"
-        );
-    }
-
-    /// `lease_duration` is a bare u64 straight off the Vault response and forms
-    /// the other side of the same comparison, so an absurd one must not panic
-    /// either. It is indistinguishable from a non-expiring token, which is how
-    /// the zero-lease case already behaves.
-    #[tokio::test]
-    async fn test_an_unrepresentable_lease_is_treated_as_non_expiring() {
-        let (provider, _state) = scripted_provider(
-            Duration::from_secs(u64::MAX),
-            true,
-            test_policy(Duration::from_secs(30), Duration::from_secs(5)),
-        )
-        .await;
-
-        provider
-            .current()
-            .expect("a token whose expiry cannot be represented must stay usable");
-    }
-
-    /// The configured flag has to reach the HTTP client, not just the config
-    /// struct: every generation (authenticated and login) builds its own client,
-    /// and a Vault with a self-signed certificate fails the handshake unless
-    /// each one carries the setting.
-    #[test]
-    fn test_skip_tls_verify_reaches_every_vault_client_generation() {
-        for skip_tls_verify in [false, true] {
-            let settings = VaultConnectionSettings {
-                address: "https://vault.example.com:8200".to_string(),
-                namespace: None,
-                attempt_timeout: Duration::from_secs(30),
-                skip_tls_verify,
-            };
-
-            let authenticated = settings.build_client(TEST_TOKEN).expect("authenticated client must build");
-            assert_eq!(authenticated.settings.verify, !skip_tls_verify);
-
-            let login = settings.build_login_client().expect("login client must build");
-            assert_eq!(login.settings.verify, !skip_tls_verify);
-        }
-    }
-
-    /// vaultrs derives `verify` from its own VAULT_SKIP_VERIFY variable when the
-    /// builder leaves it unset, which would disable certificate verification
-    /// without passing the KMS insecure-defaults gate.
-    #[test]
-    fn test_vaultrs_skip_verify_env_cannot_override_the_configured_setting() {
-        temp_env::with_var("VAULT_SKIP_VERIFY", Some("true"), || {
-            let client = test_settings().build_client(TEST_TOKEN).expect("client must build");
-            assert!(
-                client.settings.verify,
-                "a stray VAULT_SKIP_VERIFY must not disable verification behind the KMS configuration"
-            );
-        });
-    }
-
-    /// The projected token is read fresh per login attempt and trimmed, so a
-    /// kubelet rotation is picked up without a restart and a trailing newline
-    /// does not corrupt the assertion sent to Vault.
-    #[tokio::test]
-    async fn test_kubernetes_login_rereads_and_trims_the_service_account_token() {
-        let dir = tempfile::tempdir().expect("temp dir");
-        let path = dir.path().join("token");
-        tokio::fs::write(&path, "  first-jwt\n").await.expect("write token");
-
-        let login = KubernetesLogin::new(
-            &test_settings(),
-            DEFAULT_VAULT_KUBERNETES_MOUNT.to_string(),
-            "rustfs".to_string(),
-            path.clone(),
-        )
-        .expect("login source must build");
-
-        assert_eq!(login.resolve_jwt().await.expect("first read").expose(), "first-jwt");
-
-        tokio::fs::write(&path, "rotated-jwt").await.expect("rotate token");
-        assert_eq!(
-            login.resolve_jwt().await.expect("second read").expose(),
-            "rotated-jwt",
-            "a rotated projected token must be picked up without a restart"
-        );
-    }
-
-    /// The ServiceAccount token is re-read per attempt, so an unreadable or
-    /// empty one fails that attempt without reaching Vault; the refresh loop
-    /// keeps retrying, which is what lets a late projection heal the source.
-    #[tokio::test]
-    async fn test_kubernetes_login_rejects_an_unusable_service_account_token() {
-        let dir = tempfile::tempdir().expect("temp dir");
-        let missing = dir.path().join("absent-token");
-        let empty = dir.path().join("empty-token");
-        tokio::fs::write(&empty, "  \n").await.expect("write empty token");
-
-        for (path, expected) in [(missing, "Failed to read"), (empty, "is empty")] {
-            let login =
-                KubernetesLogin::new(&test_settings(), DEFAULT_VAULT_KUBERNETES_MOUNT.to_string(), "rustfs".to_string(), path)
-                    .expect("login source must build");
-
-            let error = login
-                .acquire()
-                .await
-                .expect_err("an unusable ServiceAccount token must fail the attempt");
-            assert!(matches!(error.class, ErrorClass::Fatal));
-            assert!(error.error.to_string().contains(expected), "got {}", error.error);
-        }
-    }
-
     #[tokio::test(start_paused = true)]
     async fn test_renewal_task_renews_at_half_ttl() {
         let (provider, state) = scripted_provider(

crates/kms/src/backends/vault_transit.rs

@@ -415,7 +415,6 @@ impl VaultTransitKmsClient {
             address: config.address.clone(),
             namespace: config.namespace.clone(),
             attempt_timeout: kms_config.effective_timeout(),
-            skip_tls_verify: config.tls.as_ref().is_some_and(|tls| tls.skip_verify),
         };
         let source = token_source_for(&config.auth_method, &settings)?;
         let policy = VaultCredentialPolicy::from_kms_config(

crates/kms/src/backup/vault_restore.rs

@@ -450,10 +450,6 @@ impl VaultRestoreClient {
             address: target.address.clone(),
             namespace: target.namespace.clone(),
             attempt_timeout: kms_config.effective_timeout(),
-            // A restore target carries no TLS settings, so certificates are
-            // always verified: recovery is the last path that should accept an
-            // unauthenticated Vault.
-            skip_tls_verify: false,
         };
         let source = token_source_for(&target.auth_method, &settings)?;
         let policy = VaultCredentialPolicy::from_kms_config(

crates/kms/src/config.rs

@@ -25,10 +25,6 @@ use url::Url;
 
 pub const ENV_KMS_ALLOW_INSECURE_DEV_DEFAULTS: &str = "RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS";
 pub const ENV_KMS_ALLOW_IMMEDIATE_DELETION: &str = "RUSTFS_KMS_ALLOW_IMMEDIATE_DELETION";
-pub const ENV_KMS_VAULT_ADDRESS: &str = "RUSTFS_KMS_VAULT_ADDRESS";
-pub const ENV_KMS_VAULT_TOKEN: &str = "RUSTFS_KMS_VAULT_TOKEN";
-pub const ENV_KMS_VAULT_NAMESPACE: &str = "RUSTFS_KMS_VAULT_NAMESPACE";
-pub const ENV_KMS_VAULT_MOUNT_PATH: &str = "RUSTFS_KMS_VAULT_MOUNT_PATH";
 pub const ENV_KMS_VAULT_SKIP_TLS_VERIFY: &str = "RUSTFS_KMS_VAULT_SKIP_TLS_VERIFY";
 pub const ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "RUSTFS_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT";
 pub const ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX: &str = "RUSTFS_KMS_VAULT_TRANSIT_METADATA_PREFIX";
@@ -39,9 +35,6 @@ pub const ENV_KMS_VAULT_APPROLE_SECRET_ID: &str = "RUSTFS_KMS_VAULT_APPROLE_SECR
 pub const ENV_KMS_VAULT_APPROLE_SECRET_ID_FILE: &str = "RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE";
 pub const ENV_KMS_VAULT_APPROLE_MOUNT: &str = "RUSTFS_KMS_VAULT_APPROLE_MOUNT";
 pub const ENV_KMS_VAULT_TOKEN_FILE: &str = "RUSTFS_KMS_VAULT_TOKEN_FILE";
-pub const ENV_KMS_VAULT_KUBERNETES_ROLE: &str = "RUSTFS_KMS_VAULT_KUBERNETES_ROLE";
-pub const ENV_KMS_VAULT_KUBERNETES_MOUNT: &str = "RUSTFS_KMS_VAULT_KUBERNETES_MOUNT";
-pub const ENV_KMS_VAULT_KUBERNETES_JWT_PATH: &str = "RUSTFS_KMS_VAULT_KUBERNETES_JWT_PATH";
 pub const ENV_KMS_AWS_REGION: &str = "RUSTFS_KMS_AWS_REGION";
 pub const ENV_KMS_AWS_ENDPOINT_URL: &str = "RUSTFS_KMS_AWS_ENDPOINT_URL";
 /// Age in whole seconds beyond which a key is reported as due for rotation;
@@ -52,9 +45,6 @@ pub const ENV_KMS_ROTATION_MAX_WRAPS: &str = "RUSTFS_KMS_ROTATION_MAX_WRAPS";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "secret";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX: &str = "rustfs/kms/transit-metadata";
 pub const DEFAULT_VAULT_APPROLE_MOUNT: &str = "approle";
-pub const DEFAULT_VAULT_KUBERNETES_MOUNT: &str = "kubernetes";
-/// Where the kubelet projects a pod's ServiceAccount token by default.
-pub const DEFAULT_VAULT_KUBERNETES_JWT_PATH: &str = "/var/run/secrets/kubernetes.io/serviceaccount/token";
 
 /// Upper bound applied to `KmsConfig::timeout` when deriving backend behavior.
 ///
@@ -94,14 +84,6 @@ fn default_vault_approle_mount() -> String {
     DEFAULT_VAULT_APPROLE_MOUNT.to_string()
 }
 
-fn default_vault_kubernetes_mount() -> String {
-    DEFAULT_VAULT_KUBERNETES_MOUNT.to_string()
-}
-
-fn default_vault_kubernetes_jwt_path() -> PathBuf {
-    PathBuf::from(DEFAULT_VAULT_KUBERNETES_JWT_PATH)
-}
-
 pub const KMS_CONFIG_REDACTION_RULES: &[RedactionRule] = &[
     RedactionRule::new("kms.local.master_key", RedactionLevel::Secret, "local backend key encryption material"),
     RedactionRule::new("kms.vault.token", RedactionLevel::Secret, "vault authentication token"),
@@ -508,23 +490,6 @@ pub enum VaultAuthMethod {
         #[serde(default)]
         refresh_safety_window_secs: Option<u64>,
     },
-    /// Kubernetes authentication: the pod's ServiceAccount token is exchanged
-    /// for a lease-bound Vault token that is renewed in the background.
-    Kubernetes {
-        /// Vault role bound to this ServiceAccount.
-        role: String,
-        /// Kubernetes auth engine mount path.
-        #[serde(default = "default_vault_kubernetes_mount")]
-        mount: String,
-        /// Projected ServiceAccount token to present. Re-read on every login so
-        /// a token the kubelet rotates is picked up without a restart.
-        #[serde(default = "default_vault_kubernetes_jwt_path")]
-        jwt_path: PathBuf,
-        /// Fail-closed margin in seconds, as on `AppRole`. Defaults to the
-        /// per-attempt timeout.
-        #[serde(default)]
-        refresh_safety_window_secs: Option<u64>,
-    },
     /// Agent-managed token file (for example a Vault Agent auto-auth sink):
     /// the token is read from `path` and re-read periodically so a token
     /// rotated by the agent is picked up without a restart.
@@ -555,16 +520,6 @@ impl VaultAuthMethod {
         }
     }
 
-    /// Kubernetes authentication with the default mount and projected token path.
-    pub fn kubernetes(role: String) -> Self {
-        Self::Kubernetes {
-            role,
-            mount: default_vault_kubernetes_mount(),
-            jwt_path: default_vault_kubernetes_jwt_path(),
-            refresh_safety_window_secs: None,
-        }
-    }
-
     /// Agent-managed token file with the default poll interval.
     pub fn token_file(path: PathBuf) -> Self {
         Self::TokenFile {
@@ -593,20 +548,6 @@ impl fmt::Debug for VaultAuthMethod {
                 .field("mount", mount)
                 .field("refresh_safety_window_secs", refresh_safety_window_secs)
                 .finish(),
-            // No redaction: the role and mount name a Vault binding, and the
-            // ServiceAccount token itself is never held on this type.
-            Self::Kubernetes {
-                role,
-                mount,
-                jwt_path,
-                refresh_safety_window_secs,
-            } => f
-                .debug_struct("Kubernetes")
-                .field("role", role)
-                .field("mount", mount)
-                .field("jwt_path", jwt_path)
-                .field("refresh_safety_window_secs", refresh_safety_window_secs)
-                .finish(),
             Self::TokenFile {
                 path,
                 poll_interval_secs,
@@ -1087,12 +1028,50 @@ impl KmsConfig {
                 });
             }
             KmsBackend::VaultKv2 => {
-                config.backend_config =
+                let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200");
-                    BackendConfig::VaultKv2(Box::new(vault_kv2_config_from_env(VaultCliOverrides::default())?));
+                let auth_method = vault_auth_method_from_env()?;
+                let skip_tls_verify = get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false);
+
+                let mount_path = match get_env_opt_str("RUSTFS_KMS_VAULT_MOUNT_PATH") {
+                    Some(path) => {
+                        tracing::warn!(
+                            "RUSTFS_KMS_VAULT_MOUNT_PATH is deprecated for the Vault KV2 backend: it never calls the Transit engine and the value is stored but unused"
+                        );
+                        path
+                    }
+                    None => default_vault_kv2_mount_path(),
+                };
+
+                config.backend_config = BackendConfig::VaultKv2(Box::new(VaultConfig {
+                    address,
+                    auth_method,
+                    namespace: get_env_opt_str("RUSTFS_KMS_VAULT_NAMESPACE"),
+                    mount_path,
+                    kv_mount: get_env_str("RUSTFS_KMS_VAULT_KV_MOUNT", "secret"),
+                    key_path_prefix: get_env_str("RUSTFS_KMS_VAULT_KEY_PREFIX", "rustfs/kms/keys"),
+                    tls: vault_tls_config(skip_tls_verify),
+                }));
             }
             KmsBackend::VaultTransit => {
-                config.backend_config =
+                let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200");
-                    BackendConfig::VaultTransit(Box::new(vault_transit_config_from_env(VaultCliOverrides::default())?));
+                let auth_method = vault_auth_method_from_env()?;
+                let skip_tls_verify = get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false);
+
+                config.backend_config = BackendConfig::VaultTransit(Box::new(VaultTransitConfig {
+                    address,
+                    auth_method,
+                    namespace: get_env_opt_str("RUSTFS_KMS_VAULT_NAMESPACE"),
+                    mount_path: get_env_str("RUSTFS_KMS_VAULT_MOUNT_PATH", "transit"),
+                    metadata_kv_mount: get_env_str(
+                        ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT,
+                        DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT,
+                    ),
+                    metadata_key_prefix: get_env_str(
+                        ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX,
+                        DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX,
+                    ),
+                    tls: vault_tls_config(skip_tls_verify),
+                }));
             }
             KmsBackend::Static => {
                 // Read from file first, then fall back to direct env var
@@ -1223,78 +1202,6 @@ fn is_under_temp_dir(path: &Path) -> bool {
     path.starts_with(std::env::temp_dir())
 }
 
-/// Command-line values that take precedence over the matching environment
-/// variables when assembling a Vault backend configuration.
-///
-/// Every field has a `RUSTFS_KMS_VAULT_*` equivalent that the CLI layer already
-/// reads, so these are only set when the operator passed an explicit flag.
-///
-/// Deliberately not `Debug`: `token` holds the raw Vault token, and the
-/// redacting `Debug` impls elsewhere in this module exist because a derived one
-/// would print it. Denying the derive makes a future `{overrides:?}` a compile
-/// error instead of a leak.
-#[derive(Default, Clone, Copy)]
-pub struct VaultCliOverrides<'a> {
-    pub address: Option<&'a str>,
-    pub token: Option<&'a str>,
-    pub mount_path: Option<&'a str>,
-}
-
-/// Assemble the Vault KV2 backend configuration from the environment.
-///
-/// Shared by [`KmsConfig::from_env`] and the server's command-line startup path
-/// so both resolve the same auth method, namespace, TLS and mount settings.
-pub fn vault_kv2_config_from_env(overrides: VaultCliOverrides<'_>) -> Result<VaultConfig> {
-    let mount_path = match overrides
-        .mount_path
-        .map(str::to_string)
-        .or_else(|| get_env_opt_str(ENV_KMS_VAULT_MOUNT_PATH))
-    {
-        Some(path) => {
-            tracing::warn!(
-                "RUSTFS_KMS_VAULT_MOUNT_PATH is deprecated for the Vault KV2 backend: it never calls the Transit engine and the value is stored but unused"
-            );
-            path
-        }
-        None => default_vault_kv2_mount_path(),
-    };
-
-    Ok(VaultConfig {
-        address: vault_address_from_env(overrides.address),
-        auth_method: vault_auth_method_from_env(overrides.token)?,
-        namespace: get_env_opt_str(ENV_KMS_VAULT_NAMESPACE),
-        mount_path,
-        kv_mount: get_env_str("RUSTFS_KMS_VAULT_KV_MOUNT", "secret"),
-        key_path_prefix: get_env_str("RUSTFS_KMS_VAULT_KEY_PREFIX", "rustfs/kms/keys"),
-        tls: vault_tls_config(get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false)),
-    })
-}
-
-/// Assemble the Vault Transit backend configuration from the environment.
-///
-/// Companion to [`vault_kv2_config_from_env`]; see there for why both entry
-/// points share it.
-pub fn vault_transit_config_from_env(overrides: VaultCliOverrides<'_>) -> Result<VaultTransitConfig> {
-    Ok(VaultTransitConfig {
-        address: vault_address_from_env(overrides.address),
-        auth_method: vault_auth_method_from_env(overrides.token)?,
-        namespace: get_env_opt_str(ENV_KMS_VAULT_NAMESPACE),
-        mount_path: overrides
-            .mount_path
-            .map(str::to_string)
-            .unwrap_or_else(|| get_env_str(ENV_KMS_VAULT_MOUNT_PATH, "transit")),
-        metadata_kv_mount: get_env_str(ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT, DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT),
-        metadata_key_prefix: get_env_str(ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX, DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX),
-        tls: vault_tls_config(get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false)),
-    })
-}
-
-fn vault_address_from_env(override_value: Option<&str>) -> String {
-    override_value
-        .map(str::to_string)
-        .unwrap_or_else(|| get_env_str(ENV_KMS_VAULT_ADDRESS, "http://localhost:8200"))
-}
-
 /// Resolve the Vault auth method from environment variables.
 ///
 /// Setting `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID` selects AppRole authentication;
@@ -1302,59 +1209,27 @@ fn vault_address_from_env(override_value: Option<&str>) -> String {
 /// (re-read on every login, mirroring the `RUSTFS_KMS_STATIC_SECRET_KEY_FILE`
 /// precedent) or inline from `RUSTFS_KMS_VAULT_APPROLE_SECRET_ID`, with the
 /// file taking precedence. Without a role id the legacy token flow applies.
-///
+fn vault_auth_method_from_env() -> Result<VaultAuthMethod> {
-/// `RUSTFS_KMS_VAULT_KUBERNETES_ROLE` selects Kubernetes authentication, which
-/// presents the pod's projected ServiceAccount token.
-///
-/// `token_override` carries a token supplied on the command line; it stands in
-/// for `RUSTFS_KMS_VAULT_TOKEN` everywhere below, including the conflict checks,
-/// so a flag and the variable it mirrors select the same method.
-fn vault_auth_method_from_env(token_override: Option<&str>) -> Result<VaultAuthMethod> {
-    let token = token_override
-        .map(str::to_string)
-        .or_else(|| get_env_opt_str(ENV_KMS_VAULT_TOKEN));
-    let role_id = get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID);
-    let kubernetes_role = get_env_opt_str(ENV_KMS_VAULT_KUBERNETES_ROLE);
-
     if let Some(token_file) = get_env_opt_str(ENV_KMS_VAULT_TOKEN_FILE) {
         // A token file names one authoritative credential source; combining it
         // with another one would leave the effective identity ambiguous, so
         // that is a configuration error rather than a precedence rule.
-        for (name, configured) in [
+        if get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID).is_some() {
-            (ENV_KMS_VAULT_APPROLE_ROLE_ID, role_id.is_some()),
+            return Err(KmsError::configuration_error(format!(
-            (ENV_KMS_VAULT_KUBERNETES_ROLE, kubernetes_role.is_some()),
+                "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with {ENV_KMS_VAULT_APPROLE_ROLE_ID}; configure exactly one Vault auth method"
-            (ENV_KMS_VAULT_TOKEN, token.is_some()),
+            )));
-        ] {
+        }
-            if configured {
+        if get_env_opt_str("RUSTFS_KMS_VAULT_TOKEN").is_some() {
-                return Err(KmsError::configuration_error(format!(
+            return Err(KmsError::configuration_error(format!(
-                    "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with {name}; configure exactly one Vault auth method"
+                "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with RUSTFS_KMS_VAULT_TOKEN; configure exactly one Vault auth method"
-                )));
+            )));
-            }
         }
         return Ok(VaultAuthMethod::token_file(PathBuf::from(token_file)));
     }
 
-    if let Some(role) = kubernetes_role {
+    let Some(role_id) = get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID) else {
-        // Unlike a leftover static token, a second login method is never a
-        // stale remnant: both were configured deliberately and neither can be
-        // ranked over the other.
-        if role_id.is_some() {
-            return Err(KmsError::configuration_error(format!(
-                "{ENV_KMS_VAULT_KUBERNETES_ROLE} cannot be combined with {ENV_KMS_VAULT_APPROLE_ROLE_ID}; configure exactly one Vault auth method"
-            )));
-        }
-        return Ok(VaultAuthMethod::Kubernetes {
-            role,
-            mount: get_env_str(ENV_KMS_VAULT_KUBERNETES_MOUNT, DEFAULT_VAULT_KUBERNETES_MOUNT),
-            jwt_path: get_env_opt_str(ENV_KMS_VAULT_KUBERNETES_JWT_PATH)
-                .map_or_else(default_vault_kubernetes_jwt_path, PathBuf::from),
-            refresh_safety_window_secs: None,
-        });
-    }
-
-    let Some(role_id) = role_id else {
         return Ok(VaultAuthMethod::Token {
-            token: token.unwrap_or_else(|| "dev-token".to_string()),
+            token: get_env_str("RUSTFS_KMS_VAULT_TOKEN", "dev-token"),
         });
     };
 
@@ -1398,22 +1273,6 @@ fn validate_vault_auth_method(backend_name: &str, auth_method: &VaultAuthMethod)
             }
             Ok(())
         }
-        VaultAuthMethod::Kubernetes {
-            role, mount, jwt_path, ..
-        } => {
-            if role.is_empty() {
-                return Err(KmsError::configuration_error(format!("{backend_name} Kubernetes role cannot be empty")));
-            }
-            if mount.is_empty() {
-                return Err(KmsError::configuration_error(format!("{backend_name} Kubernetes mount cannot be empty")));
-            }
-            if jwt_path.as_os_str().is_empty() {
-                return Err(KmsError::configuration_error(format!(
-                    "{backend_name} Kubernetes ServiceAccount token path cannot be empty"
-                )));
-            }
-            Ok(())
-        }
         VaultAuthMethod::TokenFile {
             path,
             poll_interval_secs,
@@ -2117,106 +1976,6 @@ mod tests {
             .expect("well-formed token file auth must validate");
     }
 
-    /// A Kubernetes role alone configures the method: the credential is the
-    /// pod's projected ServiceAccount token, so nothing secret is in the
-    /// environment and the mount and token path fall back to the cluster
-    /// defaults.
-    #[test]
-    fn test_from_env_selects_kubernetes() {
-        with_vars(
-            vec![
-                ("RUSTFS_KMS_BACKEND", Some("vault-transit")),
-                (ENV_KMS_VAULT_ADDRESS, Some("https://vault.example.com")),
-                (ENV_KMS_VAULT_KUBERNETES_ROLE, Some("rustfs")),
-                (ENV_KMS_VAULT_KUBERNETES_MOUNT, None),
-                (ENV_KMS_VAULT_KUBERNETES_JWT_PATH, None),
-                (ENV_KMS_VAULT_TOKEN, None),
-                (ENV_KMS_VAULT_TOKEN_FILE, None),
-                (ENV_KMS_VAULT_APPROLE_ROLE_ID, None),
-            ],
-            || {
-                let config = KmsConfig::from_env().expect("kms config should load from env");
-                let vault = config.vault_transit_config().expect("vault transit backend config");
-                let VaultAuthMethod::Kubernetes {
-                    role,
-                    mount,
-                    jwt_path,
-                    refresh_safety_window_secs,
-                } = &vault.auth_method
-                else {
-                    panic!(
-                        "a kubernetes role in the environment must select Kubernetes auth, got {:?}",
-                        vault.auth_method
-                    );
-                };
-                assert_eq!(role, "rustfs");
-                assert_eq!(mount, DEFAULT_VAULT_KUBERNETES_MOUNT);
-                assert_eq!(jwt_path, Path::new(DEFAULT_VAULT_KUBERNETES_JWT_PATH));
-                assert_eq!(refresh_safety_window_secs, &None);
-            },
-        );
-    }
-
-    #[test]
-    fn test_from_env_kubernetes_is_mutually_exclusive_with_other_auth() {
-        with_vars(
-            vec![
-                ("RUSTFS_KMS_BACKEND", Some("vault-transit")),
-                (ENV_KMS_VAULT_KUBERNETES_ROLE, Some("rustfs")),
-                (ENV_KMS_VAULT_APPROLE_ROLE_ID, Some("env-role-id")),
-                (ENV_KMS_VAULT_TOKEN, None),
-                (ENV_KMS_VAULT_TOKEN_FILE, None),
-            ],
-            || {
-                let error = KmsConfig::from_env().expect_err("kubernetes combined with approle must be rejected");
-                assert!(error.to_string().contains(ENV_KMS_VAULT_KUBERNETES_ROLE));
-                assert!(error.to_string().contains(ENV_KMS_VAULT_APPROLE_ROLE_ID));
-            },
-        );
-    }
-
-    #[test]
-    fn test_validate_rejects_bad_kubernetes_settings() {
-        let vault_config = |auth_method: VaultAuthMethod| KmsConfig {
-            backend: KmsBackend::VaultTransit,
-            backend_config: BackendConfig::VaultTransit(Box::new(VaultTransitConfig {
-                address: "https://vault.example.com:8200".to_string(),
-                auth_method,
-                ..Default::default()
-            })),
-            ..Default::default()
-        };
-
-        let error = vault_config(VaultAuthMethod::kubernetes(String::new()))
-            .validate()
-            .expect_err("an empty kubernetes role must be rejected");
-        assert!(error.to_string().contains("role"), "got {error}");
-
-        let error = vault_config(VaultAuthMethod::Kubernetes {
-            role: "rustfs".to_string(),
-            mount: String::new(),
-            jwt_path: PathBuf::from(DEFAULT_VAULT_KUBERNETES_JWT_PATH),
-            refresh_safety_window_secs: None,
-        })
-        .validate()
-        .expect_err("an empty kubernetes mount must be rejected");
-        assert!(error.to_string().contains("mount"), "got {error}");
-
-        let error = vault_config(VaultAuthMethod::Kubernetes {
-            role: "rustfs".to_string(),
-            mount: DEFAULT_VAULT_KUBERNETES_MOUNT.to_string(),
-            jwt_path: PathBuf::new(),
-            refresh_safety_window_secs: None,
-        })
-        .validate()
-        .expect_err("an empty ServiceAccount token path must be rejected");
-        assert!(error.to_string().contains("token path"), "got {error}");
-
-        vault_config(VaultAuthMethod::kubernetes("rustfs".to_string()))
-            .validate()
-            .expect("well-formed kubernetes auth must validate");
-    }
-
     /// Every KV2 read, write and listing is routed through `kv_mount`, so an
     /// empty one names a path no Vault engine answers. The Transit backend
     /// already rejects its own empty mounts; this closes the same gap on the

crates/madmin/src/site_replication.rs

@@ -258,7 +258,7 @@ pub struct SRLDAPUser {
     pub api_version: Option<String>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+#[derive(Debug, Serialize, Deserialize, Default)]
 pub struct SRIAMUser {
     #[serde(rename = "accessKey", default)]
     pub access_key: String,
@@ -270,7 +270,7 @@ pub struct SRIAMUser {
     pub api_version: Option<String>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+#[derive(Debug, Serialize, Deserialize, Default)]
 pub struct SRGroupInfo {
     #[serde(rename = "updateReq", default)]
     pub update_req: GroupAddRemove,
@@ -346,7 +346,7 @@ pub struct SRCredInfo {
     pub api_version: Option<String>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+#[derive(Debug, Serialize, Deserialize, Default)]
 pub struct SRIAMItem {
     #[serde(default)]
     pub r#type: String,

docs/operations/kms-backend-security.md

@@ -2,7 +2,7 @@
 
 RustFS ships several KMS backends. They differ not only in deployment effort but in **where master key material lives and who can read it**. Pick a backend based on the confidentiality boundary you need, not on the name alone.
 
-For how the Vault backends authenticate (static token, AppRole, Kubernetes, Vault Agent token file) and how credential refresh and the fail-closed window behave, see the [Vault KMS authentication runbook](vault-kms-authentication.md). For what may be claimed about the cryptographic implementations themselves, see [Cryptographic compliance positioning](kms-cryptographic-compliance.md). For which RustFS identities may manage or use a given key, see [Per-key KMS authorization](kms-per-key-authorization.md). If you are migrating from MinIO, read [Migrating from MinIO: encrypted objects do not carry over](#migrating-from-minio-encrypted-objects-do-not-carry-over) first.
+For how the Vault backends authenticate (static token, AppRole, Vault Agent token file) and how credential refresh and the fail-closed window behave, see the [Vault KMS authentication runbook](vault-kms-authentication.md). For what may be claimed about the cryptographic implementations themselves, see [Cryptographic compliance positioning](kms-cryptographic-compliance.md). For which RustFS identities may manage or use a given key, see [Per-key KMS authorization](kms-per-key-authorization.md). If you are migrating from MinIO, read [Migrating from MinIO: encrypted objects do not carry over](#migrating-from-minio-encrypted-objects-do-not-carry-over) first.
 
 ## Backend comparison
 

docs/operations/vault-kms-authentication.md

@@ -8,12 +8,9 @@ This runbook covers how the RustFS Vault KMS backends (KV2 and Transit) authenti
 | --- | --- | --- | --- | --- |
 | Static token | `Token` | Whatever the operator provisioned; RustFS never renews it | None | Development; short-lived experiments |
 | AppRole | `AppRole` | Lease-bound token obtained by login; renewed by RustFS | Renew at half TTL, re-login on failure | Production without a Vault Agent sidecar |
-| Kubernetes | `Kubernetes` | Lease-bound token obtained by login; renewed by RustFS | Renew at half TTL, re-login on failure | Production on Kubernetes, with no credential to distribute |
 | Agent token file | `TokenFile` | Owned by Vault Agent; RustFS only re-reads the sink file | File re-read once per poll interval | Production with a Vault Agent (or equivalent) managing auth |
 
-Exactly one method must be configured. Setting `RUSTFS_KMS_VAULT_TOKEN_FILE` together with any other method, or `RUSTFS_KMS_VAULT_KUBERNETES_ROLE` together with `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID`, is rejected at startup with a configuration error, because the effective identity would be ambiguous. A leftover `RUSTFS_KMS_VAULT_TOKEN` alongside a configured login method is tolerated and ignored, so a stale variable cannot silently downgrade the identity.
+Exactly one method must be configured. Setting `RUSTFS_KMS_VAULT_TOKEN_FILE` together with `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID` or an explicit `RUSTFS_KMS_VAULT_TOKEN` is rejected at startup with a configuration error, because the effective identity would be ambiguous.
-
-All of these are read the same way whether the service is started with `RUSTFS_KMS_ENABLE=true` or configured later through `POST /rustfs/admin/v3/kms/configure`.
 
 The default `dev-token` fallback for `RUSTFS_KMS_VAULT_TOKEN` is rejected outside explicit development mode (`RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS=true`), as are plain-HTTP Vault addresses and disabled TLS verification.
 
@@ -59,44 +56,7 @@ Deliver the SecretID out of band — a secrets-manager-mounted file, an init-con
 
 The secret_id file is re-read on every login attempt, so rotating the SecretID is a two-step operation with no restart: generate a new SecretID (`vault write -f auth/approle/role/rustfs-kms/secret-id`), atomically replace the file, then revoke the old SecretID accessor. The already-issued token keeps renewing; the new SecretID is only needed at the next full re-login.
 
-An empty or missing secret_id file fails the login attempt immediately (no Vault round trip). At startup the error is fatal — provider construction fails and the process exits — so a file missing at boot is recovered by restarting the process, not by an in-process retry. Once RustFS is running, the same failure is retried on the normal refresh cadence, so repairing the file mid-run heals the backend without a restart.
+An empty or missing secret_id file fails the login attempt immediately (no Vault round trip) and is retried on the normal refresh cadence, so repairing the file heals the backend without a restart.
-
-## Kubernetes authentication
-
-On Kubernetes this is the method to prefer: the pod's own ServiceAccount is the identity, so there is no credential to distribute, rotate, or leak into a Secret.
-
-### Vault-side setup
-
-```shell
-vault auth enable kubernetes
-
-vault write auth/kubernetes/config \
-    kubernetes_host="https://$KUBERNETES_SERVICE_HOST:$KUBERNETES_SERVICE_PORT"
-
-vault write auth/kubernetes/role/rustfs \
-    bound_service_account_names=rustfs \
-    bound_service_account_namespaces=rustfs \
-    token_policies=rustfs-kms \
-    token_ttl=1h
-```
-
-As with AppRole, keep `token_ttl` comfortably above the RustFS per-attempt timeout (default 30s).
-
-### RustFS configuration
-
-```shell
-RUSTFS_KMS_BACKEND=vault-transit            # or "vault" for the KV2 backend
-RUSTFS_KMS_VAULT_ADDRESS=https://vault.vault.svc.cluster.local:8200
-RUSTFS_KMS_VAULT_KUBERNETES_ROLE=rustfs
-# Optional, defaults to "kubernetes":
-# RUSTFS_KMS_VAULT_KUBERNETES_MOUNT=kubernetes
-# Optional, defaults to the kubelet's projected token path:
-# RUSTFS_KMS_VAULT_KUBERNETES_JWT_PATH=/var/run/secrets/kubernetes.io/serviceaccount/token
-```
-
-RustFS logs in at startup and renews the token at half its TTL, falling back to a fresh login exactly as AppRole does. The ServiceAccount token is re-read from disk on every login rather than cached, so a projected token the kubelet rotates is picked up without a restart.
-
-A missing or empty token file fails the login attempt immediately (no Vault round trip). At startup the error is fatal — provider construction fails and the process exits — so a token projected late during a slow pod start is recovered by the pod restart loop, not by an in-process retry. Once RustFS is running, a token file that goes missing or turns empty is retried on the normal refresh cadence and heals the backend on its own.
 
 ## Vault Agent token file
 
@@ -141,13 +101,13 @@ If the agent stops refreshing the file that is fine — RustFS re-reads the same
 
 ## Fail-closed window
 
-For lease-bound credentials (AppRole and Kubernetes tokens, token files), `current()` refuses to hand out a token that is within the safety window of its expiry and has not been refreshed. Requests then fail with `KMS credentials unavailable: ...` instead of being sent with a token that could lapse mid-flight and fail unpredictably on the Vault side.
+For lease-bound credentials (AppRole tokens, token files), `current()` refuses to hand out a token that is within the safety window of its expiry and has not been refreshed. Requests then fail with `KMS credentials unavailable: ...` instead of being sent with a token that could lapse mid-flight and fail unpredictably on the Vault side.
 
 - Default window: one per-attempt timeout (`RUSTFS_KMS_TIMEOUT_SECS`, default 30s) — a request issued now can legitimately stay in flight that long, so the token must outlive it.
-- Override: `refresh_safety_window_secs` on the `AppRole`, `Kubernetes` or `TokenFile` auth configuration.
+- Override: `refresh_safety_window_secs` on the `AppRole` or `TokenFile` auth configuration.
 - Static tokens never trip the window: they carry no lease and are assumed valid until Vault says otherwise.
 
-The window is a symptom threshold, not the fault itself: by the time it trips, refresh has been failing for roughly half the token TTL (AppRole, Kubernetes) or two poll intervals (token file).
+The window is a symptom threshold, not the fault itself: by the time it trips, refresh has been failing for roughly half the token TTL (AppRole) or two poll intervals (token file).
 
 ### Troubleshooting
 
@@ -157,8 +117,6 @@ The window is a symptom threshold, not the fault itself: by the time it trips, r
 | Renewal succeeded but re-login later fails | `Vault token renewal failed; falling back to a fresh login` followed by login errors | SecretID expired/revoked or AppRole role changed; rotate the secret_id file |
 | Token file mode error at startup or during polls | `has insecure permissions` in the error | Fix the sink `mode` (0600) and the file owner; the next poll heals the provider |
 | Token file missing/empty errors | `Failed to read Vault token file` / `token file ... is empty` | Vault Agent down or sink misconfigured; restart the agent, the next poll heals the provider |
-| Kubernetes login fails with a permission error | `Vault Kubernetes login failed` | The pod's ServiceAccount is not in the role's `bound_service_account_names`/`_namespaces`, or `auth/kubernetes/config` names the wrong API server |
+| Startup fails immediately with a configuration error naming two env vars | — | Two auth methods configured at once; keep exactly one of token, AppRole, token file |
-| Kubernetes ServiceAccount token errors | `Failed to read Kubernetes ServiceAccount token` / `ServiceAccount token ... is empty` | The token is not projected into the pod (check `automountServiceAccountToken` and the volume mount); the next refresh cycle heals the provider |
-| Startup fails immediately with a configuration error naming two env vars | — | Two auth methods configured at once; keep exactly one of token, AppRole, Kubernetes, token file |
 
 When diagnosing, confirm three clocks/lifetimes in order: the Vault token TTL (`vault token lookup` with the token's accessor), the RustFS refresh cadence (half TTL or the poll interval), and the fail-closed window. The renewal task logs every failed cycle, so a silent gap in warnings combined with `CredentialsUnavailable` errors points at the process clock or a paused runtime rather than Vault.

rustfs/src/admin/handlers/kms_backup.rs

@@ -286,7 +286,6 @@ fn auth_method_kind(auth: &VaultAuthMethod) -> String {
     match auth {
         VaultAuthMethod::Token { .. } => "token",
         VaultAuthMethod::AppRole { .. } => "approle",
-        VaultAuthMethod::Kubernetes { .. } => "kubernetes",
         VaultAuthMethod::TokenFile { .. } => "token-file",
     }
     .to_string()
@@ -485,10 +484,7 @@ fn business_trust_root_secrets(config: &KmsConfig) -> Vec<Zeroizing<String>> {
             secrets.push(Zeroizing::new(role_id.clone()));
             secrets.push(Zeroizing::new(secret_id.clone()));
         }
-        // Kubernetes and TokenFile hold no inline plaintext credential: the
+        VaultAuthMethod::TokenFile { .. } => {}
-        // ServiceAccount token and the agent-managed token live in files, and
-        // the role names a Vault binding rather than half a credential pair.
-        VaultAuthMethod::Kubernetes { .. } | VaultAuthMethod::TokenFile { .. } => {}
     };
 
     match &config.backend_config {

rustfs/src/app/object_usecase.rs

@@ -46,10 +46,9 @@ use super::storage_api::object_usecase::bucket::{
     replication::{
         DeleteReplicationConfigSnapshot, REPLICATE_INCOMING_DELETE, ReplicationStatusType, commit_force_delete_intent,
         delete_replication_state_from_config, delete_replication_version_id, deleted_object_has_pending_replication_delete,
-        force_delete_target_set, get_read_proxy_targets, has_active_delete_rule, load_delete_config_snapshot,
+        force_delete_target_set, has_active_delete_rule, load_delete_config_snapshot, must_replicate_object,
-        must_replicate_object, persist_force_delete_intent, record_replication_proxy, schedule_object_replication,
+        persist_force_delete_intent, schedule_object_replication, schedule_replication_delete, schedule_replication_deletes,
-        schedule_replication_delete, schedule_replication_deletes, set_deleted_object_replication_state,
+        set_deleted_object_replication_state, should_schedule_delete_replication, should_use_existing_delete_replication_info,
-        should_schedule_delete_replication, should_use_existing_delete_replication_info,
     },
     tagging::decode_tags,
     validate_restore_request,
@@ -6599,226 +6598,6 @@ impl DefaultObjectUsecase {
         })
     }
 
-    /// Headers a proxied read forwards verbatim to the replication target:
-    /// only the client's SSE-C key family, so the target performs the real
-    /// SSE-C decryption (never the replication-check exemption). HTTP
-    /// conditional headers (If-Match & co.) are deliberately NOT forwarded —
-    /// MinIO does not forward them either, and a remote 304/412 would leak a
-    /// conditional evaluation against a replica the local site never saw.
-    /// Range and part-number travel as typed SDK parameters instead.
-    fn proxy_read_passthrough_headers(headers: &HeaderMap) -> HeaderMap {
-        const FORWARDED: &[&str] = &[
-            "x-amz-server-side-encryption-customer-algorithm",
-            "x-amz-server-side-encryption-customer-key",
-            "x-amz-server-side-encryption-customer-key-md5",
-        ];
-        let mut forwarded = HeaderMap::new();
-        for name in FORWARDED {
-            if let Ok(header_name) = http::HeaderName::from_str(name)
-                && let Some(value) = headers.get(&header_name)
-            {
-                forwarded.insert(header_name, value.clone());
-            }
-        }
-        forwarded
-    }
-
-    /// True when a proxied SDK call failed because the target does not have
-    /// the object either (service-level not-found or a raw 404, which also
-    /// covers NoSuchVersion): the caller tries the next target silently.
-    fn proxy_sdk_error_is_not_found<E>(err: &aws_sdk_s3::error::SdkError<E>) -> bool {
-        err.raw_response().is_some_and(|resp| resp.status().as_u16() == 404)
-    }
-
-    /// Serve a GET whose local read failed with not-found by proxying to the
-    /// bucket's replication targets (MinIO `proxyGetToReplicationTarget`,
-    /// backlog#1675 P1-5). Returns None when no target can serve the object;
-    /// the caller then returns the original local error.
-    async fn proxy_get_object_to_replication_targets(
-        req: &S3Request<GetObjectInput>,
-        bucket: &str,
-        key: &str,
-        opts: &ObjectOptions,
-    ) -> Option<GetObjectOutput> {
-        let targets = get_read_proxy_targets(bucket, key, opts).await;
-        if targets.is_empty() {
-            return None;
-        }
-        let extra_headers = Self::proxy_read_passthrough_headers(&req.headers);
-        let range = req
-            .headers
-            .get(http::header::RANGE)
-            .and_then(|value| value.to_str().ok())
-            .map(str::to_owned);
-        let part_number = req.input.part_number;
-
-        for target in targets {
-            match target
-                .get_object(
-                    &target.bucket,
-                    key,
-                    opts.version_id.clone(),
-                    range.clone(),
-                    part_number,
-                    extra_headers.clone(),
-                )
-                .await
-            {
-                Ok(remote) => {
-                    // MinIO-aligned accounting: one total per proxy attempt
-                    // (targets were available), one failed when no target
-                    // served it — never per target.
-                    record_replication_proxy(bucket, "GetObject", false).await;
-                    return Some(Self::proxy_sdk_get_output_to_s3s(remote));
-                }
-                Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
-                    debug!(bucket, key, arn = %target.arn, "read proxy: target does not have the object");
-                }
-                Err(err) => {
-                    warn!(bucket, key, arn = %target.arn, error = %err, "read proxy: GET against replication target failed");
-                }
-            }
-        }
-        record_replication_proxy(bucket, "GetObject", true).await;
-        None
-    }
-
-    /// Serve a HEAD whose local lookup failed with not-found by proxying to
-    /// the bucket's replication targets (MinIO `proxyHeadToRepTarget`).
-    async fn proxy_head_object_to_replication_targets(
-        req: &S3Request<HeadObjectInput>,
-        bucket: &str,
-        key: &str,
-        opts: &ObjectOptions,
-    ) -> Option<HeadObjectOutput> {
-        let targets = get_read_proxy_targets(bucket, key, opts).await;
-        if targets.is_empty() {
-            return None;
-        }
-        let extra_headers = Self::proxy_read_passthrough_headers(&req.headers);
-        let range = req
-            .headers
-            .get(http::header::RANGE)
-            .and_then(|value| value.to_str().ok())
-            .map(str::to_owned);
-        let part_number = req.input.part_number;
-
-        for target in targets {
-            match target
-                .head_object_for_proxy(
-                    &target.bucket,
-                    key,
-                    opts.version_id.clone(),
-                    range.clone(),
-                    part_number,
-                    extra_headers.clone(),
-                )
-                .await
-            {
-                Ok(remote) => {
-                    // MinIO-aligned accounting: one total per proxy attempt,
-                    // one failed when no target served it.
-                    record_replication_proxy(bucket, "HeadObject", false).await;
-                    return Some(Self::proxy_sdk_head_output_to_s3s(remote));
-                }
-                Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
-                    debug!(bucket, key, arn = %target.arn, "read proxy: target does not have the object");
-                }
-                Err(err) => {
-                    warn!(bucket, key, arn = %target.arn, error = %err, "read proxy: HEAD against replication target failed");
-                }
-            }
-        }
-        record_replication_proxy(bucket, "HeadObject", true).await;
-        None
-    }
-
-    /// Translate a proxied SDK GET response into the s3s output, forwarding
-    /// the body as a stream (no buffering, no local persistence).
-    fn proxy_sdk_get_output_to_s3s(remote: aws_sdk_s3::operation::get_object::GetObjectOutput) -> GetObjectOutput {
-        let body = remote.body;
-        let body_stream = tokio_util::io::ReaderStream::with_capacity(body.into_async_read(), 64 * 1024);
-        GetObjectOutput {
-            body: Some(StreamingBlob::wrap(body_stream)),
-            content_length: remote.content_length,
-            content_range: remote.content_range,
-            content_type: remote.content_type.as_deref().and_then(|v| ContentType::from_str(v).ok()),
-            content_encoding: remote.content_encoding,
-            content_disposition: remote.content_disposition,
-            content_language: remote.content_language,
-            cache_control: remote.cache_control,
-            accept_ranges: Some(ACCEPT_RANGES_BYTES.to_string()),
-            e_tag: remote.e_tag.as_deref().and_then(|v| ETag::from_str(v).ok()),
-            last_modified: remote
-                .last_modified
-                .and_then(|dt| OffsetDateTime::from_unix_timestamp_nanos(dt.as_nanos()).ok())
-                .map(Timestamp::from),
-            metadata: remote.metadata,
-            version_id: remote.version_id,
-            server_side_encryption: remote
-                .server_side_encryption
-                .map(|sse| ServerSideEncryption::from(sse.as_str().to_string())),
-            sse_customer_algorithm: remote.sse_customer_algorithm,
-            sse_customer_key_md5: remote.sse_customer_key_md5,
-            ssekms_key_id: remote.ssekms_key_id,
-            parts_count: remote.parts_count,
-            tag_count: remote.tag_count,
-            storage_class: remote.storage_class.map(|sc| StorageClass::from(sc.as_str().to_string())),
-            expiration: remote.expiration,
-            restore: remote.restore,
-            checksum_crc32: remote.checksum_crc32,
-            checksum_crc32c: remote.checksum_crc32_c,
-            checksum_crc64nvme: remote.checksum_crc64_nvme,
-            checksum_sha1: remote.checksum_sha1,
-            checksum_sha256: remote.checksum_sha256,
-            checksum_type: remote.checksum_type.map(|ct| ChecksumType::from(ct.as_str().to_string())),
-            ..Default::default()
-        }
-    }
-
-    /// Translate a proxied SDK HEAD response into the s3s output.
-    ///
-    /// Known gaps: the SDK's HeadObjectOutput does not model 206/Content-Range
-    /// for a ranged HEAD (the SDK exposes no content_range member on HEAD),
-    /// and s3s' typed HeadObjectOutput has no tag_count field (the local path
-    /// injects x-amz-tagging-count as a raw header) — both are dropped for
-    /// proxied HEADs.
-    fn proxy_sdk_head_output_to_s3s(remote: aws_sdk_s3::operation::head_object::HeadObjectOutput) -> HeadObjectOutput {
-        HeadObjectOutput {
-            content_length: remote.content_length,
-            content_type: remote.content_type.as_deref().and_then(|v| ContentType::from_str(v).ok()),
-            content_encoding: remote.content_encoding,
-            content_disposition: remote.content_disposition,
-            content_language: remote.content_language,
-            cache_control: remote.cache_control,
-            accept_ranges: Some(ACCEPT_RANGES_BYTES.to_string()),
-            e_tag: remote.e_tag.as_deref().and_then(|v| ETag::from_str(v).ok()),
-            last_modified: remote
-                .last_modified
-                .and_then(|dt| OffsetDateTime::from_unix_timestamp_nanos(dt.as_nanos()).ok())
-                .map(Timestamp::from),
-            metadata: remote.metadata,
-            version_id: remote.version_id,
-            server_side_encryption: remote
-                .server_side_encryption
-                .map(|sse| ServerSideEncryption::from(sse.as_str().to_string())),
-            sse_customer_algorithm: remote.sse_customer_algorithm,
-            sse_customer_key_md5: remote.sse_customer_key_md5,
-            ssekms_key_id: remote.ssekms_key_id,
-            parts_count: remote.parts_count,
-            storage_class: remote.storage_class.map(|sc| StorageClass::from(sc.as_str().to_string())),
-            expiration: remote.expiration,
-            restore: remote.restore,
-            checksum_crc32: remote.checksum_crc32,
-            checksum_crc32c: remote.checksum_crc32_c,
-            checksum_crc64nvme: remote.checksum_crc64_nvme,
-            checksum_sha1: remote.checksum_sha1,
-            checksum_sha256: remote.checksum_sha256,
-            checksum_type: remote.checksum_type.map(|ct| ChecksumType::from(ct.as_str().to_string())),
-            ..Default::default()
-        }
-    }
-
     #[instrument(name = "execute_get_object", level = "trace", skip(self, req))]
     pub async fn execute_get_object(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
         self.execute_get_object_boxed(req).await
@@ -6944,19 +6723,6 @@ impl DefaultObjectUsecase {
         {
             Ok(prepared_read) => prepared_read,
             Err(err) => {
-                // Active-active replication lag window: an object missing
-                // locally (and only missing — other errors keep their
-                // semantics) may still be served by proxying the GET to a
-                // replication target (backlog#1675 P1-5).
-                if matches!(*err.code(), S3ErrorCode::NoSuchKey | S3ErrorCode::NoSuchVersion)
-                    && let Some(output) = Self::proxy_get_object_to_replication_targets(&req, &bucket, &key, &opts).await
-                {
-                    lifecycle.finish_ok();
-                    let response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await;
-                    let result = Ok(response);
-                    let _ = helper.version_id(version_id_for_event).complete(&result);
-                    return result;
-                }
                 lifecycle.finish_err();
                 return Err(err);
             }
@@ -8866,17 +8632,6 @@ impl DefaultObjectUsecase {
                         let msg = head_prefix_not_found_message(&bucket, &key, has_children);
                         return Err(S3Error::with_message(S3ErrorCode::NoSuchKey, msg));
                     }
-                    // Active-active replication lag window: an object missing
-                    // locally may still be served by proxying the HEAD to a
-                    // replication target (backlog#1675 P1-5).
-                    if let Some(output) = Self::proxy_head_object_to_replication_targets(&req, &bucket, &key, &opts).await {
-                        let response = wrap_response_with_cors(&bucket, &req.method, &req.headers, output).await;
-                        let result = Ok(response);
-                        let _ = helper
-                            .version_id(req.input.version_id.clone().unwrap_or_default())
-                            .complete(&result);
-                        return result;
-                    }
                     return Err(S3Error::new(S3ErrorCode::NoSuchKey));
                 }
                 // Other errors, such as insufficient permissions, still return the original error

rustfs/src/app/storage_api.rs

@@ -627,24 +627,6 @@ pub(crate) mod bucket {
         #[cfg(test)]
         pub(crate) use replication_contracts::replication_statuses_map;
 
-        /// Remote replication-target client used by the read-proxy path.
-        pub(crate) type ProxyTargetClient = crate::storage::storage_api::ecstore_bucket::bucket_target_sys::TargetClient;
-
-        /// Proxy-request metric recorder (get/head/tagging totals + failures).
-        pub(crate) use crate::storage::storage_api::record_replication_proxy;
-
-        /// Replication targets eligible to serve a proxied GET/HEAD/Tagging of
-        /// an object not present locally (MinIO `getProxyTargets`; empty when
-        /// the request was itself proxied, versioning is suspended, or no
-        /// replication rule matches). backlog#1675 P1-5.
-        pub(crate) async fn get_read_proxy_targets(
-            bucket: &str,
-            object: &str,
-            opts: &crate::storage::storage_api::StorageObjectOptions,
-        ) -> Vec<Arc<ProxyTargetClient>> {
-            replication_contracts::get_proxy_targets(bucket, object, opts).await
-        }
-
         pub(crate) async fn persist_force_delete_intent(
             store: Arc<crate::storage::storage_api::ECStore>,
             bucket: String,

rustfs/src/init.rs

@@ -304,37 +304,30 @@ fn build_local_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::c
     Ok(kms_config)
 }
 
-/// Collect the Vault settings the command line owns.
-///
-/// Everything else — auth method, namespace, TLS, KV mount and metadata paths —
-/// is resolved from the environment by the KMS crate, so this path and
-/// [`rustfs_kms::config::KmsConfig::from_env`] cannot drift apart. The address
-/// stays required here so a missing one is still named instead of silently
-/// falling back to the crate's localhost default.
-fn vault_cli_overrides<'a>(
-    cfg: &'a config::Config,
-    backend_name: &str,
-) -> std::io::Result<rustfs_kms::config::VaultCliOverrides<'a>> {
-    let address = cfg
-        .kms_vault_address
-        .as_deref()
-        .ok_or_else(|| Error::other(format!("Vault address is required for {backend_name} backend")))?;
-
-    Ok(rustfs_kms::config::VaultCliOverrides {
-        address: Some(address),
-        token: cfg.kms_vault_token.as_deref(),
-        mount_path: cfg.kms_vault_mount_path.as_deref(),
-    })
-}
-
 /// Build KMS configuration for Vault backend
 fn build_vault_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::config::KmsConfig> {
-    let backend_config = rustfs_kms::config::vault_kv2_config_from_env(vault_cli_overrides(cfg, "vault")?)
+    let vault_address = cfg
-        .map_err(|e| Error::other(format!("Vault KMS configuration failed: {e}")))?;
+        .kms_vault_address
+        .as_ref()
+        .ok_or_else(|| Error::other("Vault address is required for vault backend"))?;
+    let vault_token = cfg
+        .kms_vault_token
+        .as_ref()
+        .ok_or_else(|| Error::other("Vault token is required for vault backend"))?;
 
     let kms_config = rustfs_kms::config::KmsConfig {
         backend: rustfs_kms::config::KmsBackend::VaultKv2,
-        backend_config: rustfs_kms::config::BackendConfig::VaultKv2(Box::new(backend_config)),
+        backend_config: rustfs_kms::config::BackendConfig::VaultKv2(Box::new(rustfs_kms::config::VaultConfig {
+            address: vault_address.clone(),
+            auth_method: rustfs_kms::config::VaultAuthMethod::Token {
+                token: vault_token.clone(),
+            },
+            namespace: None,
+            mount_path: cfg.kms_vault_mount_path.clone().unwrap_or_else(|| "transit".to_string()),
+            kv_mount: "secret".to_string(),
+            key_path_prefix: "rustfs/kms/keys".to_string(),
+            tls: None,
+        })),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
         allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         default_key_id: cfg.kms_default_key_id.clone(),
@@ -351,12 +344,26 @@ fn build_vault_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::c
 
 /// Build KMS configuration for Vault Transit backend
 fn build_vault_transit_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::config::KmsConfig> {
-    let backend_config = rustfs_kms::config::vault_transit_config_from_env(vault_cli_overrides(cfg, "vault-transit")?)
+    let vault_address = cfg
-        .map_err(|e| Error::other(format!("Vault Transit KMS configuration failed: {e}")))?;
+        .kms_vault_address
+        .as_ref()
+        .ok_or_else(|| Error::other("Vault address is required for vault-transit backend"))?;
+    let vault_token = cfg
+        .kms_vault_token
+        .as_ref()
+        .ok_or_else(|| Error::other("Vault token is required for vault-transit backend"))?;
 
     let kms_config = rustfs_kms::config::KmsConfig {
         backend: rustfs_kms::config::KmsBackend::VaultTransit,
-        backend_config: rustfs_kms::config::BackendConfig::VaultTransit(Box::new(backend_config)),
+        backend_config: rustfs_kms::config::BackendConfig::VaultTransit(Box::new(rustfs_kms::config::VaultTransitConfig {
+            address: vault_address.clone(),
+            auth_method: rustfs_kms::config::VaultAuthMethod::Token {
+                token: vault_token.clone(),
+            },
+            namespace: None,
+            mount_path: cfg.kms_vault_mount_path.clone().unwrap_or_else(|| "transit".to_string()),
+            ..rustfs_kms::config::VaultTransitConfig::default()
+        })),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
         allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         default_key_id: cfg.kms_default_key_id.clone(),
@@ -1398,10 +1405,7 @@ pub async fn init_sftp_system() -> Result<Option<ShutdownHandle>, Box<dyn std::e
 
 #[cfg(test)]
 mod tests {
-    use super::{
+    use super::{build_aws_kms_config, notification_config_to_event_rules, resolve_buffer_profile_config};
-        build_aws_kms_config, build_vault_kms_config, build_vault_transit_kms_config, notification_config_to_event_rules,
-        resolve_buffer_profile_config,
-    };
     use crate::config::{BufferConfig, WorkloadProfile};
     use rustfs_config::KI_B;
     use rustfs_s3_types::EventName;
@@ -1495,151 +1499,6 @@ mod tests {
         assert!(err.to_string().contains("Invalid ARN"), "unexpected error: {err}");
     }
 
-    fn vault_kms_test_config(backend: &str) -> crate::config::Config {
-        let mut config = crate::config::Config::new("127.0.0.1:9000", vec!["/tmp/rustfs-vault-kms".to_string()]);
-        config.kms_enable = true;
-        config.kms_backend = backend.to_string();
-        config.kms_vault_address = Some("https://vault.example.com:8200".to_string());
-        config
-    }
-
-    /// The Vault auth method and the settings the CLI has no flag for come from
-    /// the environment, so startup and `KmsConfig::from_env` cannot disagree.
-    /// Regression: startup used to hardcode token auth and require a token,
-    /// which made every non-token method unreachable through `RUSTFS_KMS_ENABLE`.
-    #[test]
-    fn build_vault_transit_kms_config_resolves_auth_and_mounts_from_env() {
-        let config = temp_env::with_vars(
-            [
-                ("RUSTFS_KMS_VAULT_TOKEN", None),
-                ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", None),
-                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", Some("env-role-id")),
-                ("RUSTFS_KMS_VAULT_APPROLE_SECRET_ID", Some("env-secret-id")),
-                ("RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE", None),
-                ("RUSTFS_KMS_VAULT_NAMESPACE", Some("team-a")),
-                ("RUSTFS_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT", Some("rustfs-kv")),
-            ],
-            || {
-                build_vault_transit_kms_config(&vault_kms_test_config("vault-transit"))
-                    .expect("vault transit KMS configuration should build")
-            },
-        );
-
-        let vault = config.vault_transit_config().expect("vault transit backend config");
-        let rustfs_kms::config::VaultAuthMethod::AppRole { role_id, secret_id, .. } = &vault.auth_method else {
-            panic!("approle in the environment must select AppRole auth, got {:?}", vault.auth_method);
-        };
-        assert_eq!(role_id, "env-role-id");
-        assert_eq!(secret_id, "env-secret-id");
-        assert_eq!(vault.namespace.as_deref(), Some("team-a"));
-        assert_eq!(vault.metadata_kv_mount, "rustfs-kv");
-    }
-
-    /// Kubernetes auth needs no credential in the environment at all: the role
-    /// selects it and the pod's projected ServiceAccount token supplies the rest.
-    #[test]
-    fn build_vault_transit_kms_config_selects_kubernetes_auth() {
-        let config = temp_env::with_vars(
-            [
-                ("RUSTFS_KMS_VAULT_TOKEN", None),
-                ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
-                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", Some("rustfs")),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_MOUNT", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_JWT_PATH", None),
-            ],
-            || {
-                build_vault_transit_kms_config(&vault_kms_test_config("vault-transit"))
-                    .expect("vault transit KMS configuration should build")
-            },
-        );
-
-        let vault = config.vault_transit_config().expect("vault transit backend config");
-        let rustfs_kms::config::VaultAuthMethod::Kubernetes {
-            role, mount, jwt_path, ..
-        } = &vault.auth_method
-        else {
-            panic!(
-                "a kubernetes role in the environment must select Kubernetes auth, got {:?}",
-                vault.auth_method
-            );
-        };
-        assert_eq!(role, "rustfs");
-        assert_eq!(mount, rustfs_kms::config::DEFAULT_VAULT_KUBERNETES_MOUNT);
-        assert_eq!(jwt_path, std::path::Path::new(rustfs_kms::config::DEFAULT_VAULT_KUBERNETES_JWT_PATH));
-    }
-
-    /// Two credential sources leave the effective identity ambiguous, so
-    /// startup refuses rather than picking one.
-    #[test]
-    fn build_vault_kms_config_refuses_two_auth_methods() {
-        temp_env::with_vars(
-            [
-                ("RUSTFS_KMS_VAULT_TOKEN", None),
-                ("RUSTFS_KMS_VAULT_TOKEN_FILE", Some("/run/vault-agent/token")),
-                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", Some("rustfs")),
-            ],
-            || {
-                let error = build_vault_kms_config(&vault_kms_test_config("vault"))
-                    .expect_err("two Vault auth methods must not start the server");
-                assert!(error.to_string().contains("exactly one"), "unexpected error: {error}");
-            },
-        );
-    }
-
-    /// The KV2 backend has its own builder, so the key-location settings have
-    /// to be proven separately from the Transit one: pointing at the wrong KV
-    /// mount or prefix makes existing keys look absent.
-    #[test]
-    fn build_vault_kms_config_resolves_kv_mount_and_prefix_from_env() {
-        let config = temp_env::with_vars(
-            [
-                ("RUSTFS_KMS_VAULT_TOKEN", Some("a-real-token")),
-                ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
-                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
-                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", None),
-                ("RUSTFS_KMS_VAULT_KV_MOUNT", Some("rustfs-kv")),
-                ("RUSTFS_KMS_VAULT_KEY_PREFIX", Some("tenant/keys")),
-            ],
-            || build_vault_kms_config(&vault_kms_test_config("vault")).expect("vault KV2 KMS configuration should build"),
-        );
-
-        let vault = config.vault_config().expect("vault kv2 backend config");
-        assert_eq!(vault.kv_mount, "rustfs-kv");
-        assert_eq!(vault.key_path_prefix, "tenant/keys");
-    }
-
-    /// Skipping TLS verification was silently dropped on this path before, so
-    /// an operator who asked for it still got a verified connection. Now that it
-    /// is honoured it must fail closed without the development opt-in, rather
-    /// than quietly downgrading the Vault connection.
-    #[test]
-    fn build_vault_transit_kms_config_refuses_skip_tls_verify_without_opt_in() {
-        let vars = [
-            ("RUSTFS_KMS_VAULT_TOKEN", Some("a-real-token")),
-            ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
-            ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
-            ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", None),
-            ("RUSTFS_KMS_VAULT_SKIP_TLS_VERIFY", Some("true")),
-        ];
-
-        temp_env::with_vars(vars, || {
-            let error = build_vault_transit_kms_config(&vault_kms_test_config("vault-transit"))
-                .expect_err("skipping TLS verification must not start the server");
-            assert!(error.to_string().contains("TLS"), "unexpected error: {error}");
-        });
-
-        temp_env::with_vars(vars, || {
-            let mut cfg = vault_kms_test_config("vault-transit");
-            cfg.kms_allow_insecure_dev_defaults = true;
-            let config = build_vault_transit_kms_config(&cfg).expect("the development opt-in should accept skip-verify");
-            let vault = config.vault_transit_config().expect("vault transit backend config");
-            assert!(vault.tls.as_ref().is_some_and(|tls| tls.skip_verify));
-        });
-    }
-
     fn aws_kms_test_config() -> crate::config::Config {
         let mut config = crate::config::Config::new("127.0.0.1:9000", vec!["/tmp/rustfs-aws-kms".to_string()]);
         config.kms_enable = true;

rustfs/src/storage/ecfs.rs

@@ -12,15 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::StorageVersioningConfigExt as _;
 use super::{
     BUCKET_ACCELERATE_CONFIG, BUCKET_LOGGING_CONFIG, BUCKET_REQUEST_PAYMENT_CONFIG, BUCKET_VERSIONING_CONFIG,
     BUCKET_WEBSITE_CONFIG, BucketVersioningSys, OBJECT_LOCK_CONFIG, StorageError, check_retention_for_modification, decode_tags,
     decode_tags_to_map, delete_bucket_metadata_config_if_incarnation, encode_tags, get_bucket_accelerate_config,
-    get_bucket_logging_config, get_bucket_object_lock_config, get_bucket_request_payment_config, get_bucket_website_config,
+    get_bucket_logging_config, get_bucket_object_lock_config, get_bucket_replication_config, get_bucket_request_payment_config,
-    is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, record_replication_proxy, serialize,
+    get_bucket_website_config, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
-    update_bucket_metadata_config_if_incarnation,
+    record_replication_proxy, serialize, update_bucket_metadata_config_if_incarnation,
 };
+use super::{StorageReplicationConfigExt as _, StorageVersioningConfigExt as _};
 use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook;
 use crate::error::ApiError;
 use crate::storage::access::{apply_bucket_generation_guard, bucket_config_mutation_incarnation, has_bypass_governance_header};
@@ -59,7 +59,7 @@ const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock";
 const LOG_SUBSYSTEM_TAGGING: &str = "tagging";
 
 use crate::app::storage_api::object_usecase::bucket::replication::{
-    ReplicateDecision, get_read_proxy_targets, must_replicate_metadata, schedule_metadata_replication,
+    ReplicateDecision, must_replicate_metadata, schedule_metadata_replication,
 };
 use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOptions;
 
@@ -105,152 +105,18 @@ impl FS {
         &self.server_ctx
     }
 
-    /// Not-found classifier for proxied SDK tagging calls: a raw 404 covers
+    async fn replication_tagging_enabled(bucket: &str, object: &str) -> bool {
-    /// NoSuchKey and NoSuchVersion alike; the caller silently tries the next
+        get_bucket_replication_config(bucket)
-    /// replication target.
+            .await
-    fn proxy_sdk_error_is_not_found<E>(err: &aws_sdk_s3::error::SdkError<E>) -> bool {
+            .map(|(cfg, _)| cfg.has_active_rules(object, true))
-        err.raw_response().is_some_and(|resp| resp.status().as_u16() == 404)
+            .unwrap_or(false)
     }
 
-    /// Selector options for a tagging proxy. Reuses `get_opts` so the
+    async fn record_replication_tagging_metric(bucket: &str, object: &str, api: &str, is_err: bool) {
-    /// anti-loop `source-proxy-request` header family and the bucket's
+        if !Self::replication_tagging_enabled(bucket, object).await {
-    /// version-suspension state gate proxying exactly like GET/HEAD.
+            return;
-    async fn tagging_proxy_opts(
-        bucket: &str,
-        object: &str,
-        version_id: Option<String>,
-        headers: &http::HeaderMap,
-    ) -> Option<ObjectOptions> {
-        get_opts(bucket, object, version_id, None, headers).await.ok()
-    }
-
-    /// Serve a GetObjectTagging for an object missing locally by proxying to
-    /// the bucket's replication targets (MinIO `proxyGetTaggingToRepTarget`,
-    /// backlog#1675 P1-5). None means no target had the object.
-    async fn proxy_get_object_tagging(
-        bucket: &str,
-        object: &str,
-        version_id: Option<String>,
-        headers: &http::HeaderMap,
-    ) -> Option<TagSet> {
-        let opts = Self::tagging_proxy_opts(bucket, object, version_id, headers).await?;
-        let targets = get_read_proxy_targets(bucket, object, &opts).await;
-        if targets.is_empty() {
-            return None;
         }
-        for target in targets {
+        record_replication_proxy(bucket, api, is_err).await;
-            match target
-                .get_object_tagging(&target.bucket, object, opts.version_id.clone())
-                .await
-            {
-                Ok(remote) => {
-                    // MinIO-aligned accounting: one total per proxy attempt,
-                    // one failed when no target served it.
-                    record_replication_proxy(bucket, "GetObjectTagging", false).await;
-                    return Some(
-                        remote
-                            .tag_set
-                            .into_iter()
-                            .map(|tag| Tag {
-                                key: Some(tag.key),
-                                value: Some(tag.value),
-                            })
-                            .collect(),
-                    );
-                }
-                Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
-                    debug!(bucket, object, arn = %target.arn, "tagging proxy: target does not have the object");
-                }
-                Err(err) => {
-                    warn!(bucket, object, arn = %target.arn, error = %err, "tagging proxy: GetObjectTagging against replication target failed");
-                }
-            }
-        }
-        record_replication_proxy(bucket, "GetObjectTagging", true).await;
-        None
-    }
-
-    /// Apply a PutObjectTagging for an object missing locally on a
-    /// replication target (MinIO `proxyTaggingToRepTarget`).
-    async fn proxy_put_object_tagging(
-        bucket: &str,
-        object: &str,
-        version_id: Option<String>,
-        headers: &http::HeaderMap,
-        tag_set: &TagSet,
-    ) -> Option<()> {
-        let opts = Self::tagging_proxy_opts(bucket, object, version_id, headers).await?;
-        let mut tagging = aws_sdk_s3::types::Tagging::builder();
-        for tag in tag_set {
-            let sdk_tag = aws_sdk_s3::types::Tag::builder()
-                .key(tag.key.clone().unwrap_or_default())
-                .value(tag.value.clone().unwrap_or_default())
-                .build()
-                .ok()?;
-            tagging = tagging.tag_set(sdk_tag);
-        }
-        let tagging = tagging.build().ok()?;
-        let targets = get_read_proxy_targets(bucket, object, &opts).await;
-        if targets.is_empty() {
-            return None;
-        }
-        for target in targets {
-            match target
-                .put_object_tagging(&target.bucket, object, opts.version_id.clone(), tagging.clone())
-                .await
-            {
-                Ok(_) => {
-                    // MinIO-aligned accounting: one total per proxy attempt,
-                    // one failed when no target served it.
-                    record_replication_proxy(bucket, "PutObjectTagging", false).await;
-                    return Some(());
-                }
-                Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
-                    debug!(bucket, object, arn = %target.arn, "tagging proxy: target does not have the object");
-                }
-                Err(err) => {
-                    warn!(bucket, object, arn = %target.arn, error = %err, "tagging proxy: PutObjectTagging against replication target failed");
-                }
-            }
-        }
-        record_replication_proxy(bucket, "PutObjectTagging", true).await;
-        None
-    }
-
-    /// Apply a DeleteObjectTagging for an object missing locally on a
-    /// replication target (MinIO `proxyTaggingToRepTarget`).
-    async fn proxy_delete_object_tagging(
-        bucket: &str,
-        object: &str,
-        version_id: Option<String>,
-        headers: &http::HeaderMap,
-    ) -> Option<()> {
-        let opts = Self::tagging_proxy_opts(bucket, object, version_id, headers).await?;
-        let targets = get_read_proxy_targets(bucket, object, &opts).await;
-        if targets.is_empty() {
-            return None;
-        }
-        for target in targets {
-            match target
-                .delete_object_tagging(&target.bucket, object, opts.version_id.clone())
-                .await
-            {
-                Ok(_) => {
-                    // MinIO-aligned accounting: one total per proxy attempt,
-                    // one failed when no target served it.
-                    record_replication_proxy(bucket, "DeleteObjectTagging", false).await;
-                    return Some(());
-                }
-                Err(err) if Self::proxy_sdk_error_is_not_found(&err) => {
-                    debug!(bucket, object, arn = %target.arn, "tagging proxy: target does not have the object");
-                }
-                Err(err) => {
-                    warn!(bucket, object, arn = %target.arn, error = %err, "tagging proxy: DeleteObjectTagging against replication target failed");
-                }
-            }
-        }
-        record_replication_proxy(bucket, "DeleteObjectTagging", true).await;
-        None
     }
 
     pub async fn get_object_tag_conditions_for_policy(
@@ -581,27 +447,7 @@ impl S3 for FS {
         let mut opts = get_opts(&bucket, &object, version_id.clone(), None, &req.headers)
             .await
             .map_err(ApiError::from)?;
-        let existing_object_info = match store.get_object_info(&bucket, &object, &opts).await {
+        let existing_object_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?;
-            Ok(info) => info,
-            Err(e) => {
-                // Replication lag window: apply the tagging delete on a
-                // replication target that already has the object
-                // (backlog#1675 P1-5). No local object exists, so no bucket
-                // notification event is emitted for the proxied write.
-                if (is_err_object_not_found(&e) || is_err_version_not_found(&e))
-                    && Self::proxy_delete_object_tagging(&bucket, &object, version_id.clone(), &req.headers)
-                        .await
-                        .is_some()
-                {
-                    counter!("rustfs_delete_object_tagging_success").increment(1);
-                    let duration = start_time.elapsed();
-                    histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "delete")
-                        .record(duration.as_secs_f64());
-                    return Ok(S3Response::new(DeleteObjectTaggingOutput { version_id }));
-                }
-                return Err(ApiError::from(e).into());
-            }
-        };
         let dsc = must_replicate_metadata(
             &bucket,
             &object,
@@ -624,6 +470,7 @@ impl S3 for FS {
         }
 
         let delete_tags_result = store.delete_object_tags(&bucket, &object, &opts).await;
+        Self::record_replication_tagging_metric(&bucket, &object, "DeleteObjectTagging", delete_tags_result.is_err()).await;
         let object_info = delete_tags_result.map_err(|e| {
             error!(
                 component = LOG_COMPONENT_STORAGE,
@@ -1081,49 +928,32 @@ impl S3 for FS {
             ..Default::default()
         };
 
-        let tags = match store.get_object_tags(bucket, object, &opts).await {
+        let tags_result = store.get_object_tags(bucket, object, &opts).await;
-            Ok(tags) => tags,
+        Self::record_replication_tagging_metric(bucket, object, "GetObjectTagging", tags_result.is_err()).await;
-            Err(e) => {
+        let tags = tags_result.map_err(|e| {
-                // Replication lag window: the object may exist on a
+            if is_err_object_not_found(&e) {
-                // replication target even though it is missing locally —
+                debug!(
-                // proxy the tagging read there (backlog#1675 P1-5).
-                if (is_err_object_not_found(&e) || is_err_version_not_found(&e))
-                    && let Some(tag_set) =
-                        Self::proxy_get_object_tagging(bucket, object, req.input.version_id.clone(), &req.headers).await
-                {
-                    counter!("rustfs_get_object_tagging_success").increment(1);
-                    let duration = start_time.elapsed();
-                    histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "get")
-                        .record(duration.as_secs_f64());
-                    return Ok(S3Response::new(GetObjectTaggingOutput {
-                        tag_set,
-                        version_id: req.input.version_id.clone(),
-                    }));
-                }
-                if is_err_object_not_found(&e) {
-                    debug!(
-                        component = LOG_COMPONENT_STORAGE,
-                        subsystem = LOG_SUBSYSTEM_TAGGING,
-                        event = "object_tagging_not_found",
-                        bucket = %bucket,
-                        object = %object,
-                        error = %e,
-                        "Object tags not found"
-                    );
-                    return Err(s3_error!(NoSuchKey));
-                }
-                error!(
                     component = LOG_COMPONENT_STORAGE,
                     subsystem = LOG_SUBSYSTEM_TAGGING,
-                    event = "object_tagging_get_failed",
+                    event = "object_tagging_not_found",
                     bucket = %bucket,
                     object = %object,
                     error = %e,
-                    "Failed to load object tags"
+                    "Object tags not found"
                 );
-                return Err(ApiError::from(e).into());
+                return s3_error!(NoSuchKey);
             }
-        };
+            error!(
+                component = LOG_COMPONENT_STORAGE,
+                subsystem = LOG_SUBSYSTEM_TAGGING,
+                event = "object_tagging_get_failed",
+                bucket = %bucket,
+                object = %object,
+                error = %e,
+                "Failed to load object tags"
+            );
+            ApiError::from(e).into()
+        })?;
 
         let tag_set = decode_tags(tags.as_str());
         debug!(
@@ -1799,36 +1629,14 @@ impl S3 for FS {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
 
-        let tags = encode_tags(tagging.tag_set.clone());
+        let tags = encode_tags(tagging.tag_set);
         debug!("Encoded tags: {}", tags);
 
         let version_id = req.input.version_id.clone();
         let mut opts = get_opts(&bucket, &object, version_id.clone(), None, &req.headers)
             .await
             .map_err(ApiError::from)?;
-        let existing_object_info = match store.get_object_info(&bucket, &object, &opts).await {
+        let existing_object_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?;
-            Ok(info) => info,
-            Err(e) => {
-                // Replication lag window: apply the tagging update on a
-                // replication target that already has the object
-                // (backlog#1675 P1-5). No local object exists, so no bucket
-                // notification event is emitted for the proxied write.
-                if (is_err_object_not_found(&e) || is_err_version_not_found(&e))
-                    && Self::proxy_put_object_tagging(&bucket, &object, version_id.clone(), &req.headers, &tagging.tag_set)
-                        .await
-                        .is_some()
-                {
-                    counter!("rustfs_put_object_tagging_success").increment(1);
-                    let duration = start_time.elapsed();
-                    histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "put")
-                        .record(duration.as_secs_f64());
-                    return Ok(S3Response::new(PutObjectTaggingOutput {
-                        version_id: req.input.version_id.clone(),
-                    }));
-                }
-                return Err(ApiError::from(e).into());
-            }
-        };
         let dsc = must_replicate_metadata(
             &bucket,
             &object,
@@ -1851,6 +1659,7 @@ impl S3 for FS {
         }
 
         let put_tags_result = store.put_object_tags(&bucket, &object, &tags, &opts).await;
+        Self::record_replication_tagging_metric(&bucket, &object, "PutObjectTagging", put_tags_result.is_err()).await;
         let object_info = put_tags_result.map_err(|e| {
             error!("Failed to put object tags: {}", e);
             counter!("rustfs_put_object_tagging_failure").increment(1);

rustfs/src/storage/mod.rs

@@ -57,24 +57,24 @@ pub(crate) use storage_api::{
     QuotaError, RUSTFS_META_BUCKET, RawFileInfo, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
     ReplicationStats, ReplicationStatusType, Result, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC,
     StorageDeletedObject, StorageDiskRpcExt, StorageError, StorageGetObjectReader, StorageObjectInfo, StorageObjectOptions,
-    StorageObjectToDelete, StoragePeerS3ClientExt, StoragePutObjReader, StorageVersioningConfigExt, TONIC_RPC_PREFIX,
+    StorageObjectToDelete, StoragePeerS3ClientExt, StoragePutObjReader, StorageReplicationConfigExt, StorageVersioningConfigExt,
-    TierConfigMgr, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, WorkloadAdmissionSnapshotProviderRef, WriteEncryption,
+    TONIC_RPC_PREFIX, TierConfigMgr, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, WorkloadAdmissionSnapshotProviderRef,
-    WritePlan, access_consumer, add_object_lock_years, all_local_disk, all_local_disk_path, check_retention_for_modification,
+    WriteEncryption, WritePlan, access_consumer, add_object_lock_years, all_local_disk, all_local_disk_path,
-    collect_local_metrics, compression_metadata_value, contract, decode_tags, decode_tags_to_map, delete_bucket_metadata_config,
+    check_retention_for_modification, collect_local_metrics, compression_metadata_value, contract, decode_tags,
-    delete_bucket_metadata_config_if_incarnation, disk_drive_path, disk_endpoint, ecfs_consumer, ecfs_extend_consumer,
+    decode_tags_to_map, delete_bucket_metadata_config, delete_bucket_metadata_config_if_incarnation, disk_drive_path,
-    ecstore_admin, ecstore_bucket, ecstore_capacity, ecstore_client, ecstore_cluster, ecstore_compression, ecstore_config,
+    disk_endpoint, ecfs_consumer, ecfs_extend_consumer, ecstore_admin, ecstore_bucket, ecstore_capacity, ecstore_client,
-    ecstore_data_usage, ecstore_disk, ecstore_error, ecstore_event, ecstore_layout, ecstore_metrics, ecstore_notification,
+    ecstore_cluster, ecstore_compression, ecstore_config, ecstore_data_usage, ecstore_disk, ecstore_error, ecstore_event,
-    ecstore_rebalance, ecstore_rio, ecstore_rpc, ecstore_set_disk, ecstore_storage, ecstore_tier, encode_tags,
+    ecstore_layout, ecstore_metrics, ecstore_notification, ecstore_rebalance, ecstore_rio, ecstore_rpc, ecstore_set_disk,
-    find_local_disk_by_ref, get_bucket_accelerate_config, get_bucket_cors_config, get_bucket_logging_config, get_bucket_metadata,
+    ecstore_storage, ecstore_tier, encode_tags, find_local_disk_by_ref, get_bucket_accelerate_config, get_bucket_cors_config,
-    get_bucket_notification_config, get_bucket_object_lock_config, get_bucket_request_payment_config, get_bucket_sse_config,
+    get_bucket_logging_config, get_bucket_metadata, get_bucket_notification_config, get_bucket_object_lock_config,
-    get_bucket_website_config, get_local_server_property, get_lock_acquire_timeout, head_prefix_consumer, helper_consumer,
+    get_bucket_replication_config, get_bucket_request_payment_config, get_bucket_sse_config, get_bucket_website_config,
-    init_background_replication, init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx,
+    get_local_server_property, get_lock_acquire_timeout, head_prefix_consumer, helper_consumer, init_background_replication,
-    init_lock_clients, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class,
+    init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx, init_lock_clients,
-    options_consumer, prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, rpc_consumer,
+    is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class, options_consumer,
-    runtime_sources_consumer, s3_api_consumer, serialize, table_catalog_path_hash, to_s3s_etag,
+    prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, rpc_consumer, runtime_sources_consumer,
-    topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, try_migrate_iam_config,
+    s3_api_consumer, serialize, table_catalog_path_hash, to_s3s_etag, topology_snapshot_from_endpoint_pools_with_capabilities,
-    try_migrate_server_config, update_bucket_metadata_config, update_bucket_metadata_config_if_incarnation, verify_rpc_signature,
+    try_migrate_bucket_metadata, try_migrate_iam_config, try_migrate_server_config, update_bucket_metadata_config,
-    wrap_reader,
+    update_bucket_metadata_config_if_incarnation, verify_rpc_signature, wrap_reader,
 };
 
 #[cfg(test)]

rustfs/src/storage/options.rs

@@ -19,10 +19,9 @@ use http::{HeaderMap, HeaderValue};
 use rustfs_utils::http::{
     AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_FORCE_DELETE, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP,
     SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_SSEC_CRC,
-    SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_PROXY_REQUEST,
+    SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP,
-    SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
+    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP,
-    SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID,
+    SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, SUFFIX_TAGGING_TIMESTAMP, get_header,
-    SUFFIX_TAGGING_TIMESTAMP, get_header,
     header_compat::{MINIO_ENCRYPTION_PREFIX, RUSTFS_ENCRYPTION_PREFIX},
     insert_header_map, insert_str,
     metadata_compat::{MINIO_INTERNAL_PREFIX, RUSTFS_INTERNAL_PREFIX},
@@ -277,19 +276,6 @@ pub async fn get_opts(
     // Background scanner still performs full integrity checks asynchronously.
     opts.skip_verify_bitrot = get_skip_verify_bitrot();
 
-    // Anti-loop markers for the replication read proxy
-    // (`{x-rustfs-,x-minio-}source-proxy-request` header family).
-    // MinIO semantics: the header being PRESENT at all (`ProxyHeaderSet`)
-    // disables proxying, whatever its value — a peer's replication worker
-    // sends "false" on its convergence HEADs so the receiver answers locally
-    // instead of proxying the miss back (a proxied echo would fake
-    // convergence and the object would never replicate). Deliberately not
-    // gated on replication authorization: the header only disables proxying
-    // (it grants nothing).
-    let proxy_header = get_header(headers, SUFFIX_SOURCE_PROXY_REQUEST);
-    opts.proxy_header_set = proxy_header.is_some();
-    opts.proxy_request = proxy_header.map(|v| v.as_ref() == "true").unwrap_or_default();
-
     fill_conditional_writes_opts_from_header(headers, &mut opts)?;
 
     Ok(opts)
@@ -2558,80 +2544,4 @@ mod tests {
             }
         }
     }
-
-    /// The replication read-proxy anti-loop markers must be honored under
-    /// both interop prefixes (a MinIO peer sends x-minio-, a RustFS peer
-    /// sends both). `proxy_request` is set only for the literal value
-    /// "true", while `proxy_header_set` (MinIO `ProxyHeaderSet`) is set by
-    /// the header's mere presence — "false" (the replication worker's
-    /// convergence-HEAD marker) and arbitrary values included — so the
-    /// selector refuses to proxy either way.
-    #[tokio::test]
-    async fn test_get_opts_parses_source_proxy_request_under_both_prefixes() {
-        for header_name in ["x-rustfs-source-proxy-request", "x-minio-source-proxy-request"] {
-            let mut headers = HeaderMap::new();
-            headers.insert(header_name, HeaderValue::from_static("true"));
-            let opts = get_opts("test-bucket", "test-object", None, None, &headers)
-                .await
-                .expect("get_opts should succeed");
-            assert!(opts.proxy_request, "{header_name} must set opts.proxy_request");
-            assert!(opts.proxy_header_set, "{header_name} must set opts.proxy_header_set");
-        }
-
-        let opts = get_opts("test-bucket", "test-object", None, None, &HeaderMap::new())
-            .await
-            .expect("get_opts should succeed");
-        assert!(!opts.proxy_request, "absent header must leave proxy_request off");
-        assert!(!opts.proxy_header_set, "absent header must leave proxy_header_set off");
-
-        for (header_name, value) in [
-            ("x-minio-source-proxy-request", "false"),
-            ("x-rustfs-source-proxy-request", "false"),
-            ("x-minio-source-proxy-request", "anything-else"),
-        ] {
-            let mut headers = HeaderMap::new();
-            headers.insert(header_name, HeaderValue::from_static(value));
-            let opts = get_opts("test-bucket", "test-object", None, None, &headers)
-                .await
-                .expect("get_opts should succeed");
-            assert!(!opts.proxy_request, "{header_name}: non-'true' value must leave proxy_request off");
-            assert!(
-                opts.proxy_header_set,
-                "{header_name}: value {value:?} must still set proxy_header_set (presence disables proxying)"
-            );
-        }
-    }
-
-    /// Pin that the source-proxy-request transport family cannot be
-    /// materialized as bare stored metadata via an `x-*-meta-` disguise: the
-    /// reserved-key namespacing (`x-rustfs-source-` / `x-minio-source-`
-    /// prefixes in `is_reserved_user_metadata_key`) must keep covering it.
-    #[test]
-    fn test_source_proxy_request_family_is_reserved_user_metadata() {
-        let mut headers = HeaderMap::new();
-        headers.insert("x-amz-meta-x-minio-source-proxy-request", HeaderValue::from_static("true"));
-        headers.insert("x-rustfs-meta-x-rustfs-source-proxy-request", HeaderValue::from_static("true"));
-        // The bare transport header itself is not a user-metadata prefix and
-        // must never land in stored metadata at all.
-        headers.insert("x-minio-source-proxy-request", HeaderValue::from_static("true"));
-
-        let metadata = extract_metadata(&headers);
-
-        assert!(
-            !metadata.contains_key("x-minio-source-proxy-request"),
-            "bare source-proxy-request key must not be storable: {metadata:?}"
-        );
-        assert!(
-            !metadata.contains_key("x-rustfs-source-proxy-request"),
-            "bare source-proxy-request key must not be storable: {metadata:?}"
-        );
-        assert!(
-            metadata.contains_key("x-amz-meta-x-minio-source-proxy-request"),
-            "disguised key must be namespaced back under x-amz-meta-: {metadata:?}"
-        );
-        assert!(
-            metadata.contains_key("x-amz-meta-x-rustfs-source-proxy-request"),
-            "disguised key must be namespaced back under x-amz-meta-: {metadata:?}"
-        );
-    }
 }

rustfs/src/storage/storage_api.rs

@@ -805,10 +805,6 @@ impl StorageReplicationStatsHandle {
             proxy_head_failed: metrics.proxied.head_failed,
             proxy_put_tag_total: metrics.proxied.put_tag_total,
             proxy_put_tag_failed: metrics.proxied.put_tag_failed,
-            proxy_get_tag_total: metrics.proxied.get_tag_total,
-            proxy_get_tag_failed: metrics.proxied.get_tag_failed,
-            proxy_delete_tag_total: metrics.proxied.delete_tag_total,
-            proxy_delete_tag_failed: metrics.proxied.delete_tag_failed,
             replica_size: metrics.replica_size,
             replica_count: metrics.replica_count,
         }
@@ -845,10 +841,6 @@ pub(crate) struct ReplicationSiteMetricsSnapshot {
     pub(crate) proxy_head_failed: i64,
     pub(crate) proxy_put_tag_total: i64,
     pub(crate) proxy_put_tag_failed: i64,
-    pub(crate) proxy_get_tag_total: i64,
-    pub(crate) proxy_get_tag_failed: i64,
-    pub(crate) proxy_delete_tag_total: i64,
-    pub(crate) proxy_delete_tag_failed: i64,
     pub(crate) replica_size: i64,
     pub(crate) replica_count: i64,
 }
@@ -1495,6 +1487,12 @@ pub(crate) async fn get_bucket_object_lock_config(
     ecstore_bucket::metadata_sys::get_object_lock_config(bucket).await
 }
 
+pub(crate) async fn get_bucket_replication_config(
+    bucket: &str,
+) -> Result<(s3s::dto::ReplicationConfiguration, time::OffsetDateTime)> {
+    ecstore_bucket::metadata_sys::get_replication_config(bucket).await
+}
+
 pub(crate) async fn persist_force_delete_intent(
     api: Arc<ECStore>,
     entry: ecstore_bucket::replication::MrfReplicateEntry,
@@ -1840,6 +1838,18 @@ pub(crate) async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore>
     ecstore_storage::find_local_disk_by_ref(disk_ref).await
 }
 
+pub(crate) trait StorageReplicationConfigExt {
+    fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
+}
+
+impl StorageReplicationConfigExt for s3s::dto::ReplicationConfiguration {
+    fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool {
+        <s3s::dto::ReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::has_active_rules(
+            self, prefix, recursive,
+        )
+    }
+}
+
 pub(crate) trait StorageVersioningConfigExt {
     fn enabled(&self) -> bool;
 }