test(interop): fix the fixture lab's all-cases and SSE-C capture paths

Two defects kept most of the fixture matrix uncapturable, both found by actually running the lab rather than reading it.

`./capture_via_docker.sh all` never worked. With no case ids to forward, `case_args` is an empty array, and `${arr[@]}` under `set -u` aborts on bash 3.2 — still the default /bin/bash on macOS — with "case_args[@]: unbound variable". The build_args array this branch added has the same shape and would have hit it on the CI path, where no mirror overrides are set. Both now use ${arr[@]+"${arr[@]}"}, which is empty-safe on 3.2.

The SSE-C cases could not be captured at all: MinIO refuses SSE-C over a plain-HTTP connection, and the script pinned the lab to its default http endpoint with no way to override. lab.py already provisions a self-signed certificate and speaks https end to end, so this only needed an endpoint passthrough — MINIO_LAB_ENDPOINT, documented alongside the requirement.

With both fixed, one command captures the full six-case matrix (SSE-S3 / SSE-KMS / SSE-C x singlepart / multipart), which is what backlog#1638's D4 matrix has to be measured against.

Refs rustfs/backlog#1638.

crates/checksums/src/base64.rs

@@ -11,6 +11,7 @@
 // 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.
+#![allow(dead_code)]
 
 use base64_simd::STANDARD;
 

crates/checksums/src/http.rs

@@ -38,10 +38,7 @@ pub const XXHASH_3_HEADER_NAME: &str = "x-amz-checksum-xxhash3";
 pub const XXHASH_64_HEADER_NAME: &str = "x-amz-checksum-xxhash64";
 pub const XXHASH_128_HEADER_NAME: &str = "x-amz-checksum-xxhash128";
 
-#[allow(
-    dead_code,
-    reason = "Content-MD5 wire name, resolved by header_name() below and asserted by this crate's tests (backlog#1823)"
-)]
+#[allow(dead_code)]
 pub(crate) static MD5_HEADER_NAME: &str = "content-md5";
 
 pub const CHECKSUM_ALGORITHMS_IN_PRIORITY_ORDER: [&str; 5] =

crates/checksums/src/lib.rs

@@ -476,19 +476,13 @@ impl Checksum for Xxhash64 {
     }
 }
 
+#[allow(dead_code)]
 #[derive(Debug, Default)]
-#[allow(
-    dead_code,
-    reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
-)]
 struct Md5 {
     hasher: md5::Md5,
 }
 
-#[allow(
-    dead_code,
-    reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
-)]
+#[allow(dead_code)]
 impl Md5 {
     fn update(&mut self, bytes: &[u8]) {
         use md5::Digest;

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,
-    DeleteObjectTaggingInput, DeleteObjectTaggingOutput, ETag, GetBucketVersioningInput, GetBucketVersioningOutput,
-    GetObjectInput, GetObjectOutput, GetObjectTaggingInput, GetObjectTaggingOutput, HeadBucketInput, HeadBucketOutput,
+    CreateMultipartUploadInput, CreateMultipartUploadOutput, DeleteMarkerEntry, DeleteObjectInput, DeleteObjectOutput, ETag,
+    GetBucketVersioningInput, GetBucketVersioningOutput, GetObjectInput, GetObjectOutput, HeadBucketInput, HeadBucketOutput,
     HeadObjectInput, HeadObjectOutput, ListObjectVersionsInput, ListObjectVersionsOutput, ObjectVersionId, PutObjectInput,
-    PutObjectOutput, PutObjectTaggingInput, PutObjectTaggingOutput, StreamingBlob, Tag, TagSet, Timestamp, TimestampFormat,
-    UploadPartInput, UploadPartOutput,
+    PutObjectOutput, StreamingBlob, Timestamp, TimestampFormat, 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/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,
-            invalid_replication_config_status_field, persist_force_delete_intent, read_durable_mrf_backlog,
-            replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns,
-            resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication,
-            should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
-            unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns,
-            version_purge_status_to_filemeta,
+            get_global_replication_stats, init_background_replication, invalid_replication_config_status_field,
+            persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta,
+            replication_statuses_map, replication_target_arns, resync_start_conflict_id, should_remove_replication_target,
+            should_schedule_delete_replication, should_use_existing_delete_replication_info,
+            should_use_existing_delete_replication_source, unsupported_replication_config_field,
+            validate_replication_config_structure, validate_replication_config_target_arns, 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_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_LEGALHOLD_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST,
+    SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK,
+    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,
 };
@@ -87,6 +82,7 @@ use tracing::warn;
 use url::Url;
 use uuid::Uuid;
 
+const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60);
 const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
 const REDACTED_CREDENTIAL: &str = "<redacted>";
 
@@ -1451,43 +1447,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 +1852,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 +1876,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/lifecycle/bucket_lifecycle_ops.rs

@@ -126,23 +126,11 @@ const EVENT_LIFECYCLE_EXPIRED_DETECTED: &str = "lifecycle_expired_detected";
 const EVENT_LIFECYCLE_NOT_ENQUEUED: &str = "lifecycle_not_enqueued";
 const EVENT_LIFECYCLE_DELETE_DISPATCHED: &str = "lifecycle_delete_dispatched";
 const EVENT_LIFECYCLE_DELETE_COMPLETED: &str = "lifecycle_delete_completed";
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 const EVENT_LIFECYCLE_TIER_AUDIT: &str = "lifecycle_tier_audit";
 const EVENT_LIFECYCLE_TIER_OPERATION_FAILED: &str = "lifecycle_tier_operation_failed";
 const EVENT_LIFECYCLE_DELETE_FAILED: &str = "lifecycle_delete_failed";
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub type TimeFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub type TraceFn =
     Arc<dyn Fn(String, HashMap<String, String>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
 pub type ExpiryOpType = Box<dyn ExpiryOp + Send + Sync + 'static>;
@@ -152,21 +140,9 @@ static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
 static MANUAL_TRANSITION_JOB_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
 
 pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count";
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub const AMZ_TAG_DIRECTIVE: &str = "X-Amz-Tagging-Directive";
 pub const AMZ_ENCRYPTION_AES: &str = "AES256";
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms";
 
 pub const ERR_INVALID_STORAGECLASS: &str = "invalid tier.";
@@ -304,10 +280,6 @@ impl LifecycleSys {
         }
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-    )]
     pub fn trace(oi: &ObjectInfo) -> TraceFn {
         let bucket = oi.bucket.clone();
         let name = oi.name.clone();
@@ -598,10 +570,6 @@ async fn delete_free_version_remote_object(
     Ok(())
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 async fn delete_free_version_remote_object_then<T, F, Fut>(
     oi: &ObjectInfo,
     tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
@@ -2900,10 +2868,6 @@ fn stale_upload_default_due(initiated: OffsetDateTime, default_expiry: StdDurati
     initiated + time::Duration::seconds(default_expiry.as_secs() as i64)
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 async fn stale_upload_current_size(set: &Arc<SetDisks>, metadata: &HashMap<String, String>, upload_dir: &str) -> Option<usize> {
     stale_upload_current_size_with_opts(set, metadata, upload_dir, false).await
 }
@@ -3388,10 +3352,6 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu
     Ok(())
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 fn mark_delete_opts_skip_decommissioned_on_remote_success(opts: &mut ObjectOptions, remote_delete_succeeded: bool) {
     if remote_delete_succeeded {
         opts.skip_decommissioned = true;
@@ -4379,10 +4339,6 @@ pub async fn expire_transitioned_object(
     Ok(dobj)
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub fn gen_transition_objname(bucket: &str) -> Result<String, Error> {
     let us = Uuid::new_v4().to_string();
     let mut hasher = Sha256::new();
@@ -4417,10 +4373,6 @@ pub async fn transition_object(api: Arc<ECStore>, oi: &ObjectInfo, lae: LcAuditE
     result
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
     let tier = _tier.to_string();
     Arc::new(move || {
@@ -4439,10 +4391,6 @@ pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
     })
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn get_transitioned_object_reader(
     bucket: &str,
     object: &str,
@@ -5197,10 +5145,6 @@ async fn lifecycle_delete_config_snapshot(api: &ECStore, oi: &ObjectInfo) -> Res
     ReplicationObjectBridge::delete_request_config(api, &oi.bucket).await
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
     let mut success = false;
     match event.action {
@@ -7478,10 +7422,6 @@ mod tests {
     // process environment while `env::set_var`/`env::remove_var` is active.
     // SAFETY: keep this note adjacent to the allowance for the repository guard.
     #[allow(unsafe_code)]
-    #[allow(
-        dead_code,
-        reason = "transition-queue env fixture kept for tests that scope those vars; no test uses it today (backlog#1823)"
-    )]
     async fn with_transition_queue_env_async<F, Fut>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
     where
         F: FnOnce() -> Fut,

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

@@ -759,10 +759,6 @@ pub struct ManualTransitionWorkerResultRecord {
 }
 
 impl ManualTransitionWorkerResultRecord {
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-    )]
     pub fn new(job_id: Uuid, task_key: impl Into<String>, result: ManualTransitionWorkerResult) -> Self {
         Self::new_with_reason(job_id, task_key, result, None)
     }
@@ -1261,10 +1257,6 @@ pub(crate) async fn save_manual_transition_task_if_absent(
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn load_manual_transition_task_record(
     api: Arc<ECStore>,
     job_id: Uuid,
@@ -1328,10 +1320,6 @@ async fn scan_manual_transition_task_journal(api: Arc<ECStore>, job_id: Uuid) ->
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn load_manual_transition_worker_result_stats(
     api: Arc<ECStore>,
     job_id: Uuid,
@@ -1467,10 +1455,6 @@ async fn scan_manual_transition_worker_result_journal(
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn reconcile_manual_transition_worker_results(
     api: Arc<ECStore>,
     job_id: Uuid,

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

@@ -15,35 +15,25 @@
 use rustfs_common::metrics::IlmAction;
 
 use crate::bucket::lifecycle::lifecycle::ObjectOpts;
-use crate::bucket::replication::ReplicationLifecycleBridge;
 pub(crate) use crate::bucket::replication::ReplicationStatusType;
 #[cfg(test)]
 pub(crate) use crate::bucket::replication::VersionPurgeStatusType;
 pub(crate) use crate::bucket::replication::{
     DeleteReplicationConfigSnapshot, ReplicationObjectBridge, replication_state_to_filemeta,
 };
+use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
 use crate::storage_api_contracts::object::DeletedObject;
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
+pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig;
+
 pub(crate) fn has_pending_version_purge(obj: &ObjectOpts) -> bool {
     obj.version_purge_status.is_pending()
 }
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) fn has_pending_object_replication(obj: &ObjectOpts) -> bool {
     replication_status_blocks_lifecycle(&obj.replication_status)
 }
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) fn has_pending_lifecycle_replication(obj: &ObjectOpts) -> bool {
     has_pending_object_replication(obj) || has_pending_version_purge(obj)
 }

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

@@ -14,10 +14,6 @@
 
 use std::collections::HashMap;
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) fn decode_tags_to_map(tags: &str) -> HashMap<String, String> {
     crate::bucket::tagging::decode_tags_to_map(tags)
 }

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

@@ -331,10 +331,6 @@ where
     persist_tier_delete_journal_entry(api, &committed).await
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn abort_tier_delete_journal_entry<S>(api: Arc<S>, je: &Jentry) -> std::io::Result<()>
 where
     S: ObjectOperations<

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

@@ -148,10 +148,6 @@ struct RecoveryCursor {
     object: String,
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn recover_tier_free_versions(
     api: Arc<ECStore>,
     limit: usize,

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

@@ -385,10 +385,6 @@ impl ExpiryOp for Jentry {
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
     let result = delete_object_from_remote_tier_raw(obj_name, rv_id, tier_name).await;
     if let Err(err) = &result
@@ -399,10 +395,6 @@ pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_na
     result
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
     #[cfg(test)]
     if let Some(result) = run_remote_tier_delete_test_hook(obj_name, rv_id, tier_name) {
@@ -413,10 +405,6 @@ async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_na
     delete_object_from_remote_tier_raw_with_manager(obj_name, rv_id, tier_name, &tier_config_mgr).await
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 async fn delete_object_from_remote_tier_raw_with_manager(
     obj_name: &str,
     rv_id: &str,
@@ -497,10 +485,6 @@ pub enum RemoteTierDeleteOutcome {
     AlreadyRemoved,
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-)]
 pub async fn delete_object_from_remote_tier_idempotent(
     obj_name: &str,
     rv_id: &str,

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

@@ -50,16 +50,8 @@ pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
 #[derive(Debug, thiserror::Error)]
 pub enum TransitionTransactionError {
     #[error("transition transaction already exists")]
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-    )]
     AlreadyExists,
     #[error("transition transaction is not found")]
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
-    )]
     NotFound,
     #[error("transition transaction is corrupt: {0}")]
     Corrupt(&'static str),

crates/ecstore/src/bucket/metadata_sys.rs

@@ -60,14 +60,12 @@ struct ConfigWriteLockProbeState {
 static CONFIG_WRITE_LOCK_PROBES: std::sync::OnceLock<StdMutex<Vec<Arc<ConfigWriteLockProbeState>>>> = std::sync::OnceLock::new();
 
 #[cfg(any(test, feature = "test-util"))]
-#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
 pub struct ConfigWriteLockProbe {
     state: Arc<ConfigWriteLockProbeState>,
 }
 
 #[cfg(any(test, feature = "test-util"))]
 impl ConfigWriteLockProbe {
-    #[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
     pub fn install(bucket: &str) -> Self {
         let state = Arc::new(ConfigWriteLockProbeState {
             bucket: bucket.to_string(),
@@ -86,7 +84,6 @@ impl ConfigWriteLockProbe {
         Self { state }
     }
 
-    #[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
     pub async fn wait_until_attempted(&self) {
         tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
             .await
@@ -893,10 +890,6 @@ pub async fn get_bucket_policy_raw(bucket: &str) -> Result<(String, OffsetDateTi
     bucket_meta_sys.get_bucket_policy_raw(bucket).await
 }
 
-#[allow(
-    dead_code,
-    reason = "free-function facade over the live BucketMetadataSys::get_bucket_acl_config; no caller in this port (backlog#1823)"
-)]
 pub async fn get_bucket_acl_config(bucket: &str) -> Result<(String, OffsetDateTime)> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1111,10 +1104,6 @@ pub async fn get_config_from_disk(bucket: &str) -> Result<BucketMetadata> {
     bucket_meta_sys.get_config_from_disk(bucket).await
 }
 
-#[allow(
-    dead_code,
-    reason = "ambient-facade variant of the live created_at_in; no caller in this port (backlog#1823)"
-)]
 pub async fn created_at(bucket: &str) -> Result<OffsetDateTime> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1628,7 +1617,6 @@ impl BucketMetadataSys {
     /// [`Self::update`], with the payload computed from the loaded metadata
     /// instead of supplied up front. Loads through this system's own store so
     /// the read and the persisted write target the same instance.
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     async fn update_config_with<F>(&self, bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
     where
         F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
@@ -1733,7 +1721,6 @@ impl BucketMetadataSys {
     /// A miss is never published as an authoritative default, and a snapshot
     /// read before delete plus same-name recreation cannot replace the new
     /// generation.
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
         if is_meta_bucketname(bucket) {
             return Err(Error::other("errInvalidArgument"));

crates/ecstore/src/bucket/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 // #730: bucket subsystems still contain staged ECStore migration code.
+#![allow(dead_code)]
 
 pub mod bandwidth;
 pub mod bucket_target_sys;

crates/ecstore/src/bucket/object_lock/objectlock_sys.rs

@@ -136,7 +136,6 @@ pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
 
 /// Check if an object has legal hold enabled.
 /// Returns true if legal hold is ON.
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> bool {
     let lhold = objectlock::get_object_legalhold_meta(user_defined);
     matches!(lhold.status, Some(ref st) if st.as_str() == ObjectLockLegalHoldStatus::ON)
@@ -152,7 +151,6 @@ fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> b
 /// # Returns
 /// * `true` if the object is locked (cannot be deleted/modified)
 /// * `false` if the object is not locked
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn is_object_locked_by_metadata(user_defined: &std::collections::HashMap<String, String>, is_delete_marker: bool) -> bool {
     // Delete markers are never locked
     if is_delete_marker {

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

@@ -193,7 +193,6 @@ pub enum QuotaError {
 }
 
 #[derive(Debug, Serialize)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub struct QuotaErrorResponse {
     #[serde(rename = "Code")]
     pub code: String,
@@ -209,7 +208,6 @@ pub struct QuotaErrorResponse {
 }
 
 impl QuotaErrorResponse {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn new(quota_error: &QuotaError, request_id: &str, host_id: &str) -> Self {
         match quota_error {
             QuotaError::QuotaExceeded { .. } => Self {

crates/ecstore/src/bucket/quota/reservation.rs

@@ -899,7 +899,6 @@ async fn save_ledger_locked(
 }
 
 #[cfg(any(test, feature = "test-util"))]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn fail_next_quota_ledger_save_for_test() {
     FAIL_NEXT_LEDGER_SAVE.store(true, std::sync::atomic::Ordering::SeqCst);
 }

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;
@@ -61,7 +60,7 @@ pub use replication_filemeta_boundary::{
 pub(crate) use replication_filemeta_boundary::{
     replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta,
 };
-pub(crate) use replication_lifecycle_bridge::ReplicationLifecycleBridge;
+pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
 pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
 pub use replication_object_bridge::ReplicationObjectBridge;
 pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig};
@@ -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_config_store.rs

@@ -37,10 +37,6 @@ impl ReplicationConfigStore {
         com::read_config_limited(api, file, max_bytes).await
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     pub(crate) async fn read_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
     where
         S: ReplicationObjectIO,

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

@@ -24,27 +24,15 @@ use super::replication_storage_boundary::{
     DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, deleted_object_for_replication,
 };
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) type ReplicationLifecycleConfig = ReplicationConfig;
 
 pub(crate) struct ReplicationLifecycleBridge;
 
 impl ReplicationLifecycleBridge {
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn new_config(config: ReplicationConfiguration) -> ReplicationLifecycleConfig {
         ReplicationConfig::new(Some(config), None)
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn has_pending_version_purge(
         config: &ReplicationLifecycleConfig,
         object_name: &str,
@@ -57,10 +45,6 @@ impl ReplicationLifecycleBridge {
                 .is_some_and(|config| config.has_active_rules(object_name, true))
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) async fn check_delete_replication(
         bucket: &str,
         object: &ObjectToDelete,
@@ -70,10 +54,6 @@ impl ReplicationLifecycleBridge {
         check_replicate_delete(bucket, object, source, opts, None).await
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn version_delete_replication_state(decision: &ReplicateDecision) -> ReplicationState {
         let pending_status = decision.pending_status();
         ReplicationState {

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

@@ -19,33 +19,17 @@ use time::OffsetDateTime;
 use super::replication_error_boundary::Result;
 use crate::bucket::msgp_decode;
 
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) struct ReplicationMsgpCodec;
 
 impl ReplicationMsgpCodec {
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn read_ext8_time<R: Read>(rd: &mut R) -> Result<OffsetDateTime> {
         msgp_decode::read_msgp_ext8_time(rd)
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn skip_value<R: Read>(rd: &mut R) -> Result<()> {
         msgp_decode::skip_msgp_value(rd)
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn write_time<W: Write>(wr: &mut W, time: OffsetDateTime) -> Result<()> {
         msgp_decode::write_msgp_time(wr, time)
     }

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

@@ -77,10 +77,6 @@ impl ReplicationObjectBridge {
         load_delete_request_config_in(ctx, bucket).await
     }
 
-    #[allow(
-        dead_code,
-        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-    )]
     pub(crate) async fn delete_config_snapshot_in(
         ctx: &ReplicationInstanceContext,
         bucket: &str,

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

@@ -231,10 +231,6 @@ pub(crate) async fn load_delete_replication_config(
     delete_snapshot_from_metadata(ReplicationMetadataStore::delete_metadata(bucket).await?)
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-)]
 pub(crate) async fn load_delete_replication_config_in(
     ctx: &ReplicationInstanceContext,
     bucket: &str,

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

@@ -217,10 +217,6 @@ impl DurableMrfBacklogTracker {
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-)]
 fn durable_mrf_backlog_tracker_from_entries(entries: &[MrfReplicateEntry]) -> DurableMrfBacklogTracker {
     let mut tracker = DurableMrfBacklogTracker {
         available: true,
@@ -667,368 +663,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 {
@@ -1078,10 +712,6 @@ pub struct ReplicationPool<S: ReplicationStorage> {
 
     // MRF worker lifecycle
     mrf_worker_cancellations: Mutex<Vec<CancellationToken>>,
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     mrf_stop_tx: Sender<()>,
 
     // Worker size tracking
@@ -1310,10 +940,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Resizes worker priority and counts
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     pub async fn resize_worker_priority(
         &self,
         pri: ReplicationPriority,
@@ -1554,10 +1180,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Queues an MRF save operation
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
         let _ = self.queue_mrf_save_admission(entry, "mrf_worker").await;
     }
@@ -1583,12 +1205,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 {
-                return;
+            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;
+                }
+            };
+            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 {
-                return;
+            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;
+                }
+                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 +1278,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 {
-                    continue;
+                let admission = match entry.op {
+                    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 +1468,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));
 
@@ -1769,10 +1651,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Worker function for handling regular replication operations
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     async fn add_worker(
         &self,
         mut rx: Receiver<ReplicationOperation>,
@@ -1786,10 +1664,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Worker function for handling large object replication operations
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     async fn add_large_worker(
         &self,
         mut rx: Receiver<ReplicationOperation>,
@@ -1804,10 +1678,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Worker function for handling MRF (Most Recent Failures) operations
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     async fn add_mrf_worker(
         &self,
         mut rx: Receiver<ReplicationOperation>,
@@ -1821,10 +1691,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Delete resync metadata from replication resync state in memory
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     pub async fn delete_resync_metadata(&self, bucket: &str) {
         let mut status_map = self.resyncer.status_map.write().await;
         status_map.remove(bucket);

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,41 +15,17 @@
 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(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT;
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const RESYNC_META_VERSION: u16 = rustfs_replication::resync::RESYNC_META_VERSION;
 pub(crate) const RESYNC_FILE_MAX_BYTES: usize = rustfs_replication::RESYNC_FILE_MAX_BYTES;
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const WIRE_ZERO_TIME_UNIX: i64 = rustfs_replication::resync::WIRE_ZERO_TIME_UNIX;
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const MRF_META_FORMAT: u16 = rustfs_replication::mrf::MRF_META_FORMAT;
-#[allow(
-    dead_code,
-    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
-)]
 pub(crate) const MRF_META_VERSION: u16 = rustfs_replication::mrf::MRF_META_VERSION;
 
 fn map_replication_error(err: rustfs_replication::Error) -> Error {

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

@@ -340,10 +340,6 @@ impl ReplicationStats {
     }
 
     /// Site replication update replica statistics
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
-    )]
     fn sr_update_replica_stat(&self, size: i64) {
         self.sr_stats.replica_size.fetch_add(size, Ordering::Relaxed);
         self.sr_stats.replica_count.fetch_add(1, Ordering::Relaxed);
@@ -1161,31 +1157,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/bucket/target/bucket_target.rs

@@ -59,10 +59,6 @@ impl fmt::Debug for Credentials {
 }
 
 #[derive(Debug, Deserialize, Serialize, Default, Clone)]
-#[allow(
-    dead_code,
-    reason = "MinIO-parity bucket-target service discriminator with no caller in this port (backlog#1823)"
-)]
 pub enum ServiceType {
     #[default]
     Replication,

crates/ecstore/src/bucket/utils.rs

@@ -73,6 +73,23 @@ pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<()> {
     check_bucket_name_common(bucket_name, true)
 }
 
+pub fn check_valid_object_name_prefix(object_name: &str) -> Result<()> {
+    if object_name.len() > 1024 {
+        return Err(Error::other("Object name cannot be longer than 1024 characters"));
+    }
+    if !object_name.is_ascii() {
+        return Err(Error::other("Object name with non-UTF-8 strings are not supported"));
+    }
+    Ok(())
+}
+
+pub fn check_valid_object_name(object_name: &str) -> Result<()> {
+    if object_name.trim().is_empty() {
+        return Err(Error::other("Object name cannot be empty"));
+    }
+    check_valid_object_name_prefix(object_name)
+}
+
 pub fn deserialize<T>(input: &[u8]) -> xml::DeResult<T>
 where
     T: for<'xml> xml::Deserialize<'xml>,
@@ -83,10 +100,6 @@ where
     Ok(ans)
 }
 
-#[allow(
-    dead_code,
-    reason = "xml serialize helper with no caller in this port; the live sibling is deserialize (backlog#1823)"
-)]
 pub fn serialize_content<T: xml::SerializeContent>(val: &T) -> xml::SerResult<String> {
     let mut buf = Vec::with_capacity(256);
     {
@@ -173,27 +186,15 @@ pub fn is_valid_object_name(object: &str) -> bool {
 /// Client-facing reason attached to rejections of object keys that Win32/NTFS
 /// cannot represent as file paths (issue #3299). Deployments on Linux/macOS
 /// accept the full S3 key character set.
-#[allow(
-    dead_code,
-    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
-)]
 pub const WINDOWS_RESERVED_CHARACTERS_REASON: &str =
     "object key contains characters unsupported on Windows hosts (one of ':', '*', '?', '\"', '|', '<', '>')";
 
 /// Client-facing reason for path segments Windows can store but not address
 /// afterwards (issue #3449): trailing dot/space or reserved DOS device names.
-#[allow(
-    dead_code,
-    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
-)]
 pub const WINDOWS_RESERVED_SEGMENT_REASON: &str = "object key contains a path segment unsupported on Windows hosts (trailing dot or space, or a reserved device name such as NUL/CON/COM1)";
 
 /// Reserved DOS device names that shadow regular files on Windows, even when
 /// an extension is appended (e.g. `NUL.txt` resolves to the `NUL` device).
-#[allow(
-    dead_code,
-    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
-)]
 const WINDOWS_RESERVED_NAMES: &[&str] = &[
     "CON", "PRN", "AUX", "NUL", "COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9", "LPT1", "LPT2", "LPT3",
     "LPT4", "LPT5", "LPT6", "LPT7", "LPT8", "LPT9",
@@ -203,10 +204,6 @@ const WINDOWS_RESERVED_NAMES: &[&str] = &[
 /// the Win32 API cannot address afterwards (issue #3449): segments ending in a
 /// dot or a space, and reserved DOS device names — bare or with an extension
 /// (`NUL.txt`), matching classic Win32 path resolution semantics.
-#[allow(
-    dead_code,
-    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
-)]
 pub fn object_name_has_windows_incompatible_segment(object: &str) -> bool {
     object.split(['/', '\\']).any(|segment| {
         if segment.ends_with('.') || segment.ends_with(' ') {

crates/ecstore/src/bucket/versioning_sys.rs

@@ -90,10 +90,6 @@ impl BucketVersioningSys {
     /// caller's own instance context so a second in-process store never
     /// answers with the first instance's versioning state; falls back to the
     /// ambient system when the instance cell is not initialized.
-    #[allow(
-        dead_code,
-        reason = "instance-scoped seam (backlog#1052) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result<VersioningConfiguration> {
         if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) {
             return Ok(VersioningConfiguration::default());

crates/ecstore/src/disk/disk_store.rs

@@ -637,23 +637,14 @@ impl Default for DiskOperationMetrics {
 }
 
 impl DiskOperationMetrics {
-    #[allow(
-        dead_code,
-        reason = "internal metrics recorder reached only from record() below (backlog#1823)"
-    )]
     fn record_call(&mut self) {
         self.lifetime_calls.fetch_add(1, Ordering::Relaxed);
     }
 
-    #[allow(
-        dead_code,
-        reason = "internal metrics recorder reached only from record() below (backlog#1823)"
-    )]
     fn record_latency(&mut self, now_sec: u64, elapsed: Duration) {
         self.record_latency_atomic(now_sec, elapsed);
     }
 
-    #[allow(dead_code, reason = "metrics roll-up with no caller in this port (backlog#1823)")]
     fn record(&mut self, now_sec: u64, elapsed: Duration) {
         self.record_call();
         self.record_latency(now_sec, elapsed);
@@ -779,7 +770,6 @@ impl DiskHealthTracker {
     }
 
     /// Set disk as faulty
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn set_faulty(&self) {
         self.status.store(DISK_HEALTH_FAULTY, Ordering::Release);
     }
@@ -860,7 +850,6 @@ impl DiskHealthTracker {
         became_offline
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn mark_offline(&self, endpoint: &Endpoint, reason: &'static str) -> bool {
         let current = self.runtime_state();
         if current == RuntimeDriveHealthState::Offline {
@@ -991,13 +980,11 @@ impl DiskHealthTracker {
     }
 
     /// Get waiting operations count
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn waiting_count(&self) -> u32 {
         self.waiting.load(Ordering::Relaxed)
     }
 
     /// Get last success timestamp
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn last_success(&self) -> i64 {
         self.last_success.load(Ordering::Acquire)
     }
@@ -1039,6 +1026,21 @@ impl Default for DiskHealthTracker {
     }
 }
 
+/// Health check context key for tracking disk operations
+#[derive(Debug, Clone)]
+struct HealthDiskCtxKey;
+
+#[derive(Debug)]
+struct HealthDiskCtxValue {
+    last_success: Arc<AtomicI64>,
+}
+
+impl HealthDiskCtxValue {
+    fn log_success(&self) {
+        self.last_success.store(current_unix_nanos(), Ordering::Relaxed);
+    }
+}
+
 /// LocalDiskWrapper wraps a DiskStore with health tracking capabilities.
 /// This is similar to Go's xlStorageDiskIDCheck.
 #[derive(Debug, Clone)]
@@ -1070,6 +1072,10 @@ impl LocalDiskWrapper {
         )
     }
 
+    pub(crate) fn new_with_health(disk: Arc<LocalDisk>, health_check: bool, health: Arc<DiskHealthTracker>) -> Self {
+        Self::new_with_health_and_metrics(disk, health_check, health, Arc::new(DiskHealthMetricEpoch::default()))
+    }
+
     pub(crate) fn new_with_reconnect_state(
         disk: Arc<LocalDisk>,
         health_check: bool,
@@ -1432,6 +1438,20 @@ impl LocalDiskWrapper {
         }
     }
 
+    async fn check_id(&self, want_id: Option<Uuid>) -> Result<()> {
+        if want_id.is_none() {
+            return Ok(());
+        }
+
+        let stored_disk_id = self.disk.get_disk_id().await?;
+
+        if stored_disk_id != want_id {
+            return Err(Error::other(format!("Disk ID mismatch wanted {want_id:?}, got {stored_disk_id:?}")));
+        }
+
+        Ok(())
+    }
+
     /// Check if disk ID is stale
     async fn check_disk_stale(&self) -> Result<()> {
         let Some(current_disk_id) = *self.disk_id.read().await else {

crates/ecstore/src/disk/error_conv.rs

@@ -48,7 +48,6 @@ pub fn to_volume_error(io_err: std::io::Error) -> std::io::Error {
     }
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn to_disk_error(io_err: std::io::Error) -> std::io::Error {
     match io_err.kind() {
         std::io::ErrorKind::NotFound => DiskError::DiskNotFound.into(),

crates/ecstore/src/disk/fs.rs

@@ -178,7 +178,6 @@ pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
     }
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
     // Try remove_file first; fall back to remove_dir_all if it's a directory
     match fs::remove_file(path.as_ref()).await {

crates/ecstore/src/disk/local.rs

@@ -665,7 +665,6 @@ async fn remove_empty_directory_tree_under_mount_lease(
 }
 
 #[cfg(unix)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 async fn remove_empty_directory_tree_with(
     root: &Path,
     before_descend: impl FnMut(&Path) -> std::io::Result<()>,
@@ -1017,29 +1016,13 @@ fn record_direct_read_page_fault_delta(path: &'static str, stage: &'static str,
 /// When enabled, shard reads bypass the page cache using O_DIRECT flag.
 /// Requires aligned buffers (typically 512 bytes or 4096 bytes).
 /// Default: false (uses page cache via mmap/pread).
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE";
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: bool = false;
 
 /// Minimum shard size threshold for O_DIRECT reads.
 /// Only shards larger than this threshold will use O_DIRECT.
 /// Default: 4MB.
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD";
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
 
 /// Enable O_DIRECT for erasure shard / multipart part data writes (Linux only).
@@ -1053,15 +1036,7 @@ const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
 /// EINVAL/EOPNOTSUPP (tmpfs, overlayfs, 9p, ...) latch the path off and fall
 /// back to buffered writes for the whole disk. Non-Linux always falls back.
 /// Default: false (buffered writes via the page cache, as before).
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE";
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: bool = false;
 const ENV_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: &str = "RUSTFS_OBJECT_MMAP_POPULATE_ENABLE";
 const DEFAULT_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: bool = false;
@@ -1120,14 +1095,12 @@ macro_rules! cached_read_env {
 
 cached_read_env! {
     /// Check if O_DIRECT reads are enabled.
-    #[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
     fn is_direct_io_read_enabled() -> bool =
         rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE);
 }
 
 cached_read_env! {
     /// Check if O_DIRECT shard/part data writes are enabled.
-    #[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
     fn is_direct_io_write_enabled() -> bool =
         rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE);
 }
@@ -1483,7 +1456,6 @@ pub(crate) fn effective_durability(volume: &str) -> DurabilityMode {
 
 cached_read_env! {
     /// Get the O_DIRECT read threshold size.
-    #[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
     fn get_direct_io_read_threshold() -> usize =
         rustfs_utils::get_env_usize(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD);
 }
@@ -1701,20 +1673,12 @@ impl DirectIoWriteState {
 /// Target staging size for O_DIRECT writes, rounded up to the DIO alignment.
 /// Bounds the per-writer aligned bounce buffer and batches many shard blocks
 /// into one positioned write to keep the syscall count low.
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 const DIRECT_WRITE_STAGING_BYTES: usize = 1024 * 1024;
 
 /// Aligned bounce-buffer capacity for a given DIO alignment: the target staging
 /// size rounded up to a whole multiple of `align` so the buffer address, every
 /// flushed batch length, and every write offset stay alignment-correct.
 /// Platform-independent (no O_DIRECT), so it is unit-tested on any host.
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 fn direct_write_staging_capacity(align: usize) -> usize {
     debug_assert!(align.is_power_of_two() && align >= 512);
     DIRECT_WRITE_STAGING_BYTES.div_ceil(align) * align
@@ -1723,10 +1687,6 @@ fn direct_write_staging_capacity(align: usize) -> usize {
 /// Split `filled` staged bytes into the alignment-sized prefix written with
 /// O_DIRECT and the sub-alignment tail written buffered. Platform-independent,
 /// so the tail-boundary math is unit-tested on any host.
-#[allow(
-    dead_code,
-    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
-)]
 fn direct_write_tail_split(filled: usize, align: usize) -> (usize, usize) {
     let aligned = filled - (filled % align);
     (aligned, filled - aligned)
@@ -2182,7 +2142,6 @@ fn set_delete_version_fail_after_data_staged(path: &str) {
 }
 
 #[cfg(test)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(crate) fn set_delete_version_fail_after_commit(root: &Path, path: &str) {
     DELETE_VERSION_FAIL_AFTER_COMMIT
         .lock()
@@ -2488,10 +2447,6 @@ enum SyncMode {
     FileOnly,
 }
 
-#[allow(
-    dead_code,
-    reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
-)]
 struct FileCacheReclaimWriter {
     inner: File,
     reclaim_len: usize,
@@ -2499,10 +2454,6 @@ struct FileCacheReclaimWriter {
     reclaimed: bool,
 }
 
-#[allow(
-    dead_code,
-    reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
-)]
 struct FileCacheReclaimReader {
     inner: File,
     reclaim_offset: u64,
@@ -2568,10 +2519,6 @@ impl<R: AsyncRead + Unpin> AsyncRead for StallTimeoutReader<R> {
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "reclaim metrics emitter reached only from the Linux-gated reclaim paths (backlog#1823)"
-)]
 fn record_file_cache_reclaim_success(kind: &'static str, reclaim_len: usize, started: std::time::Instant) {
     // Runs per read-stream page-cache reclaim window; skip the whole emission
     // (three metric-key constructions) when general metrics are disabled.
@@ -3124,7 +3071,6 @@ impl LocalIoBackend for StdBackend {
             use memmap2::MmapOptions;
             use std::time::{Duration as StdDuration, Instant as StdInstant};
 
-            #[allow(dead_code, reason = "mmap copy result slot kept beside the mapping it owns (backlog#1823)")]
             struct MmapCopyReadResult {
                 bytes: Bytes,
                 access_check_duration: StdDuration,
@@ -4758,10 +4704,6 @@ fn build_local_io_backend(root: PathBuf) -> Arc<dyn LocalIoBackend> {
     Arc::new(StdBackend::new(root))
 }
 
-#[allow(
-    dead_code,
-    reason = "path cache and cwd slots retained beside the disk root they derive from (backlog#1823)"
-)]
 pub struct LocalDisk {
     pub root: PathBuf,
     publication_root: os::PublicationRoot,
@@ -5548,7 +5490,6 @@ impl LocalDisk {
         Ok(Self::resolve_abs_path_from(&self.root, path.as_ref()))
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn io_resolve_abs_path(&self, path: impl AsRef<Path>) -> PathBuf {
         let path_ref = path.as_ref();
         let path_str = path_ref.to_string_lossy();
@@ -5626,24 +5567,15 @@ impl LocalDisk {
     }
 
     // Check if a path is valid
-    #[allow(
-        dead_code,
-        reason = "method wrapper over the live free function check_local_disk_valid_path; no caller in this port (backlog#1823)"
-    )]
     fn check_valid_path<P: AsRef<Path>>(&self, path: P) -> Result<()> {
         check_local_disk_valid_path(self.io_root(), path)
     }
 
-    #[allow(
-        dead_code,
-        reason = "method wrapper over the live free function reject_local_disk_symlink_components; no caller in this port (backlog#1823)"
-    )]
     fn reject_symlink_components(&self, path: &Path) -> Result<()> {
         reject_local_disk_symlink_components(self.io_root(), path)
     }
 
     // Batch path generation with single lock acquisition
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn get_object_paths_batch(&self, requests: &[(String, String)]) -> Result<Vec<PathBuf>> {
         let mut results = Vec::with_capacity(requests.len());
         let mut cache_misses = Vec::new();
@@ -6556,7 +6488,6 @@ impl LocalDisk {
         Ok(f)
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     async fn open_file_read_only(&self, path: impl AsRef<Path>) -> Result<File> {
         let f = super::fs::open_file(path.as_ref(), O_RDONLY).await.map_err(to_file_error)?;
         Ok(f)

crates/ecstore/src/disk/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 // #730: disk abstractions still carry staged health and direct-I/O migration paths.
+#![allow(dead_code)]
 
 pub mod disk_store;
 pub mod endpoint;
@@ -1113,10 +1114,6 @@ pub struct DiskInfo {
 }
 
 #[derive(Clone, Debug, Default)]
-#[allow(
-    dead_code,
-    reason = "MinIO-parity disk info shape with no constructor in this port (backlog#1823)"
-)]
 pub struct Info {
     pub total: u64,
     pub free: u64,
@@ -1375,7 +1372,6 @@ pub fn conv_part_err_to_int(err: &Option<Error>) -> usize {
     }
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn has_part_err(part_errs: &[usize]) -> bool {
     part_errs.iter().any(|err| *err != CHECK_PART_SUCCESS)
 }

crates/ecstore/src/disk/os.rs

@@ -571,10 +571,6 @@ fn regular_files(dir: &Path) -> io::Result<Vec<PathBuf>> {
 
 /// Fdatasync every regular file directly inside `dir`, then fsync the directory
 /// itself.
-#[allow(
-    dead_code,
-    reason = "reached only through sync_dir_files, whose callers are tests (backlog#1823)"
-)]
 pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
     for entry in std::fs::read_dir(dir.as_ref())? {
         let entry = entry?;
@@ -587,7 +583,6 @@ pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
 
 /// Async wrapper around [`sync_dir_files_std`]. Large directories flush files
 /// concurrently, bounded both per directory and process-wide.
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub async fn sync_dir_files(dir: impl AsRef<Path>) -> io::Result<()> {
     sync_dir_files_with_limiter(dir, Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS))).await
 }
@@ -1814,6 +1809,10 @@ impl RenameCommitGuard {
         })
     }
 
+    pub(crate) fn lock_destination_directory_for_path_access(&self, directory: &Path) -> io::Result<RenameDestinationPathGuard> {
+        self.destination_directory_guard(directory, false)
+    }
+
     pub(crate) fn create_destination_directory_for_path_access(
         &self,
         directory: &Path,
@@ -2859,6 +2858,13 @@ pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>
     Ok(())
 }
 
+/// Check if a file exists.
+/// Returns true if the file exists, false otherwise.
+#[tracing::instrument(level = "debug", skip_all)]
+pub fn file_exists(path: impl AsRef<Path>) -> bool {
+    std::fs::metadata(path.as_ref()).map(|_| true).unwrap_or(false)
+}
+
 /// Whether an [`io::Error`] means "the directory is not empty".
 ///
 /// POSIX lets `rmdir`/`rename` report a non-empty directory as either

crates/ecstore/src/io_support/bitrot.rs

@@ -704,7 +704,6 @@ pub(crate) async fn create_bitrot_reader_from_bytes_with_stage_metrics(
 }
 
 #[allow(clippy::too_many_arguments)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn create_deferred_bitrot_reader(
     inline_data: Option<Bytes>,
     disk: Option<DiskStore>,

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/ecstore/src/set_disk/core/io_primitives.rs

@@ -180,13 +180,11 @@ pub(in crate::set_disk) enum GetCodecStreamingReaderBuildOutcome {
     Fallback(GetCodecStreamingFallbackReason),
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(in crate::set_disk) struct MultipartCodecStreamingReader {
     pub(in crate::set_disk) readers: VecDeque<Box<dyn AsyncRead + Unpin + Send + Sync>>,
 }
 
 impl MultipartCodecStreamingReader {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) fn new(readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>>) -> Self {
         Self {
             readers: VecDeque::from(readers),
@@ -668,60 +666,6 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
     parts_metadata: &[FileInfo],
     disks: &[Option<DiskStore>],
 ) -> Option<&'static str> {
-    if let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(candidate) {
-        return Some(reason);
-    }
-
-    let Ok(erasure) = coding::Erasure::try_new_with_options(
-        candidate.erasure.data_blocks,
-        candidate.erasure.parity_blocks,
-        candidate.erasure.block_size,
-        candidate.uses_legacy_checksum,
-    ) else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
-    };
-    let data_files =
-        match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
-            disks.get(index).is_some_and(Option::is_some)
-        }) {
-            Ok(data_files) => data_files,
-            Err(reason) => return Some(reason),
-        };
-
-    let Some(part) = candidate.parts.first() else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
-    };
-    let Ok(object_size) = usize::try_from(candidate.size) else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
-    };
-    let checksum_info = candidate.erasure.get_checksum_info(part.number);
-    let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
-        HashAlgorithm::HighwayHash256SLegacy
-    } else {
-        checksum_info.algorithm
-    };
-    let read_length = inline_erasure_shard_file_offset(
-        0,
-        object_size,
-        object_size,
-        candidate.erasure.block_size,
-        erasure.data_shards,
-        candidate.uses_legacy_checksum,
-    );
-    let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
-    let Ok(mut readers) =
-        build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
-    else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
-    };
-
-    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
-        Some(body) if body.len() == object_size => None,
-        _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
-    }
-}
-
-fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Option<&'static str> {
     // `inline_data` excludes remote objects; this diagnostic reports them separately.
     if !rustfs_utils::http::contains_key_str(&candidate.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA) {
         return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE);
@@ -759,7 +703,51 @@ fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Op
     if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
         return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
     }
-    None
+
+    let Ok(erasure) = coding::Erasure::try_new_with_options(
+        candidate.erasure.data_blocks,
+        candidate.erasure.parity_blocks,
+        candidate.erasure.block_size,
+        candidate.uses_legacy_checksum,
+    ) else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
+    };
+    let data_files =
+        match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
+            disks.get(index).is_some_and(Option::is_some)
+        }) {
+            Ok(data_files) => data_files,
+            Err(reason) => return Some(reason),
+        };
+
+    let Some(part) = candidate.parts.first() else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
+    };
+    let checksum_info = candidate.erasure.get_checksum_info(part.number);
+    let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
+        HashAlgorithm::HighwayHash256SLegacy
+    } else {
+        checksum_info.algorithm
+    };
+    let read_length = inline_erasure_shard_file_offset(
+        0,
+        object_size,
+        object_size,
+        candidate.erasure.block_size,
+        erasure.data_shards,
+        candidate.uses_legacy_checksum,
+    );
+    let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
+    let Ok(mut readers) =
+        build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
+    else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
+    };
+
+    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
+        Some(body) if body.len() == object_size => None,
+        _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
+    }
 }
 
 fn data_read_inline_missing_shards_are_pending(
@@ -1848,7 +1836,6 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_all_shards(
 }
 
 #[allow(clippy::too_many_arguments)]
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum(
     files: &[FileInfo],
     disks: &[Option<DiskStore>],
@@ -2139,7 +2126,6 @@ pub(in crate::set_disk) async fn create_data_block_bitrot_readers(
     setup
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(in crate::set_disk) async fn collect_read_multiple_results<F>(
     tasks: Vec<F>,
     read_quorum: usize,
@@ -2456,7 +2442,6 @@ impl SetDisks {
         let bucket: Arc<str> = Arc::from(bucket);
         let object: Arc<str> = Arc::from(object);
         let version_id: Arc<str> = Arc::from(version_id);
-        let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
         let futures = disks.iter().enumerate().map(|(disk_index, disk)| {
             let disk = disk.clone();
             let task_opts = opts;
@@ -2464,14 +2449,10 @@ impl SetDisks {
             let bucket = bucket.clone();
             let object = object.clone();
             let version_id = version_id.clone();
-            let slowtail_fault = slowtail_fault.clone();
             tokio::spawn(async move {
                 let response_start = observe.then(Instant::now);
                 let result = if let Some(disk) = disk {
                     Self::record_read_version_call(&object, disk_index);
-                    if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(disk_index)) {
-                        tokio::time::sleep(delay).await;
-                    }
                     disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
                         .await
                 } else {
@@ -2567,7 +2548,6 @@ impl SetDisks {
         let mut scheduled_count = 0usize;
         let mut force_full_wait = false;
         let mut final_miss_reason_override = None;
-        let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
         let spawn_read_version =
             |join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
                 let task_opts = opts;
@@ -2575,7 +2555,6 @@ impl SetDisks {
                 let bucket = bucket.clone();
                 let object = object.clone();
                 let version_id = version_id.clone();
-                let slowtail_fault = slowtail_fault.clone();
                 join_set.spawn(async move {
                     let response_start = Instant::now();
                     let result = if let Some(disk) = disk {
@@ -2584,9 +2563,6 @@ impl SetDisks {
                         Self::record_read_version_call(&object, index);
                         #[cfg(test)]
                         Self::read_version_fanout_barrier(&object, index).await;
-                        if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(index)) {
-                            tokio::time::sleep(delay).await;
-                        }
                         disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
                             .await
                     } else {
@@ -2620,14 +2596,6 @@ impl SetDisks {
                     Ok(file_info) => {
                         observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed));
                         accumulator.observe_file_info(&file_info);
-                        if bounded_fanout
-                            && read_data
-                            && !force_full_wait
-                            && let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info)
-                        {
-                            force_full_wait = true;
-                            final_miss_reason_override.get_or_insert(reason);
-                        }
                         if let Some(slot) = ress.get_mut(index) {
                             *slot = file_info;
                         }
@@ -2987,7 +2955,6 @@ impl SetDisks {
         (meta_file_infos, errs)
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) async fn read_multiple_files(
         disks: &[Option<DiskStore>],
         req: ReadMultipleReq,
@@ -3021,11 +2988,14 @@ impl SetDisks {
             });
         }
 
-        let (ress, _errors) = match collect_read_multiple_results(futures, read_quorum).await {
+        let (ress, errors) = match collect_read_multiple_results(futures, read_quorum).await {
             Ok(collected) => collected,
             Err(()) => return empty_quorum_result(),
         };
 
+        // debug!("ReadMultipleResp ress {:?}", ress);
+        // debug!("ReadMultipleResp errors {:?}", errors);
+
         let mut ret = Vec::with_capacity(req.files.len());
 
         for want in req.files.iter() {
@@ -3164,7 +3134,6 @@ pub(in crate::set_disk) struct RenameDataCommit {
     pub(in crate::set_disk) committed_file_info: FileInfo,
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 type RenameDataLegacyTuple = (
     Vec<Option<DiskStore>>,
     RenameConvergence,
@@ -3174,7 +3143,6 @@ type RenameDataLegacyTuple = (
 );
 
 impl RenameDataCommit {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn into_legacy_tuple(self) -> RenameDataLegacyTuple {
         (
             self.online_disks,
@@ -3293,7 +3261,6 @@ impl SetDisks {
 
     #[tracing::instrument(level = "debug", skip(disks, file_infos))]
     #[allow(clippy::type_complexity)]
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) async fn rename_data(
         disks: &[Option<DiskStore>],
         src_bucket: &str,
@@ -5106,7 +5073,6 @@ fn is_cleanup_not_found(e: &DiskError) -> bool {
 /// normalized to `DiskNotFound`: a panic is not a "disk absent" condition and
 /// must not be silently swallowed as an ignorable error (fixes the historical
 /// `Unexpected`/`DiskNotFound` misclassification).
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn map_cleanup_join_result(joined: std::result::Result<Option<DiskError>, tokio::task::JoinError>) -> Option<DiskError> {
     match joined {
         Ok(res) => res,
@@ -5331,7 +5297,6 @@ pub(in crate::set_disk) mod rename_fanout_barrier_phase {
     /// The per-disk old-data-dir cleanup phase of the commit fan-out.
     pub const CLEANUP: &str = "cleanup";
     /// The per-disk `read_version` phase of metadata read fan-out.
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub const READ_VERSION: &str = "read_version";
 }
 
@@ -5769,130 +5734,6 @@ mod tests {
         (dirs, disks)
     }
 
-    #[test]
-    fn metadata_slowtail_fault_delay_parses_and_filters_request() {
-        temp_env::with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,3")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some("bench-bucket")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
-            ],
-            || {
-                assert_eq!(
-                    get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, true),
-                    Some(Duration::from_millis(25))
-                );
-                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 2, true).is_none());
-                assert!(get_metadata_slowtail_fault_delay("other-bucket", "objects/000001", 3, true).is_none());
-                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "other/000001", 3, true).is_none());
-                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, false).is_none());
-            },
-        );
-    }
-
-    #[test]
-    fn metadata_slowtail_fault_delay_disables_invalid_disk_list() {
-        temp_env::with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,nope")),
-            ],
-            || {
-                assert!(get_metadata_slowtail_fault_delay("bucket", "object", 1, true).is_none());
-            },
-        );
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn metadata_slowtail_fault_delays_only_data_read_metadata_task() {
-        const DISKS: usize = 4;
-        let bucket = "metadata-slowtail-fault-bucket";
-        let object = "objects/metadata-slowtail-fault-object";
-        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
-        install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
-
-        temp_env::async_with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("false")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
-            ],
-            async {
-                let read_without_data =
-                    SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", false, false, false, true, 2);
-                tokio::time::timeout(Duration::from_millis(100), read_without_data)
-                    .await
-                    .expect("non-data metadata fanout must not be delayed by the data-read slowtail hook")
-                    .expect("metadata fanout without read_data should resolve");
-
-                let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
-                    &disks, bucket, bucket, object, "", true, false, false, true, 2,
-                ));
-                assert!(
-                    tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
-                        .await
-                        .is_err(),
-                    "data-read metadata fanout must wait for the injected slow read_version response"
-                );
-                let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
-                    .await
-                    .expect("injected slowtail should eventually complete")
-                    .expect("data-read metadata fanout should resolve");
-                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
-                assert!(errs.iter().all(Option::is_none));
-                assert_eq!(diagnostics.total_responses(), DISKS);
-            },
-        )
-        .await;
-
-        drop(dirs);
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn metadata_slowtail_fault_delays_early_stop_metadata_task() {
-        const DISKS: usize = 4;
-        let bucket = "metadata-slowtail-early-stop-bucket";
-        let object = "objects/metadata-slowtail-early-stop-object";
-        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
-        install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
-
-        temp_env::async_with_vars(
-            [
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
-                (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")),
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
-                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
-            ],
-            async {
-                let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
-                    &disks, bucket, bucket, object, "", true, false, false, true, 2,
-                ));
-                assert!(
-                    tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
-                        .await
-                        .is_err(),
-                    "early-stop metadata fanout must still wait for the injected slow response after fallback to full wait"
-                );
-                let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
-                    .await
-                    .expect("injected early-stop slowtail should eventually complete")
-                    .expect("early-stop metadata fanout should resolve");
-                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
-                assert!(errs.iter().all(Option::is_none));
-                assert_eq!(diagnostics.total_responses(), DISKS);
-            },
-        )
-        .await;
-
-        drop(dirs);
-    }
-
     /// Demo / regression guard for the backlog#1325 per-disk call counters.
     ///
     /// The metadata fan-out issues each `read_version` inside its own
@@ -7240,7 +7081,7 @@ mod tests {
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn bounded_non_inline_data_get_immediately_forces_full_fanout() {
+    async fn bounded_non_inline_data_get_hedges_then_waits_for_full_fanout() {
         const DISKS: usize = 4;
         let bucket = "bounded-data-get-hedge-bucket";
         let object = "bounded-data-get-hedge-object";
@@ -7271,7 +7112,7 @@ mod tests {
                     }
                 })
                 .await
-                .expect("bounded non-inline data-read fanout should immediately schedule the spare disk");
+                .expect("bounded data-read fanout should hedge by starting the spare disk");
 
                 let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
                 assert!(
@@ -7287,7 +7128,7 @@ mod tests {
                 assert_eq!(
                     calls.total(disk_call_counters::KIND_READ_VERSION),
                     DISKS as u64,
-                    "bounded non-inline data-read fanout should issue the paused disk plus the remaining spare"
+                    "bounded data-read fanout should issue the paused disk plus one spare hedge"
                 );
                 assert_eq!(diagnostics.total_responses(), DISKS);
                 assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
@@ -7314,42 +7155,16 @@ mod tests {
                 ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
             ],
             async {
-                let barrier = rename_fanout_barrier::arm(object, 2, rename_fanout_barrier::PHASE_READ_VERSION);
                 let calls = disk_call_counters::observe(object);
-                let disks_for_read = disks.clone();
-                let mut read = tokio::spawn(async move {
-                    SetDisks::read_all_fileinfo_observed(&disks_for_read, bucket, bucket, object, "", true, false, false, true, 2)
+                let (parts_metadata, errs, diagnostics) =
+                    SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2)
                         .await
-                });
-
-                tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
-                    .await
-                    .expect("default bounded non-inline read should schedule the paused metadata task");
-                tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
-                    while calls.for_disk(disk_call_counters::KIND_READ_VERSION, 3) == 0 {
-                        tokio::task::yield_now().await;
-                    }
-                })
-                .await
-                .expect(
-                    "default bounded non-inline read should immediately force full fanout after the first non-inline response",
-                );
-
-                let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
-                assert!(
-                    pending.is_err(),
-                    "default non-inline data reads must not return before the paused metadata response"
-                );
-                barrier.release();
-                let (parts_metadata, errs, diagnostics) = read
-                    .await
-                    .expect("metadata read task should not panic")
-                    .expect("default data-read metadata should resolve");
+                        .expect("default data-read metadata should resolve");
 
                 assert_eq!(
                     calls.total(disk_call_counters::KIND_READ_VERSION),
                     DISKS as u64,
-                    "default non-inline GET data-read metadata must keep full fanout without waiting for a quorum miss first"
+                    "default non-inline GET data-read metadata must keep full fanout for read-failure tolerance"
                 );
                 assert_eq!(diagnostics.total_responses(), DISKS);
                 assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);

crates/ecstore/src/set_disk/ctx.rs

@@ -42,20 +42,12 @@ impl<'a> SetDisksCtx<'a> {
     }
 
     /// The borrowed core, for state not yet fronted by a typed accessor.
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn core(&self) -> &'a SetDisks {
         self.core
     }
 
     // --- Immutable topology / config (fixed after construction) ---
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn set_index(&self) -> usize {
         self.core.set_index
     }
@@ -64,26 +56,14 @@ impl<'a> SetDisksCtx<'a> {
         self.core.pool_index
     }
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn set_drive_count(&self) -> usize {
         self.core.set_drive_count
     }
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn default_parity_count(&self) -> usize {
         self.core.default_parity_count
     }
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn set_endpoints(&self) -> &'a [Endpoint] {
         &self.core.set_endpoints
     }
@@ -92,10 +72,6 @@ impl<'a> SetDisksCtx<'a> {
         &self.core.format
     }
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn locker_owner(&self) -> &'a str {
         &self.core.locker_owner
     }
@@ -108,10 +84,6 @@ impl<'a> SetDisksCtx<'a> {
 
     // --- Locker trio ---
 
-    #[allow(
-        dead_code,
-        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
-    )]
     pub(crate) fn lockers(&self) -> &'a [Arc<dyn LockClient>] {
         &self.core.lockers
     }

crates/ecstore/src/set_disk/mod.rs

@@ -39,6 +39,7 @@
 //! - `metadata.rs`, `replication.rs`, `shard_source.rs` — supporting helpers.
 
 // #730: SetDisks still hosts staged read/heal/write migration helpers.
+#![allow(dead_code)]
 #![allow(unused_imports)]
 #![allow(unused_variables)]
 
@@ -102,7 +103,9 @@ use crate::storage_api_contracts::{
 };
 use crate::store::utils::is_reserved_or_invalid_bucket;
 use crate::{
-    bucket::lifecycle::bucket_lifecycle_ops::{LifecycleOps, get_transitioned_object_reader_with_tier_manager, put_restore_opts},
+    bucket::lifecycle::bucket_lifecycle_ops::{
+        LifecycleOps, gen_transition_objname, get_transitioned_object_reader_with_tier_manager, put_restore_opts,
+    },
     cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
     config::storageclass,
     disk::{
@@ -174,14 +177,15 @@ use std::future::Future;
 use std::hash::{BuildHasher, Hash, Hasher};
 use std::mem::{self};
 use std::pin::Pin;
+use std::sync::OnceLock;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::{Arc, OnceLock};
 use std::task::{Context, Poll};
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 use std::{
     collections::{HashMap, HashSet},
     io::{Cursor, Write},
     path::Path,
+    sync::Arc,
     time::Duration,
 };
 use time::OffsetDateTime;
@@ -620,9 +624,7 @@ fn adaptive_duplex_buffer_size(object_size: i64) -> usize {
 // Each flag has a corresponding `*_ROLLOUT_PCT` for percentage-based gradual rollout.
 // ============================================================================
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1);
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750);
 const GET_OBJECT_METADATA_CACHE_TTL: Duration = Duration::from_secs(2); // Increased from 250ms to 2s
 const DEFAULT_GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 4096; // Increased from 1024 to 4096
@@ -696,15 +698,7 @@ const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EAR
 // the env var to `false` to fall back to full-wait metadata fanout.
 const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = true;
 
-#[allow(
-    dead_code,
-    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
-)]
 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT";
-#[allow(
-    dead_code,
-    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
-)]
 const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100;
 
 const ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE";
@@ -714,12 +708,7 @@ const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_ME
 const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
 
 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT";
-const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = true;
-
-const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";
-const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS";
-const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET";
-const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX";
+const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = false;
 
 // --- Multipart Reader-Setup Prefetch Configuration (backlog#870) ---
 
@@ -921,10 +910,6 @@ mod prepared_get_object_metadata_tests {
             .expect("test should find an object whose initial fanout covers both data shards")
     }
 
-    #[allow(
-        dead_code,
-        reason = "test fixture no assertion in this module uses today; the live namesake lives in io_primitives tests (backlog#1823)"
-    )]
     fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
         *bounded_metadata_fanout_order(bucket, object, 4, 2)
             .get(3)
@@ -1125,10 +1110,7 @@ mod prepared_get_object_metadata_tests {
 
         assert_eq!(object_size, payload.len() as i64);
         assert_eq!(restored, payload);
-        assert_eq!(
-            calls_total, 4,
-            "default production inline GET should schedule the initial bounded quorum plus one hedge"
-        );
+        assert_eq!(calls_total, 4, "default production GET should eagerly schedule the full metadata fanout");
         assert_eq!(
             recorder.histogram_values(
                 "rustfs_io_get_object_metadata_fanout_scheduled",
@@ -1702,95 +1684,6 @@ fn is_get_metadata_early_stop_bounded_fanout_enabled() -> bool {
     }
 }
 
-#[derive(Debug)]
-struct GetMetadataSlowtailFaultConfig {
-    delay: Duration,
-    disks: Arc<[usize]>,
-    bucket: Option<String>,
-    object_prefix: Option<String>,
-}
-
-#[derive(Clone, Debug)]
-struct GetMetadataSlowtailFaultRequest {
-    delay: Duration,
-    disks: Arc<[usize]>,
-}
-
-impl GetMetadataSlowtailFaultRequest {
-    fn delay_for_disk(&self, disk_index: usize) -> Option<Duration> {
-        self.disks.contains(&disk_index).then_some(self.delay)
-    }
-}
-
-fn parse_get_metadata_slowtail_fault_disks(raw: &str) -> Option<Vec<usize>> {
-    let mut disks = Vec::new();
-    for item in raw.split(',').map(str::trim).filter(|item| !item.is_empty()) {
-        let Ok(index) = item.parse::<usize>() else {
-            return None;
-        };
-        if !disks.contains(&index) {
-            disks.push(index);
-        }
-    }
-    (!disks.is_empty()).then_some(disks)
-}
-
-fn load_get_metadata_slowtail_fault_config() -> Option<GetMetadataSlowtailFaultConfig> {
-    let delay_ms = rustfs_utils::get_env_u64(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, 0);
-    if delay_ms == 0 {
-        return None;
-    }
-    let disks = parse_get_metadata_slowtail_fault_disks(&std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS).ok()?)?;
-    let bucket = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET)
-        .ok()
-        .filter(|value| !value.is_empty());
-    let object_prefix = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX)
-        .ok()
-        .filter(|value| !value.is_empty());
-    Some(GetMetadataSlowtailFaultConfig {
-        delay: Duration::from_millis(delay_ms),
-        disks: Arc::from(disks.into_boxed_slice()),
-        bucket,
-        object_prefix,
-    })
-}
-
-fn get_metadata_slowtail_fault_request(bucket: &str, object: &str, read_data: bool) -> Option<GetMetadataSlowtailFaultRequest> {
-    if !read_data {
-        return None;
-    }
-
-    #[cfg(test)]
-    let config = load_get_metadata_slowtail_fault_config();
-    #[cfg(test)]
-    let config = config.as_ref()?;
-    #[cfg(not(test))]
-    let config = ({
-        static CACHED: OnceLock<Option<GetMetadataSlowtailFaultConfig>> = OnceLock::new();
-        CACHED.get_or_init(load_get_metadata_slowtail_fault_config).as_ref()
-    })?;
-
-    if let Some(expected_bucket) = &config.bucket
-        && expected_bucket != bucket
-    {
-        return None;
-    }
-    if let Some(expected_prefix) = &config.object_prefix
-        && !object.starts_with(expected_prefix)
-    {
-        return None;
-    }
-    Some(GetMetadataSlowtailFaultRequest {
-        delay: config.delay,
-        disks: config.disks.clone(),
-    })
-}
-
-#[cfg(test)]
-fn get_metadata_slowtail_fault_delay(bucket: &str, object: &str, disk_index: usize, read_data: bool) -> Option<Duration> {
-    get_metadata_slowtail_fault_request(bucket, object, read_data)?.delay_for_disk(disk_index)
-}
-
 /// Check if multipart reads prefetch the next part's bitrot reader setup
 /// while the current part decodes (backlog#870).
 ///
@@ -1816,10 +1709,6 @@ fn is_multipart_reader_setup_prefetch_enabled() -> bool {
     }
 }
 
-#[allow(
-    dead_code,
-    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
-)]
 fn get_metadata_early_stop_rollout_pct() -> u32 {
     static CACHED: OnceLock<u32> = OnceLock::new();
     *CACHED.get_or_init(|| {
@@ -1859,10 +1748,6 @@ fn should_use_codec_streaming(config: GetCodecStreamingConfig, bucket: &str, obj
 }
 
 /// Should this specific request use metadata early-stop?
-#[allow(
-    dead_code,
-    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
-)]
 pub fn should_use_metadata_early_stop(bucket: &str, object: &str) -> bool {
     let base = is_get_metadata_early_stop_enabled();
     let pct = get_metadata_early_stop_rollout_pct();
@@ -2296,7 +2181,6 @@ fn classify_get_codec_streaming_object_class(
     GetCodecStreamingObjectClass::PlainSinglePart
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn is_get_small_object_direct_memory_eligible_with_threshold(
     range: &Option<HTTPRangeSpec>,
     object_info: &ObjectInfo,
@@ -2902,7 +2786,6 @@ pub struct SetDisks {
     /// Stable namespace shared by every object lock created for this set.
     set_lock_namespace: Arc<str>,
     pub format: FormatV3,
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     disk_health_cache: Arc<RwLock<Vec<Option<DiskHealthEntry>>>>,
     get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
     get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState,
@@ -3178,13 +3061,11 @@ struct GetObjectMetadataCacheEntry {
 
 #[derive(Clone, Debug)]
 struct DiskHealthEntry {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     last_check: Instant,
     online: bool,
 }
 
 impl DiskHealthEntry {
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn cached_value(&self) -> Option<bool> {
         if self.last_check.elapsed() <= DISK_HEALTH_CACHE_TTL {
             Some(self.online)
@@ -3778,7 +3659,6 @@ fn multipart_put_large_batch_min_size_bytes() -> usize {
     })
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath {
     if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) {
         SmallWritePath::Inline
@@ -4357,7 +4237,6 @@ fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &Obj
 ///
 /// Fail closed: when bucket metadata cannot be resolved the check stays on, so
 /// object-lock protection is never skipped because of a metadata lookup miss.
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(crate) fn object_lock_delete_check_required(bucket_meta: Option<&crate::bucket::metadata::BucketMetadata>) -> bool {
     bucket_meta.is_none_or(|meta| meta.object_locking())
 }
@@ -4633,6 +4512,15 @@ impl Hash for ObjProps {
     }
 }
 
+#[derive(Default, Clone, Debug)]
+pub struct HealEntryResult {
+    pub bytes: usize,
+    pub success: bool,
+    pub skipped: bool,
+    pub entry_done: bool,
+    pub name: String,
+}
+
 fn is_object_dangling(
     meta_arr: &[FileInfo],
     errs: &[Option<DiskError>],
@@ -5409,7 +5297,6 @@ pub fn is_valid_storage_class(storage_class: &str) -> bool {
 }
 
 /// Returns true if the storage class is a cold storage tier that requires special handling
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn is_cold_storage_class(storage_class: &str) -> bool {
     matches!(
         storage_class,
@@ -5418,7 +5305,6 @@ pub fn is_cold_storage_class(storage_class: &str) -> bool {
 }
 
 /// Returns true if the storage class is an infrequent access tier
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn is_infrequent_access_class(storage_class: &str) -> bool {
     matches!(
         storage_class,

crates/ecstore/src/set_disk/ops/heal.rs

@@ -453,9 +453,7 @@ impl SetDisks {
             ..Default::default()
         };
 
-        // Bound, not `_`: this guard must live to the end of the scope. A bare
-        // `_` would drop it here and release the namespace write lock.
-        let _write_lock_guard = if !opts.no_lock {
+        let write_lock_guard = if !opts.no_lock {
             let ns_lock = self.new_ns_lock(bucket, object).await?;
             Some(
                 ns_lock
@@ -998,7 +996,7 @@ impl SetDisks {
                                                 readers.push(None);
                                                 continue;
                                             }
-                                            Err(_e) => {
+                                            Err(e) => {
                                                 readers.push(None);
                                                 continue;
                                             }
@@ -1547,9 +1545,6 @@ impl SetDisks {
 
         for candidate in candidates.iter_mut().filter(|candidate| candidate.local_payload) {
             for (disk_index, disk) in disks.iter().enumerate() {
-                // Only the #[cfg(test)] fault-injection branch below reads this.
-                #[cfg(not(test))]
-                let _ = disk_index;
                 let Some(disk) = disk else {
                     return Ok(DanglingDeleteSafety::UnsafeToDelete);
                 };
@@ -1721,10 +1716,6 @@ impl SetDisks {
         Ok((result, None))
     }
 
-    #[allow(
-        dead_code,
-        reason = "lock-taking wrapper over the live heal_object_dir_locked; only comments reference it (backlog#1823)"
-    )]
     #[tracing::instrument(level = "trace", skip(self), fields(bucket = %bucket, object = %object))]
     pub(in crate::set_disk) async fn heal_object_dir(
         &self,

crates/ecstore/src/set_disk/ops/locking.rs

@@ -66,7 +66,6 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks {
 }
 
 impl SetDisks {
-    #[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
     pub(in crate::set_disk) fn format_lock_error(&self, bucket: &str, object: &str, mode: &str, err: &LockResult) -> String {
         match err {
             LockResult::Timeout => {
@@ -80,7 +79,6 @@ impl SetDisks {
         }
     }
 
-    #[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
     pub(in crate::set_disk) fn format_lock_error_from_error(
         &self,
         bucket: &str,
@@ -145,7 +143,6 @@ impl SetDisks {
         disks
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) async fn get_online_disks(&self) -> Vec<Option<DiskStore>> {
         let snapshot = self.drive_membership_snapshot().await;
         let mut disks = snapshot.strict_online_candidates().into_iter().map(Some).collect::<Vec<_>>();
@@ -156,10 +153,6 @@ impl SetDisks {
         disks
     }
 
-    #[allow(
-        dead_code,
-        reason = "local-only sibling of the test-covered get_online_disks; no caller in this port (backlog#1823)"
-    )]
     pub(in crate::set_disk) async fn get_online_local_disks(&self) -> Vec<Option<DiskStore>> {
         let snapshot = self.drive_membership_snapshot().await;
         let mut disks = snapshot
@@ -439,10 +432,6 @@ impl SetDisks {
         Ok((disk, fm))
     }
 
-    #[allow(
-        dead_code,
-        reason = "MinIO-parity healing-disk accessor with no caller in this port (backlog#1823)"
-    )]
     pub(in crate::set_disk) async fn get_online_disk_with_healing(
         &self,
         incl_healing: bool,
@@ -451,10 +440,6 @@ impl SetDisks {
         Ok((new_disks, healing > 0))
     }
 
-    #[allow(
-        dead_code,
-        reason = "reached only from get_online_disk_with_healing, itself uncalled in this port (backlog#1823)"
-    )]
     pub(in crate::set_disk) async fn get_online_disk_with_healing_and_info(
         &self,
         incl_healing: bool,

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -415,7 +415,6 @@ fn reduce_quorum_part_numbers(object_parts: Vec<Vec<String>>, read_quorum: usize
 /// never returned, but flips `is_truncated` to `true` and yields a
 /// `next_upload_id_marker` pointing at the last returned upload so the caller can
 /// resume paging.
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn paginate_upload_page(remaining: &[MultipartInfo], max_uploads: usize) -> (Vec<MultipartInfo>, bool, Option<String>) {
     let is_truncated = remaining.len() > max_uploads;
     let page: Vec<MultipartInfo> = remaining.iter().take(max_uploads).cloned().collect();
@@ -558,7 +557,6 @@ impl SetDisks {
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(super) async fn check_upload_id_exists(
         &self,
         bucket: &str,
@@ -1400,7 +1398,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         let mut count = max_parts;
 
-        for part in object_parts.iter() {
+        for (i, part) in object_parts.iter().enumerate() {
             if let Some(err) = &part.error {
                 warn!("list_object_parts part error: {:?}", &err);
             }
@@ -2043,8 +2041,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     && let Err(err) = checksum.add_part(&cs, ext_part.actual_size)
                 {
                     error!(
-                        "complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}, err={}",
-                        p.part_num, bucket, object, err
+                        "complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}",
+                        p.part_num, bucket, object
                     );
                     return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
                 }
@@ -2089,8 +2087,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 }
             } else if let Err(err) = wtcs.matches(&checksum_combined, uploaded_parts.len() as i32) {
                 error!(
-                    "complete_multipart_upload checksum matches failed want={}, got={}, err={}",
-                    wtcs.encoded, checksum.encoded, err
+                    "complete_multipart_upload checksum matches failed want={}, got={}",
+                    wtcs.encoded, checksum.encoded
                 );
                 return Err(Error::other(format!(
                     "complete_multipart_upload checksum matches failed want={}, got={}",

crates/ecstore/src/set_disk/ops/object.rs

@@ -3507,10 +3507,6 @@ struct TransitionUploadedSaveProbeState {
 }
 
 #[cfg(test)]
-#[allow(
-    dead_code,
-    reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-)]
 struct TransitionUploadedSaveProbe {
     state: Arc<TransitionUploadedSaveProbeState>,
 }
@@ -3521,10 +3517,6 @@ static TRANSITION_UPLOADED_SAVE_PROBE: std::sync::OnceLock<std::sync::Mutex<Opti
 
 #[cfg(test)]
 impl TransitionUploadedSaveProbe {
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn install(bucket: &str, object: &str) -> Self {
         let state = Arc::new(TransitionUploadedSaveProbeState {
             bucket: bucket.to_string(),
@@ -3541,10 +3533,6 @@ impl TransitionUploadedSaveProbe {
         Self { state }
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn attempts(&self) -> usize {
         self.state.attempts.load(std::sync::atomic::Ordering::Acquire)
     }
@@ -3750,10 +3738,6 @@ struct TransitionCommitBarrierState {
 }
 
 #[cfg(test)]
-#[allow(
-    dead_code,
-    reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-)]
 struct TransitionCommitBarrier {
     state: Arc<TransitionCommitBarrierState>,
 }
@@ -3764,26 +3748,14 @@ static TRANSITION_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Ar
 
 #[cfg(test)]
 impl TransitionCommitBarrier {
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn install_before_lock_lost_check(bucket: &str, object: &str) -> Self {
         Self::install_at(bucket, object, TransitionCommitPause::BeforeLockLost)
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn install(bucket: &str, object: &str) -> Self {
         Self::install_at(bucket, object, TransitionCommitPause::BeforeLeaseValidation)
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn install_after_lease_check(bucket: &str, object: &str) -> Self {
         Self::install_at(bucket, object, TransitionCommitPause::AfterLeaseValidation)
     }
@@ -3806,20 +3778,12 @@ impl TransitionCommitBarrier {
         Self { state }
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     async fn wait_until_paused(&self) {
         tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
             .await
             .expect("transition should reach the deterministic commit barrier");
     }
 
-    #[allow(
-        dead_code,
-        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
-    )]
     fn release(&self) {
         self.state.release.notify_one();
     }
@@ -5656,9 +5620,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         // TODO: Lifecycle
 
         let mut version_found = true;
-        // delete_object_version below derives its own majority quorum from the
-        // disk array, so the object-derived quorum here is unused.
-        let (mut goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
+        let (mut goi, write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
         if let Some(err) = &gerr
             && goi.name.is_empty()
         {
@@ -6412,7 +6374,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
 
         for disk in disks.iter() {
-            if disk.is_some() {
+            if let Some(disk) = disk {
                 continue;
             }
             let _ = self
@@ -10208,288 +10170,6 @@ mod transition_upload_integrity_tests {
         assert!(backend.contains(remote_object).await, "committed remote object should remain available");
     }
 
-    /// Compresses `plaintext` with the codec the PUT path uses, so the stored
-    /// bytes round-trip through the read path's decompressor.
-    async fn compress_for_storage(plaintext: &[u8]) -> Vec<u8> {
-        let mut reader = crate::io_support::rio::compression_reader(
-            Cursor::new(plaintext.to_vec()),
-            rustfs_utils::CompressionAlgorithm::default(),
-            false,
-        );
-        let mut compressed = Vec::new();
-        reader.read_to_end(&mut compressed).await.expect("plaintext should compress");
-        assert!(compressed.len() < plaintext.len(), "test payload must actually compress");
-        compressed
-    }
-
-    /// Writes a genuinely compressed object: stored data is `compressed`, and the
-    /// metadata marks it compressed with the plaintext length as its actual size,
-    /// exactly as the app-layer compress path records it.
-    async fn write_compressed_source(
-        set_disks: &Arc<SetDisks>,
-        disk_stores: &[DiskStore],
-        bucket: &str,
-        object: &str,
-        plaintext: &[u8],
-        compressed: &[u8],
-    ) -> ObjectInfo {
-        for disk in disk_stores {
-            disk.make_volume(bucket).await.expect("bucket volume should be created");
-        }
-        let mut user_defined = HashMap::new();
-        rustfs_utils::http::insert_str(
-            &mut user_defined,
-            rustfs_utils::http::SUFFIX_COMPRESSION,
-            crate::io_support::rio::compression_metadata_value(rustfs_utils::CompressionAlgorithm::default()),
-        );
-        rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
-        let stream = crate::io_support::rio::HashReader::from_stream(
-            Cursor::new(compressed.to_vec()),
-            compressed.len() as i64,
-            plaintext.len() as i64,
-            None,
-            None,
-            false,
-        )
-        .expect("hash reader over compressed bytes");
-        let mut reader = PutObjReader::new(stream);
-        set_disks
-            .put_object(
-                bucket,
-                object,
-                &mut reader,
-                &ObjectOptions {
-                    no_lock: true,
-                    user_defined,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("compressed object should be written")
-    }
-
-    async fn read_transitioned(
-        set_disks: &Arc<SetDisks>,
-        bucket: &str,
-        object: &str,
-        range: Option<HTTPRangeSpec>,
-        opts: &ObjectOptions,
-    ) -> (Vec<u8>, i64) {
-        let mut reader = set_disks
-            .get_object_reader(bucket, object, range, HeaderMap::new(), opts)
-            .await
-            .expect("transitioned object reader should open");
-        let published_size = reader.object_info.size;
-        let mut body = Vec::new();
-        reader
-            .stream
-            .read_to_end(&mut body)
-            .await
-            .expect("transitioned body should drain");
-        (body, published_size)
-    }
-
-    /// Transition uploads the object's STORED bytes, so a tiered read has to
-    /// apply the same transform an erasure read would. #6107 routed this path
-    /// through `ReadPlan` to stop serving an encrypted object's ciphertext;
-    /// compression rides the same plan, and nothing pinned it (backlog#1851).
-    /// Without the transform this GET returns the compressed bytes under the
-    /// compressed size — silent corruption for every client of a compressed
-    /// object that ILM has moved to a warm tier.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn transitioned_compressed_object_get_returns_plaintext() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "transitioned-compressed-get-bucket";
-        let object = "object.txt";
-        let plaintext = b"transitioned compressed objects must decompress on read ".repeat(20_000);
-        let compressed = compress_for_storage(&plaintext).await;
-        let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
-
-        let opts = ObjectOptions {
-            no_lock: true,
-            ..Default::default()
-        };
-        let (local_body, local_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
-        assert_eq!(local_body, plaintext, "control: the pre-transition read must decompress");
-        assert_eq!(
-            local_size,
-            plaintext.len() as i64,
-            "control: the pre-transition read publishes the plaintext size"
-        );
-
-        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-        let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
-        set_disks
-            .transition_object(bucket, object, &transition_options(&original, tier_name))
-            .await
-            .expect("transition should commit");
-
-        let put_versions = backend.put_versions().await;
-        assert_eq!(put_versions.len(), 1, "transition should upload one remote candidate");
-        let remote_bytes = backend
-            .bytes(&put_versions[0].0)
-            .await
-            .expect("remote candidate should be stored");
-        assert_eq!(
-            remote_bytes, compressed,
-            "transition uploads the stored representation; the read side is what has to decode it"
-        );
-
-        let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
-        assert_eq!(body, plaintext, "a tiered read must return the object's content, not its stored bytes");
-        assert_eq!(
-            published_size,
-            plaintext.len() as i64,
-            "a tiered read must publish the plaintext size, not the compressed one"
-        );
-    }
-
-    /// A ranged tiered read is expressed in plaintext coordinates, so the plan
-    /// has to translate it into the remote copy's compressed extent and skip
-    /// into the decompressed stream — the same translation the erasure path does.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn transitioned_compressed_object_range_get_returns_plaintext_slice() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "transitioned-compressed-range-bucket";
-        let object = "object.txt";
-        let plaintext = b"ranged reads of transitioned compressed objects must land in plaintext ".repeat(20_000);
-        let compressed = compress_for_storage(&plaintext).await;
-        let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
-
-        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
-        set_disks
-            .transition_object(bucket, object, &transition_options(&original, tier_name))
-            .await
-            .expect("transition should commit");
-
-        let opts = ObjectOptions {
-            no_lock: true,
-            ..Default::default()
-        };
-        // Deliberately past the compressed size, so a range still measured in
-        // stored coordinates could not produce this slice.
-        let start = compressed.len() as i64 + 4096;
-        let end = start + 511;
-        let range = HTTPRangeSpec {
-            is_suffix_length: false,
-            start,
-            end,
-        };
-        let (body, published_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
-
-        let expected = &plaintext[start as usize..=end as usize];
-        assert_eq!(body, expected, "a ranged tiered read must return that plaintext slice");
-        assert_eq!(published_size, expected.len() as i64, "a ranged tiered read publishes the slice length");
-    }
-
-    /// The restore copy-back re-writes the object under its original metadata,
-    /// which still says "compressed". It therefore has to keep receiving the
-    /// STORED bytes: `restore_request_active` holds it on the plan's `Plain`
-    /// branch, and decompressing there would write plaintext under compressed
-    /// metadata.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn restore_read_of_transitioned_compressed_object_keeps_stored_bytes() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "transitioned-compressed-restore-bucket";
-        let object = "object.txt";
-        let plaintext = b"restore copy-back must keep the stored representation intact ".repeat(20_000);
-        let compressed = compress_for_storage(&plaintext).await;
-        let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
-
-        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
-        set_disks
-            .transition_object(bucket, object, &transition_options(&original, tier_name))
-            .await
-            .expect("transition should commit");
-
-        let oi = set_disks
-            .get_object_info(
-                bucket,
-                object,
-                &ObjectOptions {
-                    no_lock: true,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("transitioned metadata should resolve");
-        let restore_opts = ObjectOptions {
-            no_lock: true,
-            part_number: Some(1),
-            transition: TransitionOptions {
-                restore_request: s3s::dto::RestoreRequest {
-                    days: Some(1),
-                    ..Default::default()
-                },
-                ..Default::default()
-            },
-            ..Default::default()
-        };
-        let mut reader = get_transitioned_object_reader_with_tier_manager(
-            bucket,
-            object,
-            &None,
-            &HeaderMap::new(),
-            &oi,
-            &restore_opts,
-            &set_disks.ctx.tier_config_mgr(),
-            set_disks.ctx.object_encryption_resolver(),
-        )
-        .await
-        .expect("restore read of the tiered copy should open");
-        let published_size = reader.object_info.size;
-        let mut body = Vec::new();
-        reader.stream.read_to_end(&mut body).await.expect("restore body should drain");
-
-        assert_eq!(body, compressed, "a restore read must copy the stored bytes back verbatim");
-        assert_eq!(
-            published_size,
-            compressed.len() as i64,
-            "a restore read must keep publishing the stored size"
-        );
-    }
-
-    /// Plain objects must keep streaming the remote bytes through untouched:
-    /// their plan is `Plain`, so the tiered read stays byte-identical.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn transitioned_plain_object_get_is_unchanged() {
-        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
-        let bucket = "transitioned-plain-get-bucket";
-        let object = "object.bin";
-        let payload = b"plain transitioned objects must keep reading back byte-identical ".repeat(1024);
-        let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
-
-        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
-        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
-        set_disks
-            .transition_object(bucket, object, &transition_options(&original, tier_name))
-            .await
-            .expect("transition should commit");
-
-        let opts = ObjectOptions {
-            no_lock: true,
-            ..Default::default()
-        };
-        let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
-        assert_eq!(body, payload);
-        assert_eq!(published_size, payload.len() as i64);
-
-        let range = HTTPRangeSpec {
-            is_suffix_length: false,
-            start: 100,
-            end: 611,
-        };
-        let (ranged_body, ranged_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
-        assert_eq!(ranged_body, &payload[100..=611]);
-        assert_eq!(ranged_size, payload.len() as i64, "a plain ranged read keeps publishing the object size");
-    }
-
     async fn corrupt_beyond_read_quorum(
         temp_dirs: &[tempfile::TempDir],
         bucket: &str,

crates/ecstore/src/set_disk/read.rs

@@ -116,7 +116,6 @@ impl SetDisks {
             .then_some(GET_METADATA_CACHE_REASON_DIST_ERASURE)
     }
 
-    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     async fn cached_get_object_fileinfo(&self, bucket: &str, object: &str) -> Option<Arc<GetObjectMetadataCacheEntry>> {
         match self.lookup_cached_get_object_fileinfo(bucket, object).await {
             MetadataCacheLookup::Hit(entry) => Some(entry),
@@ -1827,7 +1826,6 @@ fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOp
         .then_some(GET_METADATA_CACHE_REASON_META_BUCKET)
 }
 
-#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOptions, read_data: bool) -> bool {
     get_object_metadata_cache_request_bypass_reason(bucket, opts, read_data).is_none()
 }
@@ -3937,7 +3935,7 @@ mod tests {
     }
 
     #[test]
-    fn metadata_early_stop_bounded_fanout_defaults_to_enabled() {
+    fn metadata_early_stop_bounded_fanout_defaults_to_disabled() {
         temp_env::with_vars(
             [
                 (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
@@ -3946,20 +3944,20 @@ mod tests {
             ],
             || {
                 assert!(is_get_metadata_data_read_early_stop_enabled());
-                assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
+                assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
             },
         );
-        temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false"))], || {
-            assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
+        temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true"))], || {
+            assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
         });
         temp_env::with_vars(
             [
                 (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("false")),
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")),
+                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
             ],
             || {
                 assert!(!is_get_metadata_data_read_early_stop_enabled());
-                assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
+                assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
             },
         );
     }

crates/io-metrics/src/lib.rs

@@ -487,21 +487,22 @@ pub fn record_get_object_completion(total_duration_secs: f64, response_size_byte
 
 /// Record the streaming strategy chosen for a GetObject response body.
 #[inline(always)]
-pub fn record_get_object_stream_strategy(strategy: &'static str, buffer_size_bytes: usize, response_size_bytes: i64) {
+pub fn record_get_object_stream_strategy(strategy: &str, buffer_size_bytes: usize, response_size_bytes: i64) {
     if !get_stage_metrics_enabled() {
         return;
     }
-    counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy).increment(1);
-    histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy).record(usize_to_f64(buffer_size_bytes));
-    histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy)
+    counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy.to_string()).increment(1);
+    histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy.to_string())
+        .record(usize_to_f64(buffer_size_bytes));
+    histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy.to_string())
         .record(i64_non_negative_to_f64(response_size_bytes));
 }
 
 /// Record the response-body handoff shape from a GetObject reader into the S3 streaming body.
 #[inline(always)]
 pub fn record_get_object_response_handoff(
-    strategy: &'static str,
-    buffer_source: &'static str,
+    strategy: &str,
+    buffer_source: &str,
     buffer_size_bytes: usize,
     response_size_bytes: i64,
     duration_secs: f64,
@@ -511,26 +512,26 @@ pub fn record_get_object_response_handoff(
     }
     counter!(
         "rustfs_io_get_object_response_handoff_total",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string()
     )
     .increment(1);
     histogram!(
         "rustfs_io_get_object_response_handoff_buffer_size_bytes",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string()
     )
     .record(usize_to_f64(buffer_size_bytes));
     histogram!(
         "rustfs_io_get_object_response_handoff_response_size_bytes",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string()
     )
     .record(i64_non_negative_to_f64(response_size_bytes));
     histogram!(
         "rustfs_io_get_object_response_handoff_duration_seconds",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string()
     )
     .record(duration_secs);
     record_get_object_response_handoff_duration("s3_handler", duration_secs);
@@ -538,18 +539,14 @@ pub fn record_get_object_response_handoff(
 
 /// Record ReaderStream capacity chosen for GetObject handoff.
 #[inline(always)]
-pub fn record_get_object_reader_stream_buffer_size(
-    strategy: &'static str,
-    buffer_source: &'static str,
-    buffer_size_bytes: usize,
-) {
+pub fn record_get_object_reader_stream_buffer_size(strategy: &str, buffer_source: &str, buffer_size_bytes: usize) {
     if !get_stage_metrics_enabled() {
         return;
     }
     histogram!(
         "rustfs_io_get_object_reader_stream_buffer_size_bytes",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string()
     )
     .record(usize_to_f64(buffer_size_bytes));
 }
@@ -557,8 +554,8 @@ pub fn record_get_object_reader_stream_buffer_size(
 /// Record ReaderStream poll outcomes for GetObject handoff attribution.
 #[inline(always)]
 pub fn record_get_object_reader_stream_poll(
-    strategy: &'static str,
-    buffer_source: &'static str,
+    strategy: &str,
+    buffer_source: &str,
     outcome: &'static str,
     remaining_before: usize,
     bytes: usize,
@@ -570,36 +567,36 @@ pub fn record_get_object_reader_stream_poll(
     let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
     counter!(
         "rustfs_io_get_object_reader_stream_poll_total",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source,
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .increment(1);
     counter!(
         "rustfs_io_get_object_reader_stream_poll_bytes_total",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source,
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .increment(bytes);
     histogram!(
         "rustfs_io_get_object_reader_stream_poll_remaining_bytes",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source,
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .record(usize_to_f64(remaining_before));
     histogram!(
         "rustfs_io_get_object_reader_stream_poll_bytes",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source,
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .record(usize_to_f64(bytes as usize));
     histogram!(
         "rustfs_io_get_object_reader_stream_poll_duration_seconds",
-        "strategy" => strategy,
-        "buffer_source" => buffer_source,
+        "strategy" => strategy.to_string(),
+        "buffer_source" => buffer_source.to_string(),
         "outcome" => outcome
     )
     .record(duration_secs);

crates/keystone/src/client.rs

@@ -31,10 +31,7 @@ pub struct KeystoneClient {
     admin_password: Option<String>,
     admin_project: Option<String>,
     admin_domain: String,
-    #[allow(
-        dead_code,
-        reason = "TLS verification flag parsed from config; the reqwest client is built before it is consulted, so nothing reads it back (backlog#1823)"
-    )]
+    #[allow(dead_code)]
     verify_ssl: bool,
     /// Request timeout applied to the underlying HTTP client.
     timeout: std::time::Duration,

crates/keystone/src/identity.rs

@@ -20,10 +20,7 @@ use tracing::{debug, info};
 
 /// Maps Keystone identities to RustFS concepts
 pub struct KeystoneIdentityMapper {
-    #[allow(
-        dead_code,
-        reason = "keeps the Keystone client alive for the mapper's lifetime; the mapping paths do not call through it yet (backlog#1823)"
-    )]
+    #[allow(dead_code)]
     client: Arc<KeystoneClient>,
     role_policy_map: HashMap<String, String>,
     enable_tenant_prefix: bool,

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 =
-                    BackendConfig::VaultKv2(Box::new(vault_kv2_config_from_env(VaultCliOverrides::default())?));
+                let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200");
+                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 =
-                    BackendConfig::VaultTransit(Box::new(vault_transit_config_from_env(VaultCliOverrides::default())?));
+                let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200");
+                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.
-///
-/// `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);
-
+fn vault_auth_method_from_env() -> Result<VaultAuthMethod> {
     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 [
-            (ENV_KMS_VAULT_APPROLE_ROLE_ID, role_id.is_some()),
-            (ENV_KMS_VAULT_KUBERNETES_ROLE, kubernetes_role.is_some()),
-            (ENV_KMS_VAULT_TOKEN, token.is_some()),
-        ] {
-            if configured {
-                return Err(KmsError::configuration_error(format!(
-                    "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with {name}; configure exactly one Vault auth method"
-                )));
-            }
+        if get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID).is_some() {
+            return Err(KmsError::configuration_error(format!(
+                "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with {ENV_KMS_VAULT_APPROLE_ROLE_ID}; configure exactly one Vault auth method"
+            )));
+        }
+        if get_env_opt_str("RUSTFS_KMS_VAULT_TOKEN").is_some() {
+            return Err(KmsError::configuration_error(format!(
+                "{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 {
-        // 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 {
+    let Some(role_id) = get_env_opt_str(ENV_KMS_VAULT_APPROLE_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/kms/src/encryption/dek.rs

@@ -18,6 +18,8 @@
 //! data encryption keys using master keys. It abstracts the encryption
 //! operations so that different backends can share the same encryption logic.
 
+#![allow(dead_code)] // Trait methods may be used by implementations
+
 use crate::error::{KmsError, Result};
 use crate::persisted_observability::{BoundedUnknownFieldName, UnknownFieldSummary};
 use async_trait::async_trait;

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,

crates/notify/src/rules/config.rs

@@ -40,10 +40,7 @@ impl RuleEvents for RuleView {
 #[derive(Debug)]
 struct CompiledRules {
     // Keep RulesMap (can be used later if you want to make more complex judgments during the snapshot reading phase)
-    #[allow(
-        dead_code,
-        reason = "speculative retention: the comment above keeps it for richer snapshot-time judgements that no code performs yet (backlog#1823)"
-    )]
+    #[allow(dead_code)]
     rules_map: RulesMap,
     // for RulesContainer::iter_rules
     rule_views: Vec<RuleView>,

crates/notify/src/rules/rules_map.rs

@@ -187,6 +187,7 @@ impl RulesMap {
     /// # Parameters
     /// * `event_name` - The EventName from which to remove the rule.
     /// * `pattern` - The pattern of the rule to be removed.
+    #[allow(dead_code)]
     pub fn remove_rule(&mut self, event_name: &EventName, pattern: &str) {
         let mut remove_event = false;
 
@@ -208,6 +209,7 @@ impl RulesMap {
     ///
     /// # Parameters
     /// * `event_names` - A slice of EventNames to be removed.
+    #[allow(dead_code)]
     pub fn remove_rules(&mut self, event_names: &[EventName]) {
         for event_name in event_names {
             self.map.remove(event_name);
@@ -221,6 +223,7 @@ impl RulesMap {
     /// * `event_name` - The EventName to update.
     /// * `pattern` - The pattern of the rule to be updated.
     /// * `target_id` - The TargetID to be added.
+    #[allow(dead_code)]
     pub fn update_rule(&mut self, event_name: EventName, pattern: String, target_id: TargetID) {
         self.map.entry(event_name).or_default().add(pattern, target_id);
         self.total_events_mask |= event_name.mask(); // Update only the relevant bitmask

crates/notify/src/rules/target_id_set.rs

@@ -18,6 +18,12 @@ use rustfs_targets::arn::TargetID;
 /// TargetIDSet - A collection representation of TargetID.
 pub type TargetIdSet = HashSet<TargetID>;
 
+/// Provides a Go-like method for TargetIdSet (can be implemented as trait if needed)
+#[allow(dead_code)]
+pub(crate) fn new_target_id_set(target_ids: Vec<TargetID>) -> TargetIdSet {
+    target_ids.into_iter().collect()
+}
+
 // HashSet has built-in clone, union, difference and other operations.
 // But the Go version of the method returns a new Set, and the HashSet method is usually iterator or modify itself.
 // If you need to exactly match Go's API style, you can add wrapper functions.

crates/obs/src/metrics/config.rs

@@ -17,6 +17,7 @@ use std::time::Duration;
 /// Environment variable key for the global default metrics interval (seconds).
 pub const ENV_DEFAULT_METRICS_INTERVAL: &str = "RUSTFS_METRICS_DEFAULT_INTERVAL_SEC";
 /// Default interval for metrics collection if not specified otherwise.
+#[allow(dead_code)]
 pub const DEFAULT_METRICS_INTERVAL: Duration = Duration::from_secs(60);
 
 /// Environment variable key for cluster metrics interval (seconds).

crates/obs/src/metrics/report.rs

@@ -145,18 +145,21 @@ impl PrometheusMetric {
     }
 
     #[inline]
+    #[allow(dead_code)]
     pub fn with_label(mut self, key: &'static str, value: impl Into<Cow<'static, str>>) -> Self {
         self.labels.push((key, value.into()));
         self
     }
 
     #[inline]
+    #[allow(dead_code)]
     pub fn with_label_owned(mut self, key: &'static str, value: String) -> Self {
         self.labels.push((key, Cow::Owned(value)));
         self
     }
 
     #[inline]
+    #[allow(dead_code)]
     pub fn with_labels(mut self, labels: Vec<(&'static str, Cow<'static, str>)>) -> Self {
         self.labels = labels;
         self

crates/obs/src/metrics/schema/entry/descriptor.rs

@@ -16,6 +16,7 @@ use crate::{MetricName, MetricNamespace, MetricSubsystem, MetricType};
 use std::collections::HashSet;
 
 /// MetricDescriptor - Metric descriptors
+#[allow(dead_code)]
 #[derive(Debug, Clone)]
 pub struct MetricDescriptor {
     pub name: MetricName,
@@ -51,6 +52,7 @@ impl MetricDescriptor {
     }
 
     /// Get the full metric name in Prometheus style: <namespace>_<subsystem>_<name>
+    #[allow(dead_code)]
     pub fn get_full_metric_name(&self) -> String {
         let namespace = self.namespace.as_str();
         let formatted_subsystem = self.subsystem.as_str();
@@ -59,6 +61,7 @@ impl MetricDescriptor {
     }
 
     /// check whether the label is in the label set
+    #[allow(dead_code)]
     pub fn has_label(&mut self, label: &str) -> bool {
         self.get_label_set().contains(label)
     }

crates/obs/src/metrics/schema/entry/metric_name.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 /// The metric name is the individual name of the metric
+#[allow(dead_code)]
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum MetricName {
     // The generic metric name
@@ -442,6 +443,7 @@ pub enum MetricName {
 }
 
 impl MetricName {
+    #[allow(dead_code)]
     pub fn as_str(&self) -> String {
         match self {
             Self::AuthTotal => "auth_total".to_string(),

crates/obs/src/metrics/schema/entry/metric_type.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 /// MetricType - Indicates the type of indicator
+#[allow(dead_code)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum MetricType {
     Counter,
@@ -22,6 +23,7 @@ pub enum MetricType {
 
 impl MetricType {
     /// convert the metric type to a string representation
+    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Counter => "counter",
@@ -32,6 +34,7 @@ impl MetricType {
 
     /// Convert the metric type to the Prometheus value type
     /// In a Rust implementation, this might return the corresponding Prometheus Rust client type
+    #[allow(dead_code)]
     pub fn as_prom(&self) -> &'static str {
         match self {
             Self::Counter => "counter.",

crates/obs/src/metrics/schema/entry/mod.rs

@@ -56,6 +56,7 @@ pub fn new_gauge_md(
 }
 
 /// create a new histogram indicator descriptor
+#[allow(dead_code)]
 pub fn new_histogram_md(
     name: impl Into<MetricName>,
     help: impl Into<String>,

crates/obs/src/metrics/schema/entry/namespace.rs

@@ -19,6 +19,7 @@ pub enum MetricNamespace {
 }
 
 impl MetricNamespace {
+    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::RustFS => "rustfs",

crates/obs/src/metrics/schema/entry/path_utils.rs

@@ -14,6 +14,7 @@
 
 /// Format the path to the metric name format
 /// Replace '/' and '-' with '_'
+#[allow(dead_code)]
 pub fn format_path_to_metric_name(path: &str) -> String {
     path.trim_start_matches('/').replace(['/', '-'], "_")
 }

crates/obs/src/metrics/schema/entry/subsystem.rs

@@ -102,6 +102,7 @@ impl MetricSubsystem {
     }
 
     /// Get the formatted metric name format string
+    #[allow(dead_code)]
     pub fn as_str(&self) -> String {
         format_path_to_metric_name(self.path())
     }
@@ -150,6 +151,7 @@ impl MetricSubsystem {
     }
 
     /// A convenient way to create custom subsystems directly
+    #[allow(dead_code)]
     pub fn new(path: impl Into<String>) -> Self {
         Self::Custom(path.into())
     }
@@ -174,6 +176,7 @@ impl std::fmt::Display for MetricSubsystem {
     }
 }
 
+#[allow(dead_code)]
 pub mod subsystems {
     use super::MetricSubsystem;
 

crates/obs/src/telemetry/rolling.rs

@@ -38,10 +38,7 @@ pub enum Rotation {
     Minutely,
     Hourly,
     Daily,
-    #[allow(
-        dead_code,
-        reason = "constructed only by this file's rolling-appender tests; the lib target cannot see them (backlog#1823)"
-    )]
+    #[allow(dead_code)]
     Never,
 }
 

crates/policy/src/policy/function.rs

@@ -219,6 +219,10 @@ impl PartialEq for Functions {
     }
 }
 
+#[derive(Clone, Serialize, Deserialize)]
+#[allow(dead_code)]
+pub struct Value;
+
 #[cfg(test)]
 mod tests {
     use crate::policy::Functions;

crates/policy/src/policy/utils/wildcard.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[allow(dead_code)]
 pub fn is_simple_match<P, N>(pattern: P, name: N) -> bool
 where
     P: AsRef<str>,
@@ -28,10 +29,7 @@ where
     inner_match(pattern, name, false)
 }
 
-#[allow(
-    dead_code,
-    reason = "prefix-matcher asserted by this file's tests; no production caller yet (backlog#1823)"
-)]
+#[allow(dead_code)]
 pub fn is_match_as_pattern_prefix<P, N>(pattern: P, text: N) -> bool
 where
     P: AsRef<str>,

crates/protocols/src/swift/account.rs

@@ -38,6 +38,7 @@ use std::collections::HashMap;
 /// - Account format is invalid
 /// - Credentials don't contain project_id
 /// - Account project_id doesn't match credentials project_id
+#[allow(dead_code)] // Used by Swift implementation
 pub fn validate_account_access(account: &str, credentials: &Credentials) -> SwiftResult<String> {
     // Extract project_id from account (strip "AUTH_" prefix)
     let account_project_id = account
@@ -69,6 +70,7 @@ pub fn validate_account_access(account: &str, credentials: &Credentials) -> Swif
 ///
 /// Admin users (with "admin" or "reseller_admin" roles) can perform
 /// cross-tenant operations and administrative tasks.
+#[allow(dead_code)] // Used by Swift implementation
 pub fn is_admin_user(credentials: &Credentials) -> bool {
     credentials
         .claims

crates/protocols/src/swift/container.rs

@@ -144,6 +144,7 @@ impl ContainerMapper {
     /// - S3 bucket name compatible (only uses [a-z0-9-])
     /// - Deterministic mapping (same input always produces same bucket name)
     /// - Fixed-length prefix (16 hex chars = 8 bytes)
+    #[allow(dead_code)] // Used in: create/delete container operations
     pub fn swift_to_s3_bucket(&self, container: &str, project_id: &str) -> String {
         if self.config.tenant_prefix_enabled {
             let hash = self.hash_project_id(project_id);
@@ -215,6 +216,7 @@ pub fn bucket_info_to_container(info: &BucketInfo, mapper: &ContainerMapper, pro
 /// 2. Lists all S3 buckets
 /// 3. Filters to buckets belonging to this tenant (using tenant prefix)
 /// 4. Converts BucketInfo to Swift Container format
+#[allow(dead_code)] // Used by handler: list containers
 pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftResult<Vec<Container>> {
     // Validate account access and extract project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -277,6 +279,7 @@ pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftR
 /// - Returns 201 Created on success
 /// - Returns 202 Accepted if container already exists
 /// - Returns 400 Bad Request for invalid container names
+#[allow(dead_code)] // Used by handler
 pub async fn create_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<bool> {
     // Validate account access and extract project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -345,6 +348,7 @@ fn validate_container_name(container: &str) -> SwiftResult<()> {
 }
 
 /// Container metadata for HEAD response
+#[allow(dead_code)] // TODO: Remove once Swift API integration is complete
 #[derive(Debug, Clone)]
 pub struct ContainerMetadata {
     /// Number of objects in container
@@ -407,6 +411,7 @@ pub(crate) async fn get_container_custom_metadata(
 /// - HEAD /v1/{account}/{container} returns container metadata
 /// - Returns 204 No Content on success with headers
 /// - Returns 404 Not Found if container doesn't exist
+#[allow(dead_code)] // Used by handler
 pub async fn get_container_metadata(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<ContainerMetadata> {
     let (bucket_name, bucket_info, custom_metadata) = get_container_metadata_base(account, container, credentials).await?;
 
@@ -443,6 +448,7 @@ pub async fn get_container_metadata(account: &str, container: &str, credentials:
 /// - The update is additive: items the request does not name keep their stored
 ///   value, and removal is explicit, via `X-Remove-Container-Meta-{name}` or an
 ///   empty value
+#[allow(dead_code)] // Used by handler
 pub async fn update_container_metadata(
     account: &str,
     container: &str,
@@ -514,6 +520,7 @@ pub async fn update_container_metadata(
 /// - Returns 204 No Content on success
 /// - Returns 404 Not Found if container doesn't exist
 /// - Returns 409 Conflict if container is not empty
+#[allow(dead_code)] // Used by handler
 pub async fn delete_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
     // Validate account access and extract project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -596,6 +603,7 @@ pub async fn delete_container(account: &str, container: &str, credentials: &Cred
 /// - Account validation fails
 /// - Container doesn't exist
 /// - Storage layer errors occur
+#[allow(dead_code)] // Handler integration: GET container
 pub async fn list_objects(
     account: &str,
     container: &str,
@@ -698,6 +706,7 @@ pub async fn list_objects(
 /// Versioning configuration is stored as an S3 bucket tag:
 /// - Tag key: `swift-versions-location`
 /// - Tag value: archive container name
+#[allow(dead_code)] // Used by handler
 pub async fn enable_versioning(
     account: &str,
     container: &str,
@@ -786,6 +795,7 @@ pub async fn enable_versioning(
 /// * `account` - Account identifier
 /// * `container` - Container name to disable versioning on
 /// * `credentials` - Keystone credentials
+#[allow(dead_code)] // Used by handler
 pub async fn disable_versioning(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
     // Validate account access
     let project_id = validate_account_access(account, credentials)?;
@@ -845,6 +855,7 @@ pub async fn disable_versioning(account: &str, container: &str, credentials: &Cr
 /// # Returns
 /// - Some(archive_container_name) if versioning is enabled
 /// - None if versioning is not enabled
+#[allow(dead_code)] // Used by handler and object.rs
 pub async fn get_versions_location(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<Option<String>> {
     // Validate account access
     let project_id = validate_account_access(account, credentials)?;
@@ -907,6 +918,7 @@ pub async fn get_versions_location(account: &str, container: &str, credentials:
 ///     &credentials
 /// ).await?;
 /// ```
+#[allow(dead_code)] // Used by handler
 pub async fn set_container_acl(
     account: &str,
     container: &str,
@@ -1010,6 +1022,7 @@ pub async fn set_container_acl(
 ///     println!("Container is publicly readable");
 /// }
 /// ```
+#[allow(dead_code)] // Used by handler
 pub async fn get_container_acl(
     account: &str,
     container: &str,
@@ -1070,6 +1083,7 @@ pub async fn get_container_acl(
 ///
 /// # Returns
 /// Ok(()) if ACLs were deleted successfully
+#[allow(dead_code)] // Used by handler
 pub async fn delete_container_acl(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
     // Setting both ACLs to None removes them
     set_container_acl(account, container, None, None, credentials).await

crates/protocols/src/swift/errors.rs

@@ -20,6 +20,7 @@ use std::fmt;
 
 /// Swift-specific error type
 #[derive(Debug)]
+#[allow(dead_code)] // Error variants used by Swift implementation
 pub enum SwiftError {
     /// 400 Bad Request
     BadRequest(String),

crates/protocols/src/swift/object.rs

@@ -122,10 +122,12 @@ fn swift_user_metadata(headers: &HeaderMap) -> Option<HashMap<String, String>> {
 ///
 /// Handles URL encoding/decoding and path normalization for Swift object keys.
 /// Swift object names can contain any UTF-8 characters except null bytes.
+#[allow(dead_code)] // Used in: object operations
 pub struct ObjectKeyMapper;
 
 impl ObjectKeyMapper {
     /// Create a new object key mapper
+    #[allow(dead_code)] // Used in: object operations
     pub fn new() -> Self {
         Self
     }
@@ -138,6 +140,7 @@ impl ObjectKeyMapper {
     /// - Not contain null bytes
     /// - Not contain '..' path segments (directory traversal)
     /// - Not start with '/' (leading slash handled by routing)
+    #[allow(dead_code)] // Used in: object operations
     pub fn validate_object_name(object: &str) -> SwiftResult<()> {
         if object.is_empty() {
             return Err(SwiftError::BadRequest("Object name cannot be empty".to_string()));
@@ -180,6 +183,7 @@ impl ObjectKeyMapper {
     /// Example:
     /// - Swift: "photos/vacation/beach photo.jpg"
     /// - S3: "photos/vacation/beach photo.jpg"
+    #[allow(dead_code)] // Used in: object operations
     pub fn swift_to_s3_key(object: &str) -> SwiftResult<String> {
         Self::validate_object_name(object)?;
         Ok(object.to_string())
@@ -189,6 +193,7 @@ impl ObjectKeyMapper {
     ///
     /// This is essentially an identity transformation since we store
     /// Swift object names as-is in S3.
+    #[allow(dead_code)] // Used in: object operations
     pub fn s3_to_swift_name(key: &str) -> String {
         key.to_string()
     }
@@ -203,6 +208,7 @@ impl ObjectKeyMapper {
     /// - Object: "vacation/beach.jpg"
     /// - Bucket: "abc123:photos"
     /// - Key: "vacation/beach.jpg"
+    #[allow(dead_code)] // Used in: object operations
     pub fn build_s3_key(object: &str) -> SwiftResult<String> {
         Self::swift_to_s3_key(object)
     }
@@ -214,6 +220,7 @@ impl ObjectKeyMapper {
     ///
     /// Example URL: /v1/AUTH_abc/container/path%2Fto%2Ffile.txt
     /// Decoded: "path/to/file.txt"
+    #[allow(dead_code)] // Used in: object operations
     pub fn decode_object_from_url(encoded: &str) -> SwiftResult<String> {
         // Decode percent-encoding
         let decoded = urlencoding::decode(encoded).map_err(|e| SwiftError::BadRequest(format!("Invalid URL encoding: {}", e)))?;
@@ -226,6 +233,7 @@ impl ObjectKeyMapper {
     ///
     /// When constructing URLs (e.g., for redirect responses), we need to
     /// percent-encode object names.
+    #[allow(dead_code)] // Used in: object operations
     pub fn encode_object_for_url(object: &str) -> String {
         urlencoding::encode(object).to_string()
     }
@@ -233,6 +241,7 @@ impl ObjectKeyMapper {
     /// Check if object name represents a directory (pseudo-directory)
     ///
     /// In Swift, objects ending with '/' are treated as directory markers.
+    #[allow(dead_code)] // Used in: object operations
     pub fn is_directory_marker(object: &str) -> bool {
         object.ends_with('/')
     }
@@ -241,6 +250,7 @@ impl ObjectKeyMapper {
     ///
     /// Removes redundant slashes and normalizes the path while preserving
     /// trailing slashes for directory markers.
+    #[allow(dead_code)] // Used in: object operations
     pub fn normalize_path(object: &str) -> String {
         // Split by '/', filter out empty segments (except if it's the end)
         let has_trailing_slash = object.ends_with('/');
@@ -314,6 +324,7 @@ fn sanitize_storage_error<E: std::fmt::Display>(operation: &str, error: E) -> Sw
 /// # Returns
 /// * `Ok(etag)` - Object ETag on success
 /// * `Err(SwiftError)` - Error if validation fails or upload fails
+#[allow(dead_code)] // Handler integration: PUT object
 pub async fn put_object<R>(
     account: &str,
     container: &str,
@@ -434,6 +445,7 @@ where
 ///
 /// Similar to put_object, but allows directly specifying metadata instead of extracting from headers.
 /// This is used internally for storing SLO manifests and marker objects.
+#[allow(dead_code)] // Used by SLO implementation
 pub async fn put_object_with_metadata<R>(
     account: &str,
     container: &str,
@@ -537,6 +549,7 @@ where
 /// - `bytes=1000-1999` - Bytes 1000-1999
 /// - `bytes=1000-` - From byte 1000 to end
 /// - `bytes=-500` - Last 500 bytes
+#[allow(dead_code)] // Handler integration: GET object
 pub async fn get_object(
     account: &str,
     container: &str,
@@ -595,6 +608,7 @@ pub async fn get_object(
 /// # Returns
 /// * `Ok(object_info)` - Object metadata (ObjectInfo)
 /// * `Err(SwiftError)` - Error if validation fails or object not found
+#[allow(dead_code)] // Handler integration: HEAD object
 pub async fn head_object(
     account: &str,
     container: &str,
@@ -657,6 +671,7 @@ pub async fn head_object(
 /// # Returns
 /// * `Ok(())` - Object deleted successfully (or didn't exist)
 /// * `Err(SwiftError)` - Error if validation fails or deletion fails
+#[allow(dead_code)] // Handler integration: DELETE object
 pub async fn delete_object(account: &str, container: &str, object: &str, credentials: &Credentials) -> SwiftResult<()> {
     // 1. Validate account access and get project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -717,6 +732,7 @@ pub async fn delete_object(account: &str, container: &str, object: &str, credent
 /// # Returns
 /// * `Ok(())` - Metadata updated successfully
 /// * `Err(SwiftError)` - Error if validation fails, object not found, or update fails
+#[allow(dead_code)] // Handler integration: POST object
 pub async fn update_object_metadata(
     account: &str,
     container: &str,
@@ -830,6 +846,7 @@ pub async fn update_object_metadata(
 /// # Handler Integration Note
 /// The current handler architecture needs to be updated to pass headers through
 /// to support COPY method and X-Copy-From header detection. See handler.rs for details.
+#[allow(dead_code)] // Handler integration: COPY object
 #[allow(clippy::too_many_arguments)] // Necessary for full copy functionality
 pub async fn copy_object(
     src_account: &str,
@@ -962,6 +979,7 @@ pub async fn copy_object(
 /// assert_eq!(container, "my-container");
 /// assert_eq!(object, "path/to/file.txt");
 /// ```
+#[allow(dead_code)] // Handler integration: COPY method
 pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, String)> {
     let destination = destination.trim_start_matches('/');
     let parts: Vec<&str> = destination.splitn(2, '/').collect();
@@ -995,6 +1013,7 @@ pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, Strin
 /// # Returns
 /// * `Ok((container, object))` - Parsed container and object names
 /// * `Err(SwiftError)` - Error if format is invalid
+#[allow(dead_code)] // Handler integration: X-Copy-From
 pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)> {
     // Same parsing logic as Destination header
     parse_destination_header(copy_from)
@@ -1023,6 +1042,7 @@ pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)>
 /// assert_eq!(range.start, 0);
 /// assert_eq!(range.end, 1023);
 /// ```
+#[allow(dead_code)] // Handler integration: Range header
 pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
     if !range_str.starts_with("bytes=") {
         return Err(SwiftError::BadRequest("Range header must start with 'bytes='".to_string()));
@@ -1104,6 +1124,7 @@ pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
 /// let header = format_content_range(0, 1023, 5000);
 /// assert_eq!(header, "bytes 0-1023/5000");
 /// ```
+#[allow(dead_code)] // Handler integration: Range header
 pub fn format_content_range(start: i64, end: i64, total: i64) -> String {
     format!("bytes {}-{}/{}", start, end, total)
 }

crates/protocols/src/swift/router.rs

@@ -50,6 +50,7 @@ pub enum SwiftRoute {
 
 impl SwiftRoute {
     /// Get the account identifier from the route
+    #[allow(dead_code)] // Public API for future use
     pub fn account(&self) -> &str {
         match self {
             SwiftRoute::Account { account, .. } => account,
@@ -59,6 +60,7 @@ impl SwiftRoute {
     }
 
     /// Extract project_id from account string (removes AUTH_ prefix)
+    #[allow(dead_code)] // Public API for future use
     pub fn project_id(&self) -> Option<&str> {
         let account = self.account();
         ACCOUNT_PATTERN

crates/protocols/src/swift/types.rs

@@ -19,6 +19,7 @@ use std::collections::HashMap;
 
 /// Swift container metadata
 #[derive(Debug, Clone, Serialize, Deserialize)]
+#[allow(dead_code)] // Used in container listing operations
 pub struct Container {
     /// Container name
     pub name: String,
@@ -33,6 +34,7 @@ pub struct Container {
 
 /// Swift object metadata
 #[derive(Debug, Clone, Serialize, Deserialize)]
+#[allow(dead_code)] // Used in object listing operations
 pub struct Object {
     /// Object name (key)
     pub name: String,
@@ -48,6 +50,7 @@ pub struct Object {
 
 /// Swift metadata extracted from headers
 #[derive(Debug, Clone, Default)]
+#[allow(dead_code)] // Used by Swift implementation
 pub struct SwiftMetadata {
     /// Custom metadata key-value pairs (from X-Container-Meta-* or X-Object-Meta-*)
     pub metadata: HashMap<String, String>,

crates/rio-v2/tests/minio_fixture_lab/README.md

@@ -34,6 +34,14 @@ MINIO_LAB_PYTHON_IMAGE=public.ecr.aws/docker/library/python:3.12-slim \
 
 Pin the MinIO tag to the same release the Dockerfile names; an unpinned `:latest` captures whatever format that day's build writes, which is not what the interop tests were validated against.
 
+## Capturing the SSE-C cases
+
+MinIO refuses SSE-C over a plain-HTTP connection, so the `sse-c-*` cases cannot be captured against the default endpoint — `./capture_via_docker.sh all` fails on the first SSE-C upload with `InvalidRequest ... must be made over a secure connection`. The lab provisions its own self-signed certificate; point it at the HTTPS endpoint to capture them:
+
+```bash
+MINIO_LAB_ENDPOINT=https://127.0.0.1:9000 ./capture_via_docker.sh all
+```
+
 ## Layout
 
 The default root is `artifacts/minio-fixture-lab`, which is already ignored by the repository.

crates/rio-v2/tests/minio_fixture_lab/capture_via_docker.sh

@@ -12,6 +12,13 @@
 #                                                 # ignored interop tests consume
 #   ./capture_via_docker.sh sse-s3-singlepart-64k # specific case id(s)
 #   ./capture_via_docker.sh all                   # full SSE/size matrix
+#
+# The SSE-C cases are not reachable over the default plain-HTTP endpoint: MinIO
+# refuses SSE-C unless the connection is secure ("Requests specifying Server
+# Side Encryption with Customer provided keys must be made over a secure
+# connection"). Capture those by pointing the lab at its self-signed HTTPS
+# endpoint, which it provisions itself:
+#   MINIO_LAB_ENDPOINT=https://127.0.0.1:9000 ./capture_via_docker.sh all
 set -euo pipefail
 
 IMAGE="${MINIO_LAB_IMAGE:-rustfs-minio-lab:latest}"
@@ -46,7 +53,10 @@ if [ -n "${MINIO_LAB_PYTHON_IMAGE:-}" ]; then
 fi
 
 echo ">> building ${IMAGE}"
-docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" "${build_args[@]}" "${SCRIPT_DIR}"
+# ${arr[@]+"${arr[@]}"} rather than "${arr[@]}": under `set -u`, bash 3.2 —
+# still the default /bin/bash on macOS — treats an empty array expansion as an
+# unbound variable and aborts.
+docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" ${build_args[@]+"${build_args[@]}"} "${SCRIPT_DIR}"
 
 echo ">> capturing fixtures into ${FIXTURE_REL}"
 docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \
@@ -54,6 +64,7 @@ docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \
     --root "/repo/${FIXTURE_REL}" \
     --work-root /tmp/minio-lab-work \
     --minio-binary /usr/local/bin/minio \
-    "${case_args[@]}"
+    --endpoint "${MINIO_LAB_ENDPOINT:-http://127.0.0.1:9000}" \
+    ${case_args[@]+"${case_args[@]}"}
 
 echo ">> done — fixtures under ${REPO_ROOT}/${FIXTURE_REL}/cases/"

crates/scanner/src/scanner_folder.rs

@@ -4621,8 +4621,6 @@ mod tests {
         let _subscriber_guard = tracing::subscriber::set_default(subscriber);
 
         let (mut scanner, temp_dir) = build_test_scanner().await;
-        // Canonicalize for the "drive" field comparison (scanner resolves symlinks).
-        let canonical_temp_dir = std::fs::canonicalize(&temp_dir).unwrap_or_else(|_| temp_dir.clone());
         let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
 
         let object_dir = temp_dir.join("bucket").join("object");
@@ -4691,7 +4689,7 @@ mod tests {
         let fields = &events[0]["fields"];
         assert_eq!(fields["component"], LOG_COMPONENT_SCANNER);
         assert_eq!(fields["subsystem"], LOG_SUBSYSTEM_FOLDER);
-        assert_eq!(fields["drive"], canonical_temp_dir.to_string_lossy().as_ref());
+        assert_eq!(fields["drive"], temp_dir.to_string_lossy().as_ref());
         assert_eq!(fields["bucket"], "bucket");
         assert_eq!(fields["object"], "object");
         assert_eq!(fields["metadata_path"], metadata_path.to_string_lossy().as_ref());

crates/trusted-proxies/src/cloud/metadata/aws.rs

@@ -27,10 +27,6 @@ use crate::CloudMetadataFetcher;
 #[derive(Debug, Clone)]
 pub struct AwsMetadataFetcher {
     client: Client,
-    #[allow(
-        dead_code,
-        reason = "IMDS endpoint retained beside the client it configures; requests build their own URLs (backlog#1823)"
-    )]
     metadata_endpoint: String,
 }
 
@@ -50,6 +46,55 @@ impl AwsMetadataFetcher {
             metadata_endpoint: "http://169.254.169.254".to_string(),
         }
     }
+
+    /// Retrieves an IMDSv2 token for secure metadata access.
+    #[allow(dead_code)]
+    async fn get_metadata_token(&self) -> Result<String, AppError> {
+        let url = format!("{}/latest/api/token", self.metadata_endpoint);
+
+        match self
+            .client
+            .put(&url)
+            .header("X-aws-ec2-metadata-token-ttl-seconds", "21600")
+            .send()
+            .await
+        {
+            Ok(response) => {
+                if response.status().is_success() {
+                    let token = response
+                        .text()
+                        .await
+                        .map_err(|e| AppError::cloud(format!("Failed to read IMDSv2 token: {}", e)))?;
+                    Ok(token)
+                } else {
+                    debug!(
+                        event = "trusted_proxies.cloud_metadata",
+                        component = "trusted_proxies",
+                        subsystem = "aws_metadata",
+                        provider = "aws",
+                        operation = "imdsv2_token",
+                        result = "http_error",
+                        status = %response.status(),
+                        "trusted proxy cloud metadata request failed"
+                    );
+                    Err(AppError::cloud("Failed to obtain IMDSv2 token"))
+                }
+            }
+            Err(e) => {
+                debug!(
+                    event = "trusted_proxies.cloud_metadata",
+                    component = "trusted_proxies",
+                    subsystem = "aws_metadata",
+                    provider = "aws",
+                    operation = "imdsv2_token",
+                    result = "request_failed",
+                    error = %e,
+                    "trusted proxy cloud metadata request failed"
+                );
+                Err(AppError::cloud(format!("IMDSv2 request failed: {}", e)))
+            }
+        }
+    }
 }
 
 #[async_trait]

crates/trusted-proxies/src/config/env.rs

@@ -68,6 +68,7 @@ pub fn is_env_set(key: &str) -> bool {
 }
 
 /// Returns a list of all proxy-related environment variables and their current values.
+#[allow(dead_code)]
 pub fn get_all_proxy_env_vars() -> Vec<(String, String)> {
     let vars = [
         ENV_TRUSTED_PROXY_ENABLED,

crates/utils/src/io.rs

@@ -68,6 +68,7 @@ pub async fn read_full_or_eof<R: AsyncRead + Send + Sync + Unpin>(
 
 /// Read exactly buf.len() bytes into buf, or return an error if EOF is reached before any bytes are read.
 /// Like Go's io.ReadFull.
+#[allow(dead_code)]
 pub async fn read_full<R: AsyncRead + Send + Sync + Unpin>(reader: R, buf: &mut [u8]) -> std::io::Result<usize> {
     match read_full_or_eof(reader, buf).await? {
         Some(n) => Ok(n),

crates/utils/src/net.rs

@@ -431,6 +431,7 @@ pub fn parse_and_resolve_address(addr_str: &str) -> std::io::Result<SocketAddr>
     Ok(resolved_addr)
 }
 
+#[allow(dead_code)]
 pub fn bytes_stream<S, E>(stream: S, content_length: usize) -> impl Stream<Item = Result<Bytes, E>> + Send + 'static
 where
     S: Stream<Item = Result<Bytes, E>> + Send + 'static,

crates/utils/src/os/fs_type.rs

@@ -16,6 +16,7 @@
 ///
 /// The table follows Linux `include/uapi/linux/magic.h`; filesystem magic
 /// values without a stable Linux uapi source stay `UNKNOWN`.
+#[allow(dead_code)]
 pub(crate) fn get_fs_type(fs_type: u64) -> &'static str {
     // Magic numbers for various filesystems.
     match fs_type {

crates/utils/src/path.rs

@@ -70,6 +70,7 @@ pub fn is_dir_object(object: &str) -> bool {
 ///
 /// If the object name ends with `GLOBAL_DIR_SUFFIX`, it is replaced with a slash.
 /// Otherwise, the name is returned as is.
+#[allow(dead_code)]
 pub fn decode_dir_object(object: &str) -> String {
     if has_suffix(object, GLOBAL_DIR_SUFFIX) {
         format!("{}{}", object.trim_end_matches(GLOBAL_DIR_SUFFIX), SLASH_SEPARATOR)

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.
-
-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`.
+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.
 
 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.
-
-## 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.
+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.
 
 ## 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 |
-| 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 |
+| 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 |
 
 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
-        // 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 { .. } => {}
+        VaultAuthMethod::TokenFile { .. } => {}
     };
 
     match &config.backend_config {

rustfs/src/admin/handlers/scanner.rs

@@ -16,8 +16,8 @@ use crate::admin::auth::validate_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_scanner_metrics_report;
 use crate::auth::{check_key_valid, get_session_token};
-use crate::module_switches::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
 use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::startup_background::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
 use chrono::Utc;
 use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};

rustfs/src/admin/handlers/table_catalog/mod.rs

@@ -2995,9 +2995,9 @@ fn table_entry_from_create_table_request(
     let CreateTableRequest {
         name,
         location,
-        mut schema,
-        mut partition_spec,
-        mut write_order,
+        schema,
+        partition_spec,
+        write_order,
         stage_create,
         mut properties,
     } = request;
@@ -3031,9 +3031,6 @@ fn table_entry_from_create_table_request(
     let metadata_location =
         crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id));
 
-    crate::table_catalog::assign_fresh_create_schema_ids(&mut schema, partition_spec.as_mut(), write_order.as_mut())
-        .map_err(catalog_store_error)?;
-
     let entry = crate::table_catalog::TableEntry {
         version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
         table_bucket: bucket.to_string(),

rustfs/src/admin/handlers/table_catalog/tests.rs

@@ -1873,66 +1873,6 @@ fn create_table_request_accepts_standard_iceberg_rest_shape() {
     assert_eq!(request.name, "events");
 }
 
-#[test]
-fn create_table_assigns_positive_ids_to_spark_schema() {
-    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
-    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
-        "name": "events",
-        "schema": {
-            "type": "struct",
-            "schema-id": 0,
-            "fields": [
-                {"id": 0, "name": "id", "required": false, "type": "long"},
-                {"id": 1, "name": "payload", "required": false, "type": "string"}
-            ]
-        },
-        "partition-spec": {"spec-id": 0, "fields": []},
-        "properties": {"owner": "spark"}
-    }))
-    .expect("Spark create table request should parse");
-
-    let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
-        .expect("catalog should assign positive field IDs");
-
-    assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
-    assert_eq!(metadata["schemas"][0]["fields"][1]["id"], 2);
-    assert_eq!(metadata["last-column-id"], 2);
-}
-
-#[test]
-fn create_table_assigns_fresh_id_to_negative_temporary_field_id() {
-    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
-    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
-        "name": "events",
-        "schema": {
-            "type": "struct",
-            "identifier-field-ids": [-1],
-            "fields": [{"id": -1, "name": "id", "required": true, "type": "long"}]
-        },
-        "partition-spec": {
-            "fields": [{"source-id": -1, "name": "id", "transform": "identity"}]
-        },
-        "write-order": {
-            "fields": [{
-                "source-id": -1,
-                "transform": "identity",
-                "direction": "asc",
-                "null-order": "nulls-first"
-            }]
-        }
-    }))
-    .expect("create table request with a negative temporary field ID should parse");
-
-    let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
-        .expect("catalog should replace the negative temporary field ID");
-
-    assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
-    assert_eq!(metadata["schemas"][0]["identifier-field-ids"], serde_json::json!([1]));
-    assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
-    assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 1);
-    assert_eq!(metadata["last-column-id"], 1);
-}
-
 #[test]
 fn create_table_request_honors_supported_format_version_property() {
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -2050,142 +1990,6 @@ fn catalog_assigns_read_only_schema_spec_and_sort_order_ids() {
     assert_eq!(updated["default-sort-order-id"], 0);
 }
 
-#[test]
-fn create_table_assigns_fresh_schema_field_ids_and_rewrites_references() {
-    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
-    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
-        "name": "events",
-        "schema": {
-            "type": "struct",
-            "schema-id": 41,
-            "identifier-field-ids": [0],
-            "fields": [
-                {"id": 0, "name": "id", "required": true, "type": "long"},
-                {
-                    "id": 10,
-                    "name": "details",
-                    "required": false,
-                    "type": {
-                        "type": "struct",
-                        "fields": [{"id": 11, "name": "category", "required": false, "type": "string"}]
-                    }
-                },
-                {
-                    "id": 20,
-                    "name": "tags",
-                    "required": false,
-                    "type": {
-                        "type": "list",
-                        "element-id": 21,
-                        "element-required": false,
-                        "element": "string"
-                    }
-                },
-                {
-                    "id": 30,
-                    "name": "attributes",
-                    "required": false,
-                    "type": {
-                        "type": "map",
-                        "key-id": 31,
-                        "key": "string",
-                        "value-id": 32,
-                        "value-required": false,
-                        "value": {
-                            "type": "struct",
-                            "fields": [{"id": 33, "name": "score", "required": false, "type": "int"}]
-                        }
-                    }
-                }
-            ]
-        },
-        "partition-spec": {
-            "spec-id": 42,
-            "fields": [{"source-id": 0, "name": "id", "transform": "identity"}]
-        },
-        "write-order": {
-            "order-id": 43,
-            "fields": [{
-                "source-id": 11,
-                "transform": "identity",
-                "direction": "asc",
-                "null-order": "nulls-first"
-            }]
-        }
-    }))
-    .expect("create table request should parse");
-
-    let (_, metadata) =
-        table_entry_from_create_table_request("warehouse", &namespace, request).expect("catalog should assign fresh field IDs");
-
-    let schema = &metadata["schemas"][0];
-    assert_eq!(schema["fields"][0]["id"], 1);
-    assert_eq!(schema["fields"][1]["id"], 2);
-    assert_eq!(schema["fields"][2]["id"], 3);
-    assert_eq!(schema["fields"][3]["id"], 4);
-    assert_eq!(schema["fields"][1]["type"]["fields"][0]["id"], 5);
-    assert_eq!(schema["fields"][2]["type"]["element-id"], 6);
-    assert_eq!(schema["fields"][3]["type"]["key-id"], 7);
-    assert_eq!(schema["fields"][3]["type"]["value-id"], 8);
-    assert_eq!(schema["fields"][3]["type"]["value"]["fields"][0]["id"], 9);
-    assert_eq!(schema["identifier-field-ids"], serde_json::json!([1]));
-    assert_eq!(metadata["last-column-id"], 9);
-    assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
-    assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 5);
-}
-
-#[test]
-fn create_table_rejects_duplicate_temporary_schema_field_ids() {
-    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
-    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
-        "name": "events",
-        "schema": {
-            "type": "struct",
-            "fields": [
-                {"id": 0, "name": "id", "required": false, "type": "long"},
-                {"id": 0, "name": "payload", "required": false, "type": "string"}
-            ]
-        }
-    }))
-    .expect("create table request should parse");
-
-    let error = table_entry_from_create_table_request("warehouse", &namespace, request)
-        .expect_err("duplicate temporary field IDs must be rejected");
-
-    assert_eq!(error.message(), Some("duplicate create schema field id 0"));
-}
-
-#[test]
-fn create_table_rejects_excessive_schema_nesting() {
-    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
-    let mut field_type = serde_json::Value::from("long");
-    for element_id in 1..=crate::table_catalog::ICEBERG_MAX_SCHEMA_NESTING_DEPTH + 1 {
-        field_type = serde_json::json!({
-            "type": "list",
-            "element-id": element_id,
-            "element-required": false,
-            "element": field_type
-        });
-    }
-    let request = CreateTableRequest {
-        name: "events".to_string(),
-        location: None,
-        schema: serde_json::json!({
-            "type": "struct",
-            "fields": [{"id": 0, "name": "nested", "required": false, "type": field_type}]
-        }),
-        partition_spec: None,
-        write_order: None,
-        stage_create: false,
-        properties: BTreeMap::new(),
-    };
-
-    let error = table_entry_from_create_table_request("warehouse", &namespace, request)
-        .expect_err("excessively nested create schemas must be rejected");
-
-    assert_eq!(error.message(), Some("create schema exceeds the maximum nesting depth"));
-}
-
 #[test]
 fn standard_commit_binds_new_specs_and_sort_orders_to_current_schema() {
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -2443,11 +2247,7 @@ fn create_table_counts_collection_ids_in_last_column_id() {
     let (_, metadata) =
         table_entry_from_create_table_request("warehouse", &namespace, request).expect("table metadata should be created");
 
-    let schema = &metadata["schemas"][0];
-    assert_eq!(schema["fields"][0]["type"]["element-id"], 3);
-    assert_eq!(schema["fields"][1]["type"]["key-id"], 4);
-    assert_eq!(schema["fields"][1]["type"]["value-id"], 5);
-    assert_eq!(metadata["last-column-id"], 5);
+    assert_eq!(metadata["last-column-id"], 9);
 }
 
 #[test]